diff --git a/internal/terraform/context_plan2_test.go b/internal/terraform/context_plan2_test.go
index b5c65d8ee8e7..63fa95cdbe8e 100644
--- a/internal/terraform/context_plan2_test.go
+++ b/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)
+}
diff --git a/internal/terraform/node_resource_abstract.go b/internal/terraform/node_resource_abstract.go
index a15214b790ad..73e8c4859b7a 100644
--- a/internal/terraform/node_resource_abstract.go
+++ b/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)
 	}
 
diff --git a/internal/terraform/node_resource_plan_instance.go b/internal/terraform/node_resource_plan_instance.go
index bafbb88f49ae..8d5e449b0107 100644
--- a/internal/terraform/node_resource_plan_instance.go
+++ b/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
diff --git a/internal/terraform/upgrade_resource_state.go b/internal/terraform/upgrade_resource_state.go
index 2b748c5c6948..647db361cc7e 100644
--- a/internal/terraform/upgrade_resource_state.go
+++ b/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
 	}