core: Allow data resource count to be unknown during refresh

The count for a data resource can potentially depend on a managed resource
that isn't recorded in the state yet, in which case references to it will
always return unknown.

Ideally we'd do the data refreshes during the plan phase as discussed in
#17034, which would avoid this problem by planning the managed resources
in the same walk, but for now we'll just skip refreshing any data
resources with an unknown count during refresh and defer that work to the
apply phase, just as we'd do if there were unknown values in the main
configuration for the data resource.

terraform/context_refresh_test.go

@@ -976,6 +976,65 @@ func TestContext2Refresh_stateBasic(t *testing.T) {
 	}
 }
 
+func TestContext2Refresh_dataCount(t *testing.T) {
+	p := testProvider("test")
+	m := testModule(t, "refresh-data-count")
+
+	// This test is verifying that a data resource count can refer to a
+	// resource attribute that can't be known yet during refresh (because
+	// the resource in question isn't in the state at all). In that case,
+	// we skip the data resource during refresh and process it during the
+	// subsequent plan step instead.
+	//
+	// Normally it's an error for "count" to be computed, but during the
+	// refresh step we allow it because we _expect_ to be working with an
+	// incomplete picture of the world sometimes, particularly when we're
+	// creating object for the first time against an empty state.
+	//
+	// For more information, see:
+	//    https://github.com/hashicorp/terraform/issues/21047
+
+	p.GetSchemaReturn = &ProviderSchema{
+		ResourceTypes: map[string]*configschema.Block{
+			"test": {
+				Attributes: map[string]*configschema.Attribute{
+					"things": {Type: cty.List(cty.String), Optional: true},
+				},
+			},
+		},
+		DataSources: map[string]*configschema.Block{
+			"test": {},
+		},
+	}
+
+	ctx := testContext2(t, &ContextOpts{
+		ProviderResolver: providers.ResolverFixed(
+			map[string]providers.Factory{
+				"test": testProviderFuncFixed(p),
+			},
+		),
+		Config: m,
+	})
+
+	s, diags := ctx.Refresh()
+	if p.ReadResourceCalled {
+		// The managed resource doesn't exist in the state yet, so there's
+		// nothing to refresh.
+		t.Errorf("ReadResource was called, but should not have been")
+	}
+	if p.ReadDataSourceCalled {
+		// The data resource should've been skipped because its count cannot
+		// be determined yet.
+		t.Errorf("ReadDataSource was called, but should not have been")
+	}
+
+	if diags.HasErrors() {
+		t.Fatalf("refresh errors: %s", diags.Err())
+	}
+
+	checkStateString(t, s, `<no state>`)
+}
+
 func TestContext2Refresh_dataOrphan(t *testing.T) {
 	p := testProvider("null")
 	state := MustShimLegacyState(&State{

terraform/eval_count.go

@@ -23,24 +23,39 @@ import (
 // the "count" behavior should not be enabled for this resource at all.
 //
 // If error diagnostics are returned then the result is always the meaningless
-// placeholder value -1, except in one case: if the count expression evaluates
-// to an unknown number value then the result is zero, allowing this situation
-// to be treated by the caller as special if needed. For example, an early
-// graph walk may wish to just silently skip resources with unknown counts
-// to allow them to be dealt with in a later graph walk where more information
-// is available.
+// placeholder value -1.
 func evaluateResourceCountExpression(expr hcl.Expression, ctx EvalContext) (int, tfdiags.Diagnostics) {
-	if expr == nil {
-		return -1, nil
+	count, known, diags := evaluateResourceCountExpressionKnown(expr, ctx)
+	if !known {
+		// Currently this is a rather bad outcome from a UX standpoint, since we have
+		// no real mechanism to deal with this situation and all we can do is produce
+		// an error message.
+		// FIXME: In future, implement a built-in mechanism for deferring changes that
+		// can't yet be predicted, and use it to guide the user through several
+		// plan/apply steps until the desired configuration is eventually reached.
+		diags = diags.Append(&hcl.Diagnostic{
+			Severity: hcl.DiagError,
+			Summary:  "Invalid count argument",
+			Detail:   `The "count" value depends on resource attributes that cannot be determined until apply, so Terraform cannot predict how many instances will be created. To work around this, use the -target argument to first apply only the resources that the count depends on.`,
+			Subject:  expr.Range().Ptr(),
+		})
 	}
+	return count, diags
+}
 
-	var diags tfdiags.Diagnostics
-	var count int
+// evaluateResourceCountExpressionKnown is like evaluateResourceCountExpression
+// except that it handles an unknown result by returning count = 0 and
+// a known = false, rather than by reporting the unknown value as an error
+// diagnostic.
+func evaluateResourceCountExpressionKnown(expr hcl.Expression, ctx EvalContext) (count int, known bool, diags tfdiags.Diagnostics) {
+	if expr == nil {
+		return -1, true, nil
+	}
 
 	countVal, countDiags := ctx.EvaluateExpr(expr, cty.Number, nil)
 	diags = diags.Append(countDiags)
 	if diags.HasErrors() {
-		return -1, diags
+		return -1, true, diags
 	}
 
 	switch {
@@ -51,25 +66,9 @@ func evaluateResourceCountExpression(expr hcl.Expression, ctx EvalContext) (int,
 			Detail:   `The given "count" argument value is null. An integer is required.`,
 			Subject:  expr.Range().Ptr(),
 		})
-		return -1, diags
+		return -1, true, diags
 	case !countVal.IsKnown():
-		// Currently this is a rather bad outcome from a UX standpoint, since we have
-		// no real mechanism to deal with this situation and all we can do is produce
-		// an error message.
-		// FIXME: In future, implement a built-in mechanism for deferring changes that
-		// can't yet be predicted, and use it to guide the user through several
-		// plan/apply steps until the desired configuration is eventually reached.
-		diags = diags.Append(&hcl.Diagnostic{
-			Severity: hcl.DiagError,
-			Summary:  "Invalid count argument",
-			Detail:   `The "count" value depends on resource attributes that cannot be determined until apply, so Terraform cannot predict how many instances will be created. To work around this, use the -target argument to first apply only the resources that the count depends on.`,
-			Subject:  expr.Range().Ptr(),
-		})
-		// We return zero+errors in this one case to allow callers to handle
-		// an unknown count as special. This is rarely necessary, but is used
-		// by the validate walk in particular so that it can just skip
-		// validation in this case, assuming a later walk will take care of it.
-		return 0, diags
+		return 0, false, diags
 	}
 
 	err := gocty.FromCtyValue(countVal, &count)
@@ -80,7 +79,7 @@ func evaluateResourceCountExpression(expr hcl.Expression, ctx EvalContext) (int,
 			Detail:   fmt.Sprintf(`The given "count" argument value is unsuitable: %s.`, err),
 			Subject:  expr.Range().Ptr(),
 		})
-		return -1, diags
+		return -1, true, diags
 	}
 	if count < 0 {
 		diags = diags.Append(&hcl.Diagnostic{
@@ -89,10 +88,10 @@ func evaluateResourceCountExpression(expr hcl.Expression, ctx EvalContext) (int,
 			Detail:   `The given "count" argument value is unsuitable: negative numbers are not supported.`,
 			Subject:  expr.Range().Ptr(),
 		})
-		return -1, diags
+		return -1, true, diags
 	}
 
-	return count, diags
+	return count, true, diags
 }
 
 // fixResourceCountSetTransition is a helper function to fix up the state when a

terraform/node_data_refresh.go

@@ -27,11 +27,16 @@ var (
 func (n *NodeRefreshableDataResource) DynamicExpand(ctx EvalContext) (*Graph, error) {
 	var diags tfdiags.Diagnostics
 
-	count, countDiags := evaluateResourceCountExpression(n.Config.Count, ctx)
+	count, countKnown, countDiags := evaluateResourceCountExpressionKnown(n.Config.Count, ctx)
 	diags = diags.Append(countDiags)
 	if countDiags.HasErrors() {
 		return nil, diags.Err()
 	}
+	if !countKnown {
+		// If the count isn't known yet, we'll skip refreshing and try expansion
+		// again during the plan walk.
+		return nil, nil
+	}
 
 	// Next we need to potentially rename an instance address in the state
 	// if we're transitioning whether "count" is set at all.

terraform/test-fixtures/refresh-data-count/refresh-data-count.tf

@@ -0,0 +1,7 @@
+resource "test" "foo" {
+  things = ["foo"]
+}
+
+data "test" "foo" {
+  count = length(test.foo.things)
+}