data schema changes may prevent state decoding

Data sources do not have state migrations, so there may be no way to
decode the prior state when faced with incompatible type changes.

Because prior state is only informational to the plan, and its existence
should not effect the planning process, we can skip decoding when faced
with errors.

internal/terraform/context_plan2_test.go

@@ -2963,3 +2963,66 @@ output "a" {
 		}
 	}
 }
+
+func TestContext2Plan_dataSchemaChange(t *testing.T) {
+	// We can't decode the prior state when a data source upgrades the schema
+	// in an incompatible way. Since prior state for data sources is purely
+	// informational, decoding should be skipped altogether.
+	m := testModuleInline(t, map[string]string{
+		"main.tf": `
+data "test_object" "a" {
+  obj {
+    # args changes from a list to a map
+    args = {
+      val = "string"
+	}
+  }
+}
+`,
+	})
+
+	p := new(MockProvider)
+	p.GetProviderSchemaResponse = getProviderSchemaResponseFromProviderSchema(&ProviderSchema{
+		DataSources: map[string]*configschema.Block{
+			"test_object": {
+				Attributes: map[string]*configschema.Attribute{
+					"id": {
+						Type:     cty.String,
+						Computed: true,
+					},
+				},
+				BlockTypes: map[string]*configschema.NestedBlock{
+					"obj": {
+						Block: configschema.Block{
+							Attributes: map[string]*configschema.Attribute{
+								"args": {Type: cty.Map(cty.String), Optional: true},
+							},
+						},
+						Nesting: configschema.NestingSet,
+					},
+				},
+			},
+		},
+	})
+
+	p.ReadDataSourceFn = func(req providers.ReadDataSourceRequest) (resp providers.ReadDataSourceResponse) {
+		resp.State = req.Config
+		return resp
+	}
+
+	state := states.BuildState(func(s *states.SyncState) {
+		s.SetResourceInstanceCurrent(mustResourceInstanceAddr(`data.test_object.a`), &states.ResourceInstanceObjectSrc{
+			AttrsJSON: []byte(`{"id":"old","obj":[{"args":["string"]}]}`),
+			Status:    states.ObjectReady,
+		}, mustProviderConfig(`provider["registry.terraform.io/hashicorp/test"]`))
+	})
+
+	ctx := testContext2(t, &ContextOpts{
+		Providers: map[addrs.Provider]providers.Factory{
+			addrs.NewDefaultProvider("test"): testProviderFuncFixed(p),
+		},
+	})
+
+	_, diags := ctx.Plan(m, state, DefaultPlanOpts)
+	assertNoErrors(t, diags)
+}

internal/terraform/node_resource_abstract.go

@@ -389,15 +389,19 @@ func (n *NodeAbstractResource) readResourceInstanceState(ctx EvalContext, addr a
 	}
 	diags = diags.Append(upgradeDiags)
 	if diags.HasErrors() {
-		// Note that we don't have any channel to return warnings here. We'll
-		// accept that for now since warnings during a schema upgrade would
-		// be pretty weird anyway, since this operation is supposed to seem
-		// invisible to the user.
 		return nil, diags
 	}
 
 	obj, err := src.Decode(schema.ImpliedType())
 	if err != nil {
+		// In the case of a data source which contains incompatible state
+		// migrations, we can just ignore decoding errors and skip comparing
+		// the prior state.
+		if addr.Resource.Resource.Mode == addrs.DataResourceMode {
+			log.Printf("[DEBUG] readResourceInstanceState: data source schema change for %s prevents decoding: %s", addr, err)
+			return nil, diags
+		}
+
 		diags = diags.Append(err)
 	}
 

internal/terraform/node_resource_plan_instance.go

@@ -84,7 +84,8 @@ func (n *NodePlannableResourceInstance) dataResourceExecute(ctx EvalContext) (di
 	// However, note that we don't have any explicit mechanism for upgrading
 	// data resource results as we do for managed resources, and so the
 	// prevRunState might not conform to the current schema if the
-	// previous run was with a different provider version.
+	// previous run was with a different provider version. In that case the
+	// snapshot will be null if we could not decode it at all.
 	diags = diags.Append(n.writeResourceInstanceState(ctx, state, prevRunState))
 	if diags.HasErrors() {
 		return diags

internal/terraform/upgrade_resource_state.go

@@ -127,7 +127,7 @@ func stripRemovedStateAttributes(state []byte, ty cty.Type) []byte {
 	if err != nil {
 		// we just log any errors here, and let the normal decode process catch
 		// invalid JSON.
-		log.Printf("[ERROR] UpgradeResourceState: %s", err)
+		log.Printf("[ERROR] UpgradeResourceState: stripRemovedStateAttributes: %s", err)
 		return state
 	}