Moved for different types (#2370)

Signed-off-by: Christian Mesh <christianmesh1@gmail.com>
Signed-off-by: Ilia Gogotchuri <ilia.gogotchuri0@gmail.com>
Co-authored-by: Christian Mesh <christianmesh1@gmail.com>

CHANGELOG.md

@@ -16,6 +16,7 @@ ENHANCEMENTS:
 * OpenTofu will now recommend using `-exclude` instead of `-target`, when possible, in the error messages about unknown values in `count` and `for_each` arguments, thereby providing a more definitive workaround. ([#2154](https://github.com/opentofu/opentofu/pull/2154))
 * State encryption now supports using external programs as key providers. Additionally, the PBKDF2 key provider now supports chaining via the `chain` parameter. ([#2023](https://github.com/opentofu/opentofu/pull/2023))
 * The `element` function now accepts negative indices, which extends the existing "wrapping" model into the negative direction. In particular, choosing element `-1` selects the final element in the sequence. ([#2371](https://github.com/opentofu/opentofu/pull/2371))
+* `moved` now supports moving between different types ([#2370](https://github.com/opentofu/opentofu/pull/2370))
 
 BUG FIXES:
 

internal/builtin/providers/tf/provider.go

@@ -169,6 +169,10 @@ func (p *Provider) ImportResourceState(req providers.ImportResourceStateRequest)
 	panic("unimplemented - terraform_remote_state has no resources")
 }
 
+func (p *Provider) MoveResourceState(r providers.MoveResourceStateRequest) (resp providers.MoveResourceStateResponse) {
+	panic("unimplmented")
+}
+
 // ValidateResourceConfig is used to to validate the resource configuration values.
 func (p *Provider) ValidateResourceConfig(req providers.ValidateResourceConfigRequest) providers.ValidateResourceConfigResponse {
 	return validateDataStoreResourceConfig(req)

internal/legacy/tofu/provider_mock.go

@@ -349,6 +349,10 @@ func (p *MockProvider) ImportResourceState(r providers.ImportResourceStateReques
 	return p.ImportResourceStateResponse
 }
 
+func (p *MockProvider) MoveResourceState(_ providers.MoveResourceStateRequest) providers.MoveResourceStateResponse {
+	panic("not implemented")
+}
+
 func (p *MockProvider) ReadDataSource(r providers.ReadDataSourceRequest) providers.ReadDataSourceResponse {
 	p.Lock()
 	defer p.Unlock()

internal/plugin/grpc_provider.go

@@ -302,7 +302,7 @@ func (p *GRPCProvider) UpgradeResourceState(r providers.UpgradeResourceStateRequ
 
 	protoReq := &proto.UpgradeResourceState_Request{
 		TypeName: r.TypeName,
-		Version:  int64(r.Version),
+		Version:  r.Version,
 		RawState: &proto.RawState{
 			Json:    r.RawStateJSON,
 			Flatmap: r.RawStateFlatmap,
@@ -638,6 +638,53 @@ func (p *GRPCProvider) ImportResourceState(r providers.ImportResourceStateReques
 	return resp
 }
 
+func (p *GRPCProvider) MoveResourceState(r providers.MoveResourceStateRequest) providers.MoveResourceStateResponse {
+	var resp providers.MoveResourceStateResponse
+	logger.Trace("GRPCProvider: MoveResourceState")
+
+	schema := p.GetProviderSchema()
+	if schema.Diagnostics.HasErrors() {
+		resp.Diagnostics = schema.Diagnostics
+		return resp
+	}
+
+	resourceSchema, ok := schema.ResourceTypes[r.TargetTypeName]
+	if !ok {
+		schema.Diagnostics = schema.Diagnostics.Append(fmt.Errorf("unknown data source %q", r.TargetTypeName))
+		return resp
+	}
+
+	protoReq := &proto.MoveResourceState_Request{
+		SourceProviderAddress: r.SourceProviderAddress,
+		SourceTypeName:        r.SourceTypeName,
+		//nolint:gosec // this will be refactored eventually
+		SourceSchemaVersion: int64(r.SourceSchemaVersion),
+		SourceState: &proto.RawState{
+			Json:    r.SourceStateJSON,
+			Flatmap: r.SourceStateFlatmap,
+		},
+		SourcePrivate:  r.SourcePrivate,
+		TargetTypeName: r.TargetTypeName,
+	}
+
+	protoResp, err := p.client.MoveResourceState(p.ctx, protoReq)
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(grpcErr(err))
+		return resp
+	}
+	resp.Diagnostics = resp.Diagnostics.Append(convert.ProtoToDiagnostics(protoResp.Diagnostics))
+
+	state, err := decodeDynamicValue(protoResp.TargetState, resourceSchema.Block.ImpliedType())
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+	resp.TargetState = state
+	resp.TargetPrivate = protoResp.TargetPrivate
+
+	return resp
+}
+
 func (p *GRPCProvider) ReadDataSource(r providers.ReadDataSourceRequest) (resp providers.ReadDataSourceResponse) {
 	logger.Trace("GRPCProvider: ReadDataSource")
 

internal/plugin/grpc_provider_test.go

@@ -388,6 +388,42 @@ func TestGRPCProvider_UpgradeResourceStateJSON(t *testing.T) {
 	}
 }
 
+func TestGRPCProvider_MoveResourceState(t *testing.T) {
+	client := mockProviderClient(t)
+	p := &GRPCProvider{
+		client: client,
+	}
+
+	client.EXPECT().MoveResourceState(
+		gomock.Any(),
+		gomock.Any(),
+	).Return(&proto.MoveResourceState_Response{
+		TargetState: &proto.DynamicValue{
+			Msgpack: []byte("\x81\xa4attr\xa3bar"),
+		},
+		TargetPrivate: []byte(`{"meta": "data"}`),
+	}, nil)
+
+	resp := p.MoveResourceState(providers.MoveResourceStateRequest{
+		SourceTypeName:      "resource_old",
+		SourceSchemaVersion: 0,
+		TargetTypeName:      "resource",
+	})
+	checkDiags(t, resp.Diagnostics)
+
+	expectedState := cty.ObjectVal(map[string]cty.Value{
+		"attr": cty.StringVal("bar"),
+	})
+	expectedPrivate := []byte(`{"meta": "data"}`)
+
+	if !cmp.Equal(expectedState, resp.TargetState, typeComparer, valueComparer, equateEmpty) {
+		t.Fatal(cmp.Diff(expectedState, resp.TargetState, typeComparer, valueComparer, equateEmpty))
+	}
+	if !bytes.Equal(expectedPrivate, resp.TargetPrivate) {
+		t.Fatalf("expected %q, got %q", expectedPrivate, resp.TargetPrivate)
+	}
+}
+
 func TestGRPCProvider_Configure(t *testing.T) {
 	client := mockProviderClient(t)
 	p := &GRPCProvider{

internal/plugin6/grpc_provider.go

@@ -627,6 +627,53 @@ func (p *GRPCProvider) ImportResourceState(r providers.ImportResourceStateReques
 	return resp
 }
 
+func (p *GRPCProvider) MoveResourceState(r providers.MoveResourceStateRequest) providers.MoveResourceStateResponse {
+	logger.Trace("GRPCProvider.v6: MoveResourceState")
+	var resp providers.MoveResourceStateResponse
+
+	schema := p.GetProviderSchema()
+	if schema.Diagnostics.HasErrors() {
+		resp.Diagnostics = schema.Diagnostics
+		return resp
+	}
+
+	resourceSchema, ok := schema.ResourceTypes[r.TargetTypeName]
+	if !ok {
+		schema.Diagnostics = schema.Diagnostics.Append(fmt.Errorf("unknown data source %q", r.TargetTypeName))
+		return resp
+	}
+
+	protoReq := &proto6.MoveResourceState_Request{
+		SourceProviderAddress: r.SourceProviderAddress,
+		SourceTypeName:        r.SourceTypeName,
+		//nolint:gosec // this will be refactored eventually
+		SourceSchemaVersion: int64(r.SourceSchemaVersion),
+		SourceState: &proto6.RawState{
+			Json:    r.SourceStateJSON,
+			Flatmap: r.SourceStateFlatmap,
+		},
+		SourcePrivate:  r.SourcePrivate,
+		TargetTypeName: r.TargetTypeName,
+	}
+
+	protoResp, err := p.client.MoveResourceState(p.ctx, protoReq)
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(grpcErr(err))
+		return resp
+	}
+	resp.Diagnostics = resp.Diagnostics.Append(convert.ProtoToDiagnostics(protoResp.Diagnostics))
+
+	state, err := decodeDynamicValue(protoResp.TargetState, resourceSchema.Block.ImpliedType())
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+	resp.TargetState = state
+	resp.TargetPrivate = protoResp.TargetPrivate
+
+	return resp
+}
+
 func (p *GRPCProvider) ReadDataSource(r providers.ReadDataSourceRequest) (resp providers.ReadDataSourceResponse) {
 	logger.Trace("GRPCProvider.v6: ReadDataSource")
 

internal/plugin6/grpc_provider_test.go

@@ -395,6 +395,41 @@ func TestGRPCProvider_UpgradeResourceStateJSON(t *testing.T) {
 	}
 }
 
+func TestGRPCProvider_MoveResourceState(t *testing.T) {
+	client := mockProviderClient(t)
+	p := &GRPCProvider{
+		client: client,
+	}
+
+	client.EXPECT().MoveResourceState(
+		gomock.Any(),
+		gomock.Any(),
+	).Return(&proto.MoveResourceState_Response{
+		TargetState: &proto.DynamicValue{
+			Msgpack: []byte("\x81\xa4attr\xa3bar"),
+		},
+		TargetPrivate: []byte(`{"meta": "data"}`),
+	}, nil)
+
+	resp := p.MoveResourceState(providers.MoveResourceStateRequest{
+		SourceTypeName:      "resource_old",
+		SourceSchemaVersion: 0,
+		TargetTypeName:      "resource",
+	})
+	checkDiags(t, resp.Diagnostics)
+
+	expectedState := cty.ObjectVal(map[string]cty.Value{
+		"attr": cty.StringVal("bar"),
+	})
+	expectedPrivate := []byte(`{"meta": "data"}`)
+
+	if !cmp.Equal(expectedState, resp.TargetState, typeComparer, valueComparer, equateEmpty) {
+		t.Fatal(cmp.Diff(expectedState, resp.TargetState, typeComparer, valueComparer, equateEmpty))
+	}
+	if !bytes.Equal(expectedPrivate, resp.TargetPrivate) {
+		t.Fatalf("expected %q, got %q", expectedPrivate, resp.TargetPrivate)
+	}
+}
 func TestGRPCProvider_Configure(t *testing.T) {
 	client := mockProviderClient(t)
 	p := &GRPCProvider{

internal/provider-simple-v6/provider.go

@@ -71,8 +71,20 @@ func (s simple) ValidateDataResourceConfig(req providers.ValidateDataResourceCon
 	return resp
 }
 
-func (p simple) UpgradeResourceState(req providers.UpgradeResourceStateRequest) (resp providers.UpgradeResourceStateResponse) {
-	ty := p.schema.ResourceTypes[req.TypeName].Block.ImpliedType()
+func (s simple) MoveResourceState(req providers.MoveResourceStateRequest) providers.MoveResourceStateResponse {
+	var resp providers.MoveResourceStateResponse
+	val, err := ctyjson.Unmarshal(req.SourceStateJSON, s.schema.ResourceTypes["simple_resource"].Block.ImpliedType())
+	resp.Diagnostics = resp.Diagnostics.Append(err)
+	if err != nil {
+		return resp
+	}
+	resp.TargetState = val
+	resp.TargetPrivate = req.SourcePrivate
+	return resp
+}
+func (s simple) UpgradeResourceState(req providers.UpgradeResourceStateRequest) providers.UpgradeResourceStateResponse {
+	var resp providers.UpgradeResourceStateResponse
+	ty := s.schema.ResourceTypes[req.TypeName].Block.ImpliedType()
 	val, err := ctyjson.Unmarshal(req.RawStateJSON, ty)
 	resp.Diagnostics = resp.Diagnostics.Append(err)
 	resp.UpgradedState = val

internal/provider-simple/provider.go

@@ -70,8 +70,20 @@ func (s simple) ValidateDataResourceConfig(req providers.ValidateDataResourceCon
 	return resp
 }
 
-func (p simple) UpgradeResourceState(req providers.UpgradeResourceStateRequest) (resp providers.UpgradeResourceStateResponse) {
-	ty := p.schema.ResourceTypes[req.TypeName].Block.ImpliedType()
+func (s simple) MoveResourceState(req providers.MoveResourceStateRequest) providers.MoveResourceStateResponse {
+	var resp providers.MoveResourceStateResponse
+	val, err := ctyjson.Unmarshal(req.SourceStateJSON, s.schema.ResourceTypes["simple_resource"].Block.ImpliedType())
+	resp.Diagnostics = resp.Diagnostics.Append(err)
+	if err != nil {
+		return resp
+	}
+	resp.TargetState = val
+	resp.TargetPrivate = req.SourcePrivate
+	return resp
+}
+func (s simple) UpgradeResourceState(req providers.UpgradeResourceStateRequest) providers.UpgradeResourceStateResponse {
+	var resp providers.UpgradeResourceStateResponse
+	ty := s.schema.ResourceTypes[req.TypeName].Block.ImpliedType()
 	val, err := ctyjson.Unmarshal(req.RawStateJSON, ty)
 	resp.Diagnostics = resp.Diagnostics.Append(err)
 	resp.UpgradedState = val

internal/providers/provider.go

@@ -74,6 +74,8 @@ type Interface interface {
 	// ImportResourceState requests that the given resource be imported.
 	ImportResourceState(ImportResourceStateRequest) ImportResourceStateResponse
 
+	MoveResourceState(MoveResourceStateRequest) MoveResourceStateResponse
+
 	// ReadDataSource returns the data source's current state.
 	ReadDataSource(ReadDataSourceRequest) ReadDataSourceResponse
 
@@ -456,6 +458,37 @@ func (ir ImportedResource) AsInstanceObject() *states.ResourceInstanceObject {
 	}
 }
 
+type MoveResourceStateRequest struct {
+	// The address of the provider the resource is being moved from.
+	SourceProviderAddress string
+	// The resource type that the resource is being moved from.
+	SourceTypeName string
+	// The schema version of the resource type that the resource is being
+	// moved from.
+	SourceSchemaVersion uint64
+	// The raw state of the resource being moved. Only the json field is
+	// populated, as there should be no legacy providers using the flatmap
+	// format that support newly introduced RPCs.
+	SourceStateJSON    []byte
+	SourceStateFlatmap map[string]string // Unused
+
+	// The private state of the resource being moved.
+	SourcePrivate []byte
+
+	// The resource type that the resource is being moved to.
+	TargetTypeName string
+}
+
+type MoveResourceStateResponse struct {
+	// The state of the resource after it has been moved.
+	TargetState cty.Value
+	// The private state of the resource after it has been moved.
+	TargetPrivate []byte
+
+	// Diagnostics contains any warnings or errors from the method call.
+	Diagnostics tfdiags.Diagnostics
+}
+
 type ReadDataSourceRequest struct {
 	// TypeName is the name of the data source type to Read.
 	TypeName string

internal/refactoring/move_validate.go

@@ -160,18 +160,6 @@ func ValidateMoves(stmts []MoveStatement, rootCfg *configs.Config, declaredInsts
 					StmtRange: stmt.DeclRange,
 				})
 			}
-
-			// Resource types must match.
-			if resourceTypesDiffer(absFrom, absTo) {
-				diags = diags.Append(&hcl.Diagnostic{
-					Severity: hcl.DiagError,
-					Summary:  "Resource type mismatch",
-					Detail: fmt.Sprintf(
-						"This statement declares a move from %s to %s, which is a %s of a different type.", absFrom, absTo, noun,
-					),
-				})
-			}
-
 		}
 	}
 
@@ -256,19 +244,6 @@ func moveableObjectExists(addr addrs.AbsMoveable, in instances.Set) bool {
 	}
 }
 
-func resourceTypesDiffer(absFrom, absTo addrs.AbsMoveable) bool {
-	switch absFrom := absFrom.(type) {
-	case addrs.AbsMoveableResource:
-		// addrs.UnifyMoveEndpoints guarantees that both addresses are of the
-		// same kind, so at this point we can assume that absTo is also an
-		// addrs.AbsResourceInstance or addrs.AbsResource.
-		absTo := absTo.(addrs.AbsMoveableResource)
-		return absFrom.AffectedAbsResource().Resource.Type != absTo.AffectedAbsResource().Resource.Type
-	default:
-		return false
-	}
-}
-
 func movableObjectDeclRange(addr addrs.AbsMoveable, cfg *configs.Config) (tfdiags.SourceRange, bool) {
 	switch addr := addr.(type) {
 	case addrs.ModuleInstance:

internal/refactoring/move_validate_test.go

@@ -444,7 +444,7 @@ Each resource can have moved from only one source resource.`,
 					`other.single`,
 				),
 			},
-			WantError: `Resource type mismatch: This statement declares a move from test.nonexist1 to other.single, which is a resource of a different type.`,
+			WantError: ``,
 		},
 		"resource instance type mismatch": {
 			Statements: []MoveStatement{
@@ -453,7 +453,7 @@ Each resource can have moved from only one source resource.`,
 					`other.single`,
 				),
 			},
-			WantError: `Resource type mismatch: This statement declares a move from test.nonexist1[0] to other.single, which is a resource instance of a different type.`,
+			WantError: ``,
 		},
 		"crossing nested statements": {
 			// overlapping nested moves will result in a cycle.

internal/tofu/context_plan.go

@@ -482,6 +482,7 @@ func (c *Context) prePlanFindAndApplyMoves(config *configs.Config, prevRunState
 		moveStmts = append(moveStmts, implicitMoveStmts...)
 	}
 	moveResults := refactoring.ApplyMoves(moveStmts, prevRunState)
+
 	return moveStmts, moveResults
 }
 
@@ -938,8 +939,12 @@ func (c *Context) driftedResources(config *configs.Config, oldState, newState *s
 					prevRunAddr = move.From
 				}
 
-				newIS := newState.ResourceInstance(addr)
+				if isResourceMovedToDifferentType(addr, prevRunAddr) {
+					// We don't report drift in case of resource type change
+					continue
+				}
 
+				newIS := newState.ResourceInstance(addr)
 				schema, _ := schemas.ResourceTypeConfig(
 					provider,
 					addr.Resource.Resource.Mode,

internal/tofu/context_plan2_test.go

@@ -1391,6 +1391,403 @@ func TestContext2Plan_movedResourceBasic(t *testing.T) {
 	})
 }
 
+func constructProviderSchemaForTesting(attrs map[string]*configschema.Attribute) providers.Schema {
+	return providers.Schema{
+		Block: &configschema.Block{Attributes: attrs},
+	}
+}
+
+func TestContext2Plan_movedResourceToDifferentType(t *testing.T) {
+	type test struct {
+		name          string
+		mockProvider  *MockProvider
+		providerAddr1 *addrs.AbsProviderConfig
+		providerAddr2 *addrs.AbsProviderConfig
+		oldSchema     providers.Schema
+		newSchema     providers.Schema
+		oldType       string
+		newType       string
+		config        map[string]string
+		attrsJSON     []byte
+		wantOp        plans.Action
+		wantErrStr    string
+	}
+
+	tests := []test{
+		{
+			name: "basic case",
+			oldSchema: constructProviderSchemaForTesting(map[string]*configschema.Attribute{
+				"test_number": {
+					Type:     cty.Number,
+					Optional: true,
+				},
+			}),
+			newSchema: constructProviderSchemaForTesting(map[string]*configschema.Attribute{
+				"test_number": {
+					Type:     cty.Number,
+					Optional: true,
+				},
+			}),
+			oldType: "test_object0",
+			newType: "test_object",
+			config: map[string]string{
+				"main.tf": `
+					resource "test_object" "new" {
+
+					}
+					moved {
+						from = test_object0.old
+						to   = test_object.new
+					}
+				`,
+			},
+			attrsJSON: []byte(`{}`),
+			wantOp:    plans.NoOp,
+		},
+		{
+			name: "attributes unchanged",
+			oldSchema: constructProviderSchemaForTesting(map[string]*configschema.Attribute{
+				"test_number": {
+					Type:     cty.Number,
+					Required: true,
+				},
+			}),
+			newSchema: constructProviderSchemaForTesting(map[string]*configschema.Attribute{
+				"test_number": {
+					Type:     cty.Number,
+					Required: true,
+				},
+			}),
+			oldType: "test_object0",
+			newType: "test_object",
+			config: map[string]string{
+				"main.tf": `
+					resource "test_object" "new" {
+						test_number = 1
+					}
+					moved {
+						from = test_object0.old
+						to   = test_object.new
+					}
+				`,
+			},
+			attrsJSON: []byte(`{"test_number": 1}`),
+			wantOp:    plans.NoOp,
+		},
+		{
+			name: "attribute changed",
+			oldSchema: constructProviderSchemaForTesting(map[string]*configschema.Attribute{
+				"test_number": {
+					Type:     cty.Number,
+					Required: true,
+				},
+			}),
+			newSchema: constructProviderSchemaForTesting(map[string]*configschema.Attribute{
+				"test_number": {
+					Type:     cty.Number,
+					Required: true,
+				},
+			}),
+			oldType: "test_object0",
+			newType: "test_object",
+			config: map[string]string{
+				"main.tf": `
+					resource "test_object" "new" {
+						test_number = 1
+					}
+					moved {
+						from = test_object0.old
+						to   = test_object.new
+					}
+				`,
+			},
+			attrsJSON: []byte(`{"test_number": 2}`),
+			wantOp:    plans.Update,
+		},
+		{
+			name: "attribute removed provider doesn't remove the attribute",
+			oldSchema: constructProviderSchemaForTesting(map[string]*configschema.Attribute{
+				"test_number": {
+					Type:     cty.Number,
+					Required: true,
+				},
+				"test_string": {
+					Type:     cty.String,
+					Required: true,
+				},
+			}),
+			newSchema: constructProviderSchemaForTesting(map[string]*configschema.Attribute{
+				"test_number": {
+					Type:     cty.Number,
+					Required: true,
+				},
+			}),
+			oldType: "test_object0",
+			newType: "test_object",
+			config: map[string]string{
+				"main.tf": `
+					resource "test_object" "new" {
+						test_number = 1
+					}
+					moved {
+						from = test_object0.old
+						to   = test_object.new
+					}
+				`,
+			},
+			attrsJSON:  []byte(`{"test_number": 1, "test_string": "foo"}`),
+			wantErrStr: `unsupported attribute "test_string"`,
+		},
+		{
+			name:    "provider failure",
+			oldType: "test_object0",
+			newType: "test_object",
+			config: map[string]string{
+				"main.tf": `
+					resource "test_object" "new" {}
+					moved {
+						from = test_object0.old
+						to   = test_object.new
+					}
+				`,
+			},
+			attrsJSON: []byte(`{}`),
+			mockProvider: &MockProvider{
+				GetProviderSchemaResponse: &providers.GetProviderSchemaResponse{
+					ResourceTypes: map[string]providers.Schema{
+						"test_object": constructProviderSchemaForTesting(map[string]*configschema.Attribute{
+							"test_number": {
+								Type:     cty.Number,
+								Optional: true,
+							},
+						}),
+						"test_object0": constructProviderSchemaForTesting(map[string]*configschema.Attribute{
+							"test_number": {
+								Type:     cty.Number,
+								Optional: true,
+							},
+						}),
+					},
+				},
+				MoveResourceStateResponse: &providers.MoveResourceStateResponse{
+					Diagnostics: tfdiags.Diagnostics{tfdiags.Sourceless(tfdiags.Error, "move failed", "")},
+				},
+			},
+			wantErrStr: "move failed",
+		},
+		{
+			name: "different provider with same schema",
+			oldSchema: constructProviderSchemaForTesting(map[string]*configschema.Attribute{
+				"test_number": {
+					Type:     cty.Number,
+					Required: true,
+				},
+			}),
+			newSchema: constructProviderSchemaForTesting(map[string]*configschema.Attribute{
+				"test_number": {
+					Type:     cty.Number,
+					Required: true,
+				},
+			}),
+			oldType: "provider1_object",
+			newType: "provider2_object",
+			config: map[string]string{
+				"main.tf": `
+					resource "provider2_object" "new" {
+						test_number = 1
+					}
+					moved {
+						from = provider1_object.old
+						to   = provider2_object.new
+					}
+				`,
+			},
+			attrsJSON: []byte(`{"test_number": 1}`),
+			wantOp:    plans.NoOp,
+			providerAddr1: &addrs.AbsProviderConfig{
+				Provider: addrs.NewDefaultProvider("provider1"),
+				Module:   addrs.RootModule,
+			},
+			providerAddr2: &addrs.AbsProviderConfig{
+				Provider: addrs.NewDefaultProvider("provider2"),
+			},
+			mockProvider: &MockProvider{
+				GetProviderSchemaResponse: &providers.GetProviderSchemaResponse{
+					ResourceTypes: map[string]providers.Schema{
+						"provider2_object": constructProviderSchemaForTesting(map[string]*configschema.Attribute{
+							"test_number": {
+								Type:     cty.Number,
+								Required: true,
+							},
+						}),
+						"provider1_object": constructProviderSchemaForTesting(map[string]*configschema.Attribute{
+							"test_number": {
+								Type:     cty.Number,
+								Required: true,
+							},
+						}),
+					},
+				},
+				MoveResourceStateResponse: &providers.MoveResourceStateResponse{
+					TargetState: cty.ObjectVal(map[string]cty.Value{
+						"test_number": cty.NumberIntVal(1),
+					}),
+				},
+			},
+		},
+		{
+			name: "different provider with schema change",
+			oldSchema: constructProviderSchemaForTesting(map[string]*configschema.Attribute{
+				"test_number": {
+					Type:     cty.Number,
+					Required: true,
+				},
+			}),
+			newSchema: constructProviderSchemaForTesting(map[string]*configschema.Attribute{
+				"test_string": {
+					Type:     cty.String,
+					Required: true,
+				},
+			}),
+			oldType: "provider1_object",
+			newType: "provider2_object",
+			config: map[string]string{
+				"main.tf": `
+					resource "provider2_object" "new" {
+						test_string = "2"
+					}
+					moved {
+						from = provider1_object.old
+						to   = provider2_object.new
+					}
+				`,
+			},
+			attrsJSON: []byte(`{"test_number": 1}`),
+			wantOp:    plans.Update,
+			providerAddr1: &addrs.AbsProviderConfig{
+				Provider: addrs.NewDefaultProvider("provider1"),
+				Module:   addrs.RootModule,
+			},
+			providerAddr2: &addrs.AbsProviderConfig{
+				Provider: addrs.NewDefaultProvider("provider2"),
+			},
+			mockProvider: &MockProvider{
+				GetProviderSchemaResponse: &providers.GetProviderSchemaResponse{
+					ResourceTypes: map[string]providers.Schema{
+						"provider2_object": constructProviderSchemaForTesting(map[string]*configschema.Attribute{
+							"test_string": {
+								Type:     cty.String,
+								Required: true,
+							},
+						}),
+						"provider1_object": constructProviderSchemaForTesting(map[string]*configschema.Attribute{
+							"test_number": {
+								Type:     cty.Number,
+								Required: true,
+							},
+						}),
+					},
+				},
+				MoveResourceStateResponse: &providers.MoveResourceStateResponse{
+					TargetState: cty.ObjectVal(map[string]cty.Value{
+						"test_string": cty.StringVal("1"),
+					}),
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			oldAddr := mustResourceInstanceAddr(tt.oldType + ".old")
+			newAddr := mustResourceInstanceAddr(tt.newType + ".new")
+			m := testModuleInline(t, tt.config)
+
+			providerAddr := addrs.AbsProviderConfig{
+				Provider: addrs.NewDefaultProvider("test"),
+				Module:   addrs.RootModule,
+			}
+			if tt.providerAddr1 != nil {
+				providerAddr = *tt.providerAddr1
+			}
+
+			state := states.BuildState(
+				func(s *states.SyncState) {
+					s.SetResourceProvider(oldAddr.ContainingResource(), providerAddr)
+					s.SetResourceInstanceCurrent(oldAddr, &states.ResourceInstanceObjectSrc{
+						Status:    states.ObjectReady,
+						AttrsJSON: tt.attrsJSON,
+					}, providerAddr, addrs.NoKey)
+				})
+
+			provider := &MockProvider{
+				// New provider should know about both resource types
+				GetProviderSchemaResponse: &providers.GetProviderSchemaResponse{
+					ResourceTypes: map[string]providers.Schema{
+						tt.newType: tt.newSchema,
+						tt.oldType: tt.oldSchema,
+					},
+				},
+			}
+			if tt.mockProvider != nil {
+				provider = tt.mockProvider
+			}
+
+			providerFactory := map[addrs.Provider]providers.Factory{
+				providerAddr.Provider: testProviderFuncFixed(provider),
+			}
+			if tt.providerAddr2 != nil {
+				providerFactory[tt.providerAddr2.Provider] = testProviderFuncFixed(provider)
+			}
+
+			ctx := testContext2(t, &ContextOpts{
+				Providers: providerFactory,
+			})
+
+			plan, diags := ctx.Plan(context.Background(), m, state, &PlanOpts{
+				Mode: plans.NormalMode,
+			})
+
+			if tt.wantErrStr != "" {
+				if !diags.HasErrors() {
+					t.Fatalf("expected errors, got none")
+				}
+				if got, want := diags.Err().Error(), tt.wantErrStr; !strings.Contains(got, want) {
+					t.Fatalf("wrong error\ngot: %s\nwant substring: %s", got, want)
+				}
+				return
+			}
+
+			assertNoErrors(t, diags)
+
+			// Check that no plan was created for the old resource
+			instPlan := plan.Changes.ResourceInstance(oldAddr)
+			if instPlan != nil {
+				t.Fatalf("unexpected plan for %s; should've moved to %s", oldAddr, newAddr)
+			}
+
+			// Check that a plan was created for the new resource
+			instPlan = plan.Changes.ResourceInstance(newAddr)
+			if instPlan == nil {
+				t.Fatalf("no plan for %s at all", newAddr)
+			}
+
+			if got, want := instPlan.Addr, newAddr; !got.Equal(want) {
+				t.Errorf("wrong current address\ngot:  %s\nwant: %s", got, want)
+			}
+			if got, want := instPlan.PrevRunAddr, oldAddr; !got.Equal(want) {
+				t.Errorf("wrong previous run address\ngot:  %s\nwant: %s", got, want)
+			}
+			if got, want := instPlan.Action, tt.wantOp; got != want {
+				t.Errorf("wrong planned action\ngot:  %s\nwant: %s", got, want)
+			}
+			if got, want := instPlan.ActionReason, plans.ResourceInstanceChangeNoReason; got != want {
+				t.Errorf("wrong action reason\ngot:  %s\nwant: %s", got, want)
+			}
+		})
+	}
+}
+
 func TestContext2Plan_movedResourceCollision(t *testing.T) {
 	addrNoKey := mustResourceInstanceAddr("test_object.a")
 	addrZeroKey := mustResourceInstanceAddr("test_object.a[0]")

internal/tofu/node_resource_abstract.go

@@ -514,19 +514,22 @@ func (n *NodeAbstractResource) writeResourceState(ctx EvalContext, addr addrs.Ab
 	return diags
 }
 
+func isResourceMovedToDifferentType(newAddr, oldAddr addrs.AbsResourceInstance) bool {
+	return newAddr.Resource.Resource.Type != oldAddr.Resource.Resource.Type
+}
+
 // readResourceInstanceState reads the current object for a specific instance in
 // the state.
-func (n *NodeAbstractResourceInstance) readResourceInstanceState(ctx EvalContext, addr addrs.AbsResourceInstance) (*states.ResourceInstanceObject, tfdiags.Diagnostics) {
+func (n *NodeAbstractResourceInstance) readResourceInstanceState(evalCtx EvalContext, addr addrs.AbsResourceInstance) (*states.ResourceInstanceObject, tfdiags.Diagnostics) {
 	var diags tfdiags.Diagnostics
-	provider, providerSchema, err := getProvider(ctx, n.ResolvedProvider.ProviderConfig, n.ResolvedProviderKey)
+	provider, providerSchema, err := getProvider(evalCtx, n.ResolvedProvider.ProviderConfig, n.ResolvedProviderKey)
 	if err != nil {
-		diags = diags.Append(err)
-		return nil, diags
+		return nil, diags.Append(err)
 	}
 
 	log.Printf("[TRACE] readResourceInstanceState: reading state for %s", addr)
 
-	src := ctx.State().ResourceInstanceObject(addr, states.CurrentGen)
+	src := evalCtx.State().ResourceInstanceObject(addr, states.CurrentGen)
 	if src == nil {
 		// Presumably we only have deposed objects, then.
 		log.Printf("[TRACE] readResourceInstanceState: no state present for %s", addr)
@@ -538,11 +541,26 @@ func (n *NodeAbstractResourceInstance) readResourceInstanceState(ctx EvalContext
 		// Shouldn't happen since we should've failed long ago if no schema is present
 		return nil, diags.Append(fmt.Errorf("no schema available for %s while reading state; this is a bug in OpenTofu and should be reported", addr))
 	}
-	src, upgradeDiags := upgradeResourceState(addr, provider, src, schema, currentVersion)
-	if n.Config != nil {
-		upgradeDiags = upgradeDiags.InConfigBody(n.Config.Config, addr.String())
+
+	// prevAddr will match the newAddr if the resource wasn't moved (prevRunAddr checks move results)
+	prevAddr := n.prevRunAddr(evalCtx)
+	transformArgs := stateTransformArgs{
+		currentAddr:          addr,
+		prevAddr:             prevAddr,
+		provider:             provider,
+		objectSrc:            src,
+		currentSchema:        schema,
+		currentSchemaVersion: currentVersion,
+	}
+	if isResourceMovedToDifferentType(addr, prevAddr) {
+		src, diags = moveResourceState(transformArgs)
+	} else {
+		src, diags = upgradeResourceState(transformArgs)
+	}
+
+	if n.Config != nil {
+		diags = diags.InConfigBody(n.Config.Config, addr.String())
 	}
-	diags = diags.Append(upgradeDiags)
 	if diags.HasErrors() {
 		return nil, diags
 	}
@@ -557,9 +575,9 @@ func (n *NodeAbstractResourceInstance) readResourceInstanceState(ctx EvalContext
 
 // readResourceInstanceStateDeposed reads the deposed object for a specific
 // instance in the state.
-func (n *NodeAbstractResourceInstance) readResourceInstanceStateDeposed(ctx EvalContext, addr addrs.AbsResourceInstance, key states.DeposedKey) (*states.ResourceInstanceObject, tfdiags.Diagnostics) {
+func (n *NodeAbstractResourceInstance) readResourceInstanceStateDeposed(evalCtx EvalContext, addr addrs.AbsResourceInstance, key states.DeposedKey) (*states.ResourceInstanceObject, tfdiags.Diagnostics) {
 	var diags tfdiags.Diagnostics
-	provider, providerSchema, err := getProvider(ctx, n.ResolvedProvider.ProviderConfig, n.ResolvedProviderKey)
+	provider, providerSchema, err := getProvider(evalCtx, n.ResolvedProvider.ProviderConfig, n.ResolvedProviderKey)
 	if err != nil {
 		diags = diags.Append(err)
 		return nil, diags
@@ -571,7 +589,7 @@ func (n *NodeAbstractResourceInstance) readResourceInstanceStateDeposed(ctx Eval
 
 	log.Printf("[TRACE] readResourceInstanceStateDeposed: reading state for %s deposed object %s", addr, key)
 
-	src := ctx.State().ResourceInstanceObject(addr, key)
+	src := evalCtx.State().ResourceInstanceObject(addr, key)
 	if src == nil {
 		// Presumably we only have deposed objects, then.
 		log.Printf("[TRACE] readResourceInstanceStateDeposed: no state present for %s deposed object %s", addr, key)
@@ -582,14 +600,26 @@ func (n *NodeAbstractResourceInstance) readResourceInstanceStateDeposed(ctx Eval
 	if schema == nil {
 		// Shouldn't happen since we should've failed long ago if no schema is present
 		return nil, diags.Append(fmt.Errorf("no schema available for %s while reading state; this is a bug in OpenTofu and should be reported", addr))
-
+	}
+	// prevAddr will match the newAddr if the resource wasn't moved (prevRunAddr checks move results)
+	prevAddr := n.prevRunAddr(evalCtx)
+	transformArgs := stateTransformArgs{
+		currentAddr:          addr,
+		prevAddr:             prevAddr,
+		provider:             provider,
+		objectSrc:            src,
+		currentSchema:        schema,
+		currentSchemaVersion: currentVersion,
+	}
+	if isResourceMovedToDifferentType(addr, prevAddr) {
+		src, diags = moveResourceState(transformArgs)
+	} else {
+		src, diags = upgradeResourceState(transformArgs)
 	}
 
-	src, upgradeDiags := upgradeResourceState(addr, provider, src, schema, currentVersion)
 	if n.Config != nil {
-		upgradeDiags = upgradeDiags.InConfigBody(n.Config.Config, addr.String())
+		diags = diags.InConfigBody(n.Config.Config, addr.String())
 	}
-	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

internal/tofu/node_resource_abstract_test.go

@@ -7,6 +7,8 @@ package tofu
 
 import (
 	"fmt"
+	"github.com/opentofu/opentofu/internal/refactoring"
+	"github.com/opentofu/opentofu/internal/tfdiags"
 	"testing"
 
 	"github.com/opentofu/opentofu/internal/addrs"
@@ -178,26 +180,139 @@ func TestNodeAbstractResourceSetProvider(t *testing.T) {
 	}
 }
 
-func TestNodeAbstractResource_ReadResourceInstanceState(t *testing.T) {
-	mockProvider := mockProviderWithResourceTypeSchema("aws_instance", &configschema.Block{
-		Attributes: map[string]*configschema.Attribute{
+type readResourceInstanceStateTest struct {
+	Name               string
+	State              *states.State
+	Node               *NodeAbstractResourceInstance
+	MoveResults        refactoring.MoveResults
+	Provider           *MockProvider
+	ExpectedInstanceId string
+	WantErrorStr       string
+}
+
+func getMockProviderForReadResourceInstanceState() *MockProvider {
+	return &MockProvider{
+		GetProviderSchemaResponse: &providers.GetProviderSchemaResponse{
+			ResourceTypes: map[string]providers.Schema{
+				"aws_instance": constructProviderSchemaForTesting(map[string]*configschema.Attribute{
 					"id": {
 						Type: cty.String,
-				Optional: true,
+					},
+				}),
+				"aws_instance0": constructProviderSchemaForTesting(map[string]*configschema.Attribute{
+					"id": {
+						Type: cty.String,
+					},
+				}),
 			},
 		},
-	})
+	}
+}
+
+func getReadResourceInstanceStateTests(stateBuilder func(s *states.SyncState)) []readResourceInstanceStateTest {
+	mockProvider := getMockProviderForReadResourceInstanceState()
+
+	mockProviderWithStateChange := getMockProviderForReadResourceInstanceState()
+	// Changes id to i-abc1234
+	mockProviderWithStateChange.MoveResourceStateResponse = &providers.MoveResourceStateResponse{
+		TargetState: cty.ObjectVal(map[string]cty.Value{"id": cty.StringVal("i-abc1234")}),
+	}
+
+	mockProviderWithMoveUnsupported := getMockProviderForReadResourceInstanceState()
+	mockProviderWithMoveUnsupported.MoveResourceStateFn = func(req providers.MoveResourceStateRequest) providers.MoveResourceStateResponse {
+		return providers.MoveResourceStateResponse{
+			Diagnostics: tfdiags.Diagnostics{tfdiags.Sourceless(tfdiags.Error, "move not supported", "")},
+		}
+	}
+
 	// This test does not configure the provider, but the mock provider will
 	// check that this was called and report errors.
 	mockProvider.ConfigureProviderCalled = true
+	mockProviderWithStateChange.ConfigureProviderCalled = true
+	mockProviderWithMoveUnsupported.ConfigureProviderCalled = true
 
-	tests := map[string]struct {
-		State              *states.State
-		Node               *NodeAbstractResourceInstance
-		ExpectedInstanceId string
-	}{
-		"ReadState gets primary instance state": {
-			State: states.BuildState(func(s *states.SyncState) {
+	tests := []readResourceInstanceStateTest{
+		{
+			Name:     "gets primary instance state",
+			Provider: mockProvider,
+			State:    states.BuildState(stateBuilder),
+			Node: &NodeAbstractResourceInstance{
+				NodeAbstractResource: NodeAbstractResource{
+					ResolvedProvider: ResolvedProvider{ProviderConfig: mustProviderConfig(`provider["registry.opentofu.org/hashicorp/aws"]`)},
+				},
+				// Otherwise prevRunAddr fails, since we have no current Addr in the state
+				Addr: mustResourceInstanceAddr("aws_instance.bar"),
+			},
+			ExpectedInstanceId: "i-abc123",
+		},
+		{
+			Name:     "resource moved to another type without state change",
+			Provider: mockProvider,
+			MoveResults: refactoring.MoveResults{
+				Changes: addrs.MakeMap[addrs.AbsResourceInstance, refactoring.MoveSuccess](
+					addrs.MakeMapElem(mustResourceInstanceAddr("aws_instance.bar"), refactoring.MoveSuccess{
+						From: mustResourceInstanceAddr("aws_instance0.baz"),
+						To:   mustResourceInstanceAddr("aws_instance.bar"),
+					})),
+			},
+			State: states.BuildState(stateBuilder),
+			Node: &NodeAbstractResourceInstance{
+				NodeAbstractResource: NodeAbstractResource{
+					ResolvedProvider: ResolvedProvider{ProviderConfig: mustProviderConfig(`provider["registry.opentofu.org/hashicorp/aws"]`)},
+				},
+				// Otherwise prevRunAddr fails, since we have no current Addr in the state
+				Addr: mustResourceInstanceAddr("aws_instance.bar"),
+			},
+			ExpectedInstanceId: "i-abc123",
+		},
+		{
+			Name:     "resource moved to another type with state change",
+			Provider: mockProviderWithStateChange,
+			MoveResults: refactoring.MoveResults{
+				Changes: addrs.MakeMap[addrs.AbsResourceInstance, refactoring.MoveSuccess](
+					addrs.MakeMapElem(mustResourceInstanceAddr("aws_instance.bar"), refactoring.MoveSuccess{
+						From: mustResourceInstanceAddr("aws_instance0.baz"),
+						To:   mustResourceInstanceAddr("aws_instance.bar"),
+					})),
+			},
+			State: states.BuildState(stateBuilder),
+			Node: &NodeAbstractResourceInstance{
+				NodeAbstractResource: NodeAbstractResource{
+					ResolvedProvider: ResolvedProvider{ProviderConfig: mustProviderConfig(`provider["registry.opentofu.org/hashicorp/aws"]`)},
+				},
+				// Otherwise prevRunAddr fails, since we have no current Addr in the state
+				Addr: mustResourceInstanceAddr("aws_instance.bar"),
+			},
+			// The state change should have been applied
+			ExpectedInstanceId: "i-abc1234",
+		},
+		{
+			Name:     "resource moved to another type but move not supported by provider",
+			Provider: mockProviderWithMoveUnsupported,
+			MoveResults: refactoring.MoveResults{
+				Changes: addrs.MakeMap[addrs.AbsResourceInstance, refactoring.MoveSuccess](
+					addrs.MakeMapElem(mustResourceInstanceAddr("aws_instance.bar"), refactoring.MoveSuccess{
+						From: mustResourceInstanceAddr("aws_instance0.baz"),
+					}),
+				),
+			},
+			State: states.BuildState(stateBuilder),
+			Node: &NodeAbstractResourceInstance{
+				NodeAbstractResource: NodeAbstractResource{
+					ResolvedProvider: ResolvedProvider{ProviderConfig: mustProviderConfig(`provider["registry.opentofu.org/hashicorp/aws"]`)},
+				},
+				Addr: mustResourceInstanceAddr("aws_instance.bar"),
+			},
+			WantErrorStr: "move not supported",
+		},
+	}
+	return tests
+}
+
+// TestNodeAbstractResource_ReadResourceInstanceState tests the readResourceInstanceState and readResourceInstanceStateDeposed methods of NodeAbstractResource.
+// Those are quite similar in behavior, so we can test them together.
+func TestNodeAbstractResource_ReadResourceInstanceState(t *testing.T) {
+	tests := getReadResourceInstanceStateTests(func(s *states.SyncState) {
 		providerAddr := addrs.AbsProviderConfig{
 			Provider: addrs.NewDefaultProvider("aws"),
 			Module:   addrs.RootModule,
@@ -212,58 +327,39 @@ func TestNodeAbstractResource_ReadResourceInstanceState(t *testing.T) {
 			Status:    states.ObjectReady,
 			AttrsJSON: []byte(`{"id":"i-abc123"}`),
 		}, providerAddr, addrs.NoKey)
-			}),
-			Node: &NodeAbstractResourceInstance{NodeAbstractResource: NodeAbstractResource{
-				Addr:             mustConfigResourceAddr("aws_instance.bar"),
-				ResolvedProvider: ResolvedProvider{ProviderConfig: mustProviderConfig(`provider["registry.opentofu.org/hashicorp/aws"]`)},
-			}},
-			ExpectedInstanceId: "i-abc123",
-		},
-	}
-
-	for k, test := range tests {
-		t.Run(k, func(t *testing.T) {
+	})
+	for _, test := range tests {
+		t.Run("ReadState "+test.Name, func(t *testing.T) {
 			ctx := new(MockEvalContext)
 			ctx.StateState = test.State.SyncWrapper()
 			ctx.PathPath = addrs.RootModuleInstance
-			ctx.ProviderSchemaSchema = mockProvider.GetProviderSchema()
+			ctx.ProviderSchemaSchema = test.Provider.GetProviderSchema()
+			ctx.MoveResultsResults = test.MoveResults
+			ctx.ProviderProvider = providers.Interface(test.Provider)
 
-			ctx.ProviderProvider = providers.Interface(mockProvider)
-
-			got, readDiags := test.Node.readResourceInstanceState(ctx, test.Node.NodeAbstractResource.Addr.Resource.Instance(addrs.NoKey).Absolute(addrs.RootModuleInstance))
+			got, readDiags := test.Node.readResourceInstanceState(ctx, test.Node.Addr)
+			if test.WantErrorStr != "" {
+				if !readDiags.HasErrors() {
+					t.Fatalf("[%s] Expected error, got none", test.Name)
+				}
+				if readDiags.Err().Error() != test.WantErrorStr {
+					t.Fatalf("[%s] Expected error %q, got %q", test.Name, test.WantErrorStr, readDiags.Err().Error())
+				}
+				return
+			}
 			if readDiags.HasErrors() {
-				t.Fatalf("[%s] Got err: %#v", k, readDiags.Err())
+				t.Fatalf("[%s] Got err: %#v", test.Name, readDiags.Err())
 			}
 
 			expected := test.ExpectedInstanceId
 
 			if !(got != nil && got.Value.GetAttr("id") == cty.StringVal(expected)) {
-				t.Fatalf("[%s] Expected output with ID %#v, got: %#v", k, expected, got)
+				t.Fatalf("[%s] Expected output with ID %#v, got: %#v", test.Name, expected, got)
 			}
 		})
 	}
-}
-
-func TestNodeAbstractResource_ReadResourceInstanceStateDeposed(t *testing.T) {
-	mockProvider := mockProviderWithResourceTypeSchema("aws_instance", &configschema.Block{
-		Attributes: map[string]*configschema.Attribute{
-			"id": {
-				Type:     cty.String,
-				Optional: true,
-			},
-		},
-	})
-	// This test does not configure the provider, but the mock provider will
-	// check that this was called and report errors.
-	mockProvider.ConfigureProviderCalled = true
-
-	tests := map[string]struct {
-		State              *states.State
-		Node               *NodeAbstractResourceInstance
-		ExpectedInstanceId string
-	}{
-		"ReadStateDeposed gets deposed instance": {
-			State: states.BuildState(func(s *states.SyncState) {
+	// Deposed tests
+	deposedTests := getReadResourceInstanceStateTests(func(s *states.SyncState) {
 		providerAddr := addrs.AbsProviderConfig{
 			Provider: addrs.NewDefaultProvider("aws"),
 			Module:   addrs.RootModule,
@@ -274,38 +370,41 @@ func TestNodeAbstractResource_ReadResourceInstanceStateDeposed(t *testing.T) {
 			Name: "bar",
 		}.Absolute(addrs.RootModuleInstance)
 		s.SetResourceProvider(oneAddr, providerAddr)
-				s.SetResourceInstanceDeposed(oneAddr.Instance(addrs.NoKey), states.DeposedKey("00000001"), &states.ResourceInstanceObjectSrc{
+		s.SetResourceInstanceDeposed(oneAddr.Instance(addrs.NoKey), "00000001", &states.ResourceInstanceObjectSrc{
 			Status:    states.ObjectReady,
 			AttrsJSON: []byte(`{"id":"i-abc123"}`),
 		}, providerAddr, addrs.NoKey)
-			}),
-			Node: &NodeAbstractResourceInstance{NodeAbstractResource: NodeAbstractResource{
-				Addr:             mustConfigResourceAddr("aws_instance.bar"),
-				ResolvedProvider: ResolvedProvider{ProviderConfig: mustProviderConfig(`provider["registry.opentofu.org/hashicorp/aws"]`)},
-			}},
-			ExpectedInstanceId: "i-abc123",
-		},
-	}
-	for k, test := range tests {
-		t.Run(k, func(t *testing.T) {
+	})
+	for _, test := range deposedTests {
+		t.Run("ReadStateDeposed "+test.Name, func(t *testing.T) {
 			ctx := new(MockEvalContext)
 			ctx.StateState = test.State.SyncWrapper()
 			ctx.PathPath = addrs.RootModuleInstance
-			ctx.ProviderSchemaSchema = mockProvider.GetProviderSchema()
-			ctx.ProviderProvider = providers.Interface(mockProvider)
+			ctx.ProviderSchemaSchema = test.Provider.GetProviderSchema()
+			ctx.MoveResultsResults = test.MoveResults
+			ctx.ProviderProvider = providers.Interface(test.Provider)
 
 			key := states.DeposedKey("00000001") // shim from legacy state assigns 0th deposed index this key
-
-			got, readDiags := test.Node.readResourceInstanceStateDeposed(ctx, test.Node.NodeAbstractResource.Addr.Resource.Instance(addrs.NoKey).Absolute(addrs.RootModuleInstance), key)
+			got, readDiags := test.Node.readResourceInstanceStateDeposed(ctx, test.Node.Addr, key)
+			if test.WantErrorStr != "" {
+				if !readDiags.HasErrors() {
+					t.Fatalf("[%s] Expected error, got none", test.Name)
+				}
+				if readDiags.Err().Error() != test.WantErrorStr {
+					t.Fatalf("[%s] Expected error %q, got %q", test.Name, test.WantErrorStr, readDiags.Err().Error())
+				}
+				return
+			}
 			if readDiags.HasErrors() {
-				t.Fatalf("[%s] Got err: %#v", k, readDiags.Err())
+				t.Fatalf("[%s] Got err: %#v", test.Name, readDiags.Err())
 			}
 
 			expected := test.ExpectedInstanceId
 
 			if !(got != nil && got.Value.GetAttr("id") == cty.StringVal(expected)) {
-				t.Fatalf("[%s] Expected output with ID %#v, got: %#v", k, expected, got)
+				t.Fatalf("[%s] Expected output with ID %#v, got: %#v", test.Name, expected, got)
 			}
 		})
+
 	}
 }

internal/tofu/provider_for_test_framework.go

@@ -136,6 +136,10 @@ func (p providerForTest) ImportResourceState(providers.ImportResourceStateReques
 	panic("Importing is not supported in testing context. providerForTest must not be used to call ImportResourceState")
 }
 
+func (p providerForTest) MoveResourceState(providers.MoveResourceStateRequest) providers.MoveResourceStateResponse {
+	panic("Moving is not supported in testing context. providerForTest must not be used to call MoveResourceState")
+}
+
 // Calling the internal provider ensures providerForTest has the same behaviour as if
 // it wasn't overridden or mocked. The only exception is ImportResourceState, which panics
 // if called via providerForTest because importing is not supported in testing framework.

internal/tofu/provider_mock.go

@@ -53,6 +53,12 @@ type MockProvider struct {
 	UpgradeResourceStateRequest  providers.UpgradeResourceStateRequest
 	UpgradeResourceStateFn       func(providers.UpgradeResourceStateRequest) providers.UpgradeResourceStateResponse
 
+	MoveResourceStateCalled   bool
+	MoveResourceStateTypeName string
+	MoveResourceStateResponse *providers.MoveResourceStateResponse
+	MoveResourceStateRequest  providers.MoveResourceStateRequest
+	MoveResourceStateFn       func(providers.MoveResourceStateRequest) providers.MoveResourceStateResponse
+
 	ConfigureProviderCalled   bool
 	ConfigureProviderResponse *providers.ConfigureProviderResponse
 	ConfigureProviderRequest  providers.ConfigureProviderRequest
@@ -251,6 +257,59 @@ func (p *MockProvider) UpgradeResourceState(r providers.UpgradeResourceStateRequ
 	return resp
 }
 
+func (p *MockProvider) MoveResourceState(r providers.MoveResourceStateRequest) providers.MoveResourceStateResponse {
+	var resp providers.MoveResourceStateResponse
+	p.Lock()
+	defer p.Unlock()
+
+	if !p.ConfigureProviderCalled {
+		resp.Diagnostics = resp.Diagnostics.Append(fmt.Errorf("configure not called before MoveResourceState %s -> %s", r.SourceTypeName, r.TargetTypeName))
+		return resp
+	}
+
+	schema, ok := p.getProviderSchema().ResourceTypes[r.SourceTypeName]
+	if !ok {
+		resp.Diagnostics = resp.Diagnostics.Append(fmt.Errorf("no schema found for %q", r.SourceTypeName))
+		return resp
+	}
+
+	schemaType := schema.Block.ImpliedType()
+
+	p.MoveResourceStateCalled = true
+	p.MoveResourceStateRequest = r
+
+	if p.MoveResourceStateFn != nil {
+		return p.MoveResourceStateFn(r)
+	}
+
+	if p.MoveResourceStateResponse != nil {
+		return *p.MoveResourceStateResponse
+	}
+
+	switch {
+	case r.SourceStateFlatmap != nil:
+		v, err := hcl2shim.HCL2ValueFromFlatmap(r.SourceStateFlatmap, schemaType)
+		if err != nil {
+			resp.Diagnostics = resp.Diagnostics.Append(err)
+			return resp
+		}
+		resp.TargetState = v
+	case len(r.SourceStateJSON) > 0:
+		v, err := ctyjson.Unmarshal(r.SourceStateJSON, schemaType)
+
+		if err != nil {
+			resp.Diagnostics = resp.Diagnostics.Append(err)
+			return resp
+		}
+		resp.TargetState = v
+	default:
+		resp.TargetState = cty.NullVal(schemaType)
+	}
+
+	resp.TargetPrivate = r.SourcePrivate
+	return resp
+}
+
 func (p *MockProvider) ConfigureProvider(r providers.ConfigureProviderRequest) (resp providers.ConfigureProviderResponse) {
 	p.Lock()
 	defer p.Unlock()
@@ -501,7 +560,6 @@ func (p *MockProvider) ImportResourceState(r providers.ImportResourceStateReques
 
 	return resp
 }
-
 func (p *MockProvider) ReadDataSource(r providers.ReadDataSourceRequest) (resp providers.ReadDataSourceResponse) {
 	p.Lock()
 	defer p.Unlock()

internal/tofu/upgrade_resource_state.go

@@ -18,82 +18,149 @@ import (
 	"github.com/zclconf/go-cty/cty"
 )
 
+// stateTransformArgs is a struct for convenience that holds the parameters required for state transformations
+type stateTransformArgs struct {
+	// currentAddr is the current/latest address of the resource
+	currentAddr addrs.AbsResourceInstance
+	// prevAddr is the previous address of the resource
+	prevAddr  addrs.AbsResourceInstance
+	provider  providers.Interface
+	objectSrc *states.ResourceInstanceObjectSrc
+	// currentSchema is the latest schema of the resource
+	currentSchema *configschema.Block
+	// currentSchemaVersion is the latest schema version of the resource
+	currentSchemaVersion uint64
+}
+
+// providerStateTransform transforms the state given the current and the previous AbsResourceInstance, Provider and ResourceInstanceObjectSrc
+// and returns a new state as cty.Value, new private state as []byte and diagnostics
+type providerStateTransform func(args stateTransformArgs) (cty.Value, []byte, tfdiags.Diagnostics)
+
 // upgradeResourceState will, if necessary, run the provider-defined upgrade
 // logic against the given state object to make it compliant with the
-// current schema version. This is a no-op if the given state object is
-// already at the latest version.
+// current schema version.
 //
 // If any errors occur during upgrade, error diagnostics are returned. In that
 // case it is not safe to proceed with using the original state object.
-func upgradeResourceState(addr addrs.AbsResourceInstance, provider providers.Interface, src *states.ResourceInstanceObjectSrc, currentSchema *configschema.Block, currentVersion uint64) (*states.ResourceInstanceObjectSrc, tfdiags.Diagnostics) {
-	if addr.Resource.Resource.Mode != addrs.ManagedResourceMode {
-		// We only do state upgrading for managed resources.
-		// This was a part of the normal workflow in older versions and
-		// returned early, so we are only going to log the error for now.
-		log.Printf("[ERROR] data resource %s should not require state upgrade", addr)
-		return src, nil
-	}
+func upgradeResourceState(args stateTransformArgs) (*states.ResourceInstanceObjectSrc, tfdiags.Diagnostics) {
+	return transformResourceState(args, upgradeResourceStateTransform)
+}
+
+// upgradeResourceStateTransform is a providerStateTransform that upgrades the state via provider upgrade logic
+func upgradeResourceStateTransform(args stateTransformArgs) (cty.Value, []byte, tfdiags.Diagnostics) {
+	log.Printf("[TRACE] upgradeResourceStateTransform: address: %s", args.currentAddr)
 
 	// Remove any attributes from state that are not present in the schema.
 	// This was previously taken care of by the provider, but data sources do
 	// not go through the UpgradeResourceState process.
 	//
+	// Required for upgrade not for move,
+	// since the deprecated fields might still be relevant for the migration.
+	//
 	// Legacy flatmap state is already taken care of during conversion.
 	// If the schema version is be changed, then allow the provider to handle
 	// removed attributes.
-	if len(src.AttrsJSON) > 0 && src.SchemaVersion == currentVersion {
-		src.AttrsJSON = stripRemovedStateAttributes(src.AttrsJSON, currentSchema.ImpliedType())
+	if len(args.objectSrc.AttrsJSON) > 0 && args.objectSrc.SchemaVersion == args.currentSchemaVersion {
+		args.objectSrc.AttrsJSON = stripRemovedStateAttributes(args.objectSrc.AttrsJSON, args.currentSchema.ImpliedType())
 	}
 
-	stateIsFlatmap := len(src.AttrsJSON) == 0
-
 	// TODO: This should eventually use a proper FQN.
-	providerType := addr.Resource.Resource.ImpliedProvider()
-	if src.SchemaVersion > currentVersion {
-		log.Printf("[TRACE] upgradeResourceState: can't downgrade state for %s from version %d to %d", addr, src.SchemaVersion, currentVersion)
+	providerType := args.currentAddr.Resource.Resource.ImpliedProvider()
+
+	if args.objectSrc.SchemaVersion > args.currentSchemaVersion {
+		log.Printf("[TRACE] upgradeResourceStateTransform: can't downgrade state for %s from version %d to %d", args.currentAddr, args.objectSrc.SchemaVersion, args.currentSchemaVersion)
 		var diags tfdiags.Diagnostics
-		diags = diags.Append(tfdiags.Sourceless(
+		return cty.NilVal, nil, diags.Append(tfdiags.Sourceless(
 			tfdiags.Error,
 			"Resource instance managed by newer provider version",
 			// This is not a very good error message, but we don't retain enough
 			// information in state to give good feedback on what provider
 			// version might be required here. :(
-			fmt.Sprintf("The current state of %s was created by a newer provider version than is currently selected. Upgrade the %s provider to work with this state.", addr, providerType),
+			fmt.Sprintf("The current state of %s was created by a newer provider version than is currently selected. Upgrade the %s provider to work with this state.", args.currentAddr, providerType),
 		))
-		return nil, diags
 	}
-
-	// If we get down here then we need to upgrade the state, with the
+	// If we get down here then we need to transform the state, with the
 	// provider's help.
 	// If this state was originally created by a version of OpenTofu prior to
 	// v0.12, this also includes translating from legacy flatmap to new-style
 	// representation, since only the provider has enough information to
 	// understand a flatmap built against an older schema.
-	if src.SchemaVersion != currentVersion {
-		log.Printf("[TRACE] upgradeResourceState: upgrading state for %s from version %d to %d using provider %q", addr, src.SchemaVersion, currentVersion, providerType)
+	if args.objectSrc.SchemaVersion != args.currentSchemaVersion {
+		log.Printf("[TRACE] transformResourceState: upgrading state for %s from version %d to %d using provider %q", args.currentAddr, args.objectSrc.SchemaVersion, args.currentSchemaVersion, providerType)
 	} else {
-		log.Printf("[TRACE] upgradeResourceState: schema version of %s is still %d; calling provider %q for any other minor fixups", addr, currentVersion, providerType)
+		log.Printf("[TRACE] transformResourceState: schema version of %s is still %d; calling provider %q for any other minor fixups", args.currentAddr, args.currentSchemaVersion, providerType)
 	}
 
 	req := providers.UpgradeResourceStateRequest{
-		TypeName: addr.Resource.Resource.Type,
+		TypeName: args.currentAddr.Resource.Resource.Type,
 
 		// TODO: The internal schema version representations are all using
 		// uint64 instead of int64, but unsigned integers aren't friendly
 		// to all protobuf target languages so in practice we use int64
 		// on the wire. In future we will change all of our internal
 		// representations to int64 too.
-		Version: int64(src.SchemaVersion),
+		//nolint:gosec // this will be refactored eventually
+		Version: int64(args.objectSrc.SchemaVersion),
 	}
 
+	stateIsFlatmap := len(args.objectSrc.AttrsJSON) == 0
 	if stateIsFlatmap {
-		req.RawStateFlatmap = src.AttrsFlat
+		req.RawStateFlatmap = args.objectSrc.AttrsFlat
 	} else {
-		req.RawStateJSON = src.AttrsJSON
+		req.RawStateJSON = args.objectSrc.AttrsJSON
 	}
 
-	resp := provider.UpgradeResourceState(req)
+	resp := args.provider.UpgradeResourceState(req)
 	diags := resp.Diagnostics
+	if diags.HasErrors() {
+		log.Printf("[TRACE] upgradeResourceStateTransform: failed - address: %s", args.currentAddr)
+		return cty.NilVal, nil, diags
+	}
+
+	return resp.UpgradedState, args.objectSrc.Private, diags
+}
+
+// moveResourceState moves the state from one type or provider to another
+// this function runs the provider-defined logic for MoveResourceState and performs the additional validation defined int transformResourceState
+// If any error occurs during the validation or state transformation, error diagnostics are returned.
+// Otherwise, the new state object is returned.
+func moveResourceState(params stateTransformArgs) (*states.ResourceInstanceObjectSrc, tfdiags.Diagnostics) {
+	return transformResourceState(params, moveResourceStateTransform)
+}
+
+// moveResourceStateTransform is a providerStateTransform that moves the state via provider move logic
+func moveResourceStateTransform(args stateTransformArgs) (cty.Value, []byte, tfdiags.Diagnostics) {
+	log.Printf("[TRACE] moveResourceStateTransform: new address: %s, previous address: %s", args.currentAddr, args.prevAddr)
+	req := providers.MoveResourceStateRequest{
+		SourceProviderAddress: args.prevAddr.Resource.Resource.ImpliedProvider(),
+		SourceTypeName:        args.prevAddr.Resource.Resource.Type,
+		SourceSchemaVersion:   args.objectSrc.SchemaVersion,
+		SourceStateJSON:       args.objectSrc.AttrsJSON,
+		SourceStateFlatmap:    args.objectSrc.AttrsFlat,
+		SourcePrivate:         args.objectSrc.Private,
+		TargetTypeName:        args.currentAddr.Resource.Resource.Type,
+	}
+	resp := args.provider.MoveResourceState(req)
+	diags := resp.Diagnostics
+	if diags.HasErrors() {
+		log.Printf("[TRACE] moveResourceStateTransform: failed - new address: %s, previous address: %s - diags: %s", args.currentAddr, args.prevAddr, diags.Err().Error())
+		return cty.NilVal, nil, diags
+	}
+	return resp.TargetState, resp.TargetPrivate, diags
+}
+
+// transformResourceState transforms the state based on passed in providerStateTransform and returns new ResourceInstanceObjectSrc
+// the function takes in the current and previous AbsResourceInstance, Provider, ResourceInstanceObjectSrc, current schema and current version
+func transformResourceState(args stateTransformArgs, stateTransform providerStateTransform) (*states.ResourceInstanceObjectSrc, tfdiags.Diagnostics) {
+	if args.currentAddr.Resource.Resource.Mode != addrs.ManagedResourceMode {
+		// We only do state transformation for managed resources.
+		// This was a part of the normal workflow in older versions and
+		// returned early, so we are only going to log the error for now.
+		log.Printf("[ERROR] data resource %s should not require state transformation", args.currentAddr)
+		return args.objectSrc, nil
+	}
+
+	newState, newPrivate, diags := stateTransform(args)
 	if diags.HasErrors() {
 		return nil, diags
 	}
@@ -102,29 +169,30 @@ func upgradeResourceState(addr addrs.AbsResourceInstance, provider providers.Int
 	// going over RPC this is actually already ensured by the
 	// marshaling/unmarshaling of the new value, but we'll check it here
 	// anyway for robustness, e.g. for in-process providers.
-	newValue := resp.UpgradedState
-	if errs := newValue.Type().TestConformance(currentSchema.ImpliedType()); len(errs) > 0 {
+	if errs := newState.Type().TestConformance(args.currentSchema.ImpliedType()); len(errs) > 0 {
+		providerType := args.currentAddr.Resource.Resource.ImpliedProvider()
 		for _, err := range errs {
 			diags = diags.Append(tfdiags.Sourceless(
 				tfdiags.Error,
-				"Invalid resource state upgrade",
-				fmt.Sprintf("The %s provider upgraded the state for %s from a previous version, but produced an invalid result: %s.", providerType, addr, tfdiags.FormatError(err)),
+				"Invalid resource state transformation",
+				fmt.Sprintf("The %s provider changed the state for %s, but produced an invalid result: %s.", providerType, args.currentAddr, tfdiags.FormatError(err)),
 			))
 		}
 		return nil, diags
 	}
-
-	new, err := src.CompleteUpgrade(newValue, currentSchema.ImpliedType(), uint64(currentVersion))
+	newSrc, err := args.objectSrc.CompleteUpgrade(newState, args.currentSchema.ImpliedType(), args.currentSchemaVersion)
 	if err != nil {
 		// We already checked for type conformance above, so getting into this
 		// codepath should be rare and is probably a bug somewhere under CompleteUpgrade.
 		diags = diags.Append(tfdiags.Sourceless(
 			tfdiags.Error,
-			"Failed to encode result of resource state upgrade",
-			fmt.Sprintf("Failed to encode state for %s after resource schema upgrade: %s.", addr, tfdiags.FormatError(err)),
+			"Failed to encode result of resource state transformation",
+			fmt.Sprintf("Failed to encode state for %s after resource schema upgrade: %s.", args.currentAddr, tfdiags.FormatError(err)),
 		))
 	}
-	return new, diags
+	// Assign new private state returned from state transformation
+	newSrc.Private = newPrivate
+	return newSrc, diags
 }
 
 // stripRemovedStateAttributes deletes any attributes no longer present in the

internal/tofu/upgrade_resource_state_test.go

@@ -6,9 +6,17 @@
 package tofu
 
 import (
+	"bytes"
 	"reflect"
+	"strings"
 	"testing"
 
+	"github.com/opentofu/opentofu/internal/addrs"
+	"github.com/opentofu/opentofu/internal/configs/configschema"
+	"github.com/opentofu/opentofu/internal/providers"
+	"github.com/opentofu/opentofu/internal/states"
+	"github.com/opentofu/opentofu/internal/tfdiags"
+
 	"github.com/zclconf/go-cty/cty"
 )
 
@@ -151,3 +159,370 @@ func TestStripRemovedStateAttributes(t *testing.T) {
 		})
 	}
 }
+
+// mustResourceInstanceAddr is a helper to create an absolute resource instance to test moveResourceStateTransform.
+// current address foo2_instance.cur and prev address foo_instance.prev
+func getMoveStateArgs() stateTransformArgs {
+	providerSchemaResponse := &providers.GetProviderSchemaResponse{
+		ResourceTypes: map[string]providers.Schema{
+			"foo_instance": constructProviderSchemaForTesting(map[string]*configschema.Attribute{
+				"foo": {
+					Type:     cty.String,
+					Required: true,
+				},
+			}),
+			"foo2_instance": constructProviderSchemaForTesting(map[string]*configschema.Attribute{
+				"foo": {
+					Type:     cty.String,
+					Required: true,
+				},
+			}),
+		},
+	}
+	return stateTransformArgs{
+		currentAddr: mustResourceInstanceAddr("foo2_instance.cur"),
+		prevAddr:    mustResourceInstanceAddr("foo_instance.prev"),
+		provider: &MockProvider{
+			ConfigureProviderCalled: true,
+			MoveResourceStateResponse: &providers.MoveResourceStateResponse{
+				TargetState: cty.ObjectVal(map[string]cty.Value{
+					"foo": cty.StringVal("bar"),
+				}),
+				TargetPrivate: []byte("private"),
+			},
+			GetProviderSchemaResponse: providerSchemaResponse,
+		},
+		objectSrc: &states.ResourceInstanceObjectSrc{
+			SchemaVersion: 2,
+			AttrsJSON:     []byte(`{"foo":"bar"}`),
+			AttrsFlat:     map[string]string{"foo": "bar"},
+			Private:       []byte("private"),
+		},
+		currentSchema:        &configschema.Block{},
+		currentSchemaVersion: 3,
+	}
+}
+
+func TestMoveResourceStateTransform(t *testing.T) {
+	type test struct {
+		name        string
+		args        stateTransformArgs
+		wantRequest *providers.MoveResourceStateRequest
+		wantState   cty.Value
+		wantPrivate []byte
+	}
+
+	tests := []test{
+		// Make sure that the moveResourceStateTransform function does not change state if the provider does not return a new state
+		{
+			name: "Move no changes",
+			args: getMoveStateArgs(),
+			wantState: cty.ObjectVal(map[string]cty.Value{
+				"foo": cty.StringVal("bar"),
+			}),
+			wantPrivate: []byte("private"),
+		},
+		// Check that the request is correctly populated
+		{
+			name: "Move check request",
+			args: getMoveStateArgs(),
+			wantRequest: &providers.MoveResourceStateRequest{
+				SourceProviderAddress: "foo",
+				SourceTypeName:        "foo_instance",
+				SourceSchemaVersion:   2,
+				SourceStateJSON:       []byte(`{"foo":"bar"}`),
+				SourceStateFlatmap:    map[string]string{"foo": "bar"},
+				SourcePrivate:         []byte("private"),
+				TargetTypeName:        "foo2_instance",
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			gotState, gotPrivate, diags := moveResourceStateTransform(tt.args)
+			if tt.wantRequest != nil {
+				mockProvider, ok := tt.args.provider.(*MockProvider)
+				if !ok {
+					t.Fatalf("unexpected provider type: %T", tt.args.provider)
+				}
+				if !reflect.DeepEqual(mockProvider.MoveResourceStateRequest, *tt.wantRequest) {
+					t.Fatalf("unexpected request: got %+v, want %+v", mockProvider.MoveResourceStateRequest, tt.wantRequest)
+				}
+				return
+			}
+			if diags.HasErrors() {
+				t.Fatalf("unexpected error: %s", diags.Err())
+			}
+			if !gotState.RawEquals(tt.wantState) {
+				t.Fatalf("unexpected state: got %#v, want %#v", gotState, tt.wantState)
+			}
+			if !bytes.Equal(gotPrivate, tt.wantPrivate) {
+				t.Fatalf("unexpected private: got %#v, want %#v", gotPrivate, tt.wantPrivate)
+			}
+		})
+	}
+}
+
+// mustResourceInstanceAddr is a helper to create an absolute resource instance to test upgradeResourceStateTransform.
+// current address and the previous address are foo_instance.cur
+// schema has field1 and field2, but the json state has field1, field2, and field3
+func getUpgradeStateArgs() stateTransformArgs {
+	sch := &configschema.Block{
+		Attributes: map[string]*configschema.Attribute{
+			"field1": {
+				Type: cty.String,
+			},
+			"field2": {
+				Type: cty.Bool,
+			},
+		},
+	}
+	args := stateTransformArgs{
+		currentAddr: mustResourceInstanceAddr("foo_instance.cur"),
+		prevAddr:    mustResourceInstanceAddr("foo_instance.cur"),
+		provider: &MockProvider{
+			ConfigureProviderCalled: true,
+			UpgradeResourceStateResponse: &providers.UpgradeResourceStateResponse{
+				UpgradedState: cty.ObjectVal(map[string]cty.Value{
+					"field1": cty.StringVal("bar"),
+					"field2": cty.True,
+				}),
+			},
+			GetProviderSchemaResponse: getProviderSchemaResponseFromProviderSchema(&ProviderSchema{
+				ResourceTypes: map[string]*configschema.Block{
+					"foo_instance": sch,
+				},
+			}),
+		},
+		objectSrc: &states.ResourceInstanceObjectSrc{
+			SchemaVersion: 2,
+			// Currently json state trimming to match the current schema is done, so field3 should be removed in passed request
+			AttrsJSON: []byte(`{"field1":"bar","field2":true,"field3":"baz"}`), // field3 is not in the schema
+			AttrsFlat: map[string]string{"foo": "bar"},
+			Private:   []byte("private"),
+		},
+		currentSchema:        sch,
+		currentSchemaVersion: 2,
+	}
+	return args
+}
+
+func TestUpgradeResourceStateTransform(t *testing.T) {
+	type test struct {
+		name        string
+		args        stateTransformArgs
+		wantRequest *providers.UpgradeResourceStateRequest
+		wantState   cty.Value
+		wantPrivate []byte
+		wantErr     string
+	}
+
+	argsForDowngrade := getUpgradeStateArgs()
+	argsForDowngrade.currentSchemaVersion = 1
+
+	tests := []test{
+		{
+			name: "Upgrade basic",
+			args: getUpgradeStateArgs(),
+			// Checking schema trimming and private state update
+			wantState: cty.ObjectVal(map[string]cty.Value{
+				"field1": cty.StringVal("bar"),
+				"field2": cty.True,
+			}),
+			wantPrivate: []byte("private"),
+		},
+		{
+			name: "Upgrade check request",
+			args: getUpgradeStateArgs(),
+			// Checking that field3 is removed from the state before sending the request
+			wantRequest: &providers.UpgradeResourceStateRequest{
+				TypeName:     "foo_instance",
+				Version:      2,
+				RawStateJSON: []byte(`{"field1":"bar","field2":true}`),
+			},
+		},
+		{
+			name:    "Upgrade to lower version should fail",
+			args:    argsForDowngrade,
+			wantErr: "Resource instance managed by newer provider version",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			gotState, gotPrivate, diags := upgradeResourceStateTransform(tt.args)
+			if tt.wantErr != "" {
+				if !diags.HasErrors() {
+					t.Fatalf("expected error: %s", tt.wantErr)
+				}
+				if !strings.Contains(diags.Err().Error(), tt.wantErr) {
+					t.Fatalf("unexpected error: got %s, want %s", diags.Err(), tt.wantErr)
+				}
+				return
+			}
+			if tt.wantRequest != nil {
+				mockProvider, ok := tt.args.provider.(*MockProvider)
+				if !ok {
+					t.Fatalf("unexpected provider type: %T", tt.args.provider)
+				}
+				if !reflect.DeepEqual(mockProvider.UpgradeResourceStateRequest, *tt.wantRequest) {
+					t.Fatalf("unexpected request: got %+v, want %+v", mockProvider.UpgradeResourceStateRequest, tt.wantRequest)
+				}
+				return
+			}
+			if diags.HasErrors() {
+				t.Fatalf("unexpected error: %s", diags.Err())
+			}
+			if !gotState.RawEquals(tt.wantState) {
+				t.Fatalf("unexpected state: got %#v, want %#v", gotState, tt.wantState)
+			}
+			if !bytes.Equal(gotPrivate, tt.wantPrivate) {
+				t.Fatalf("unexpected private: got %#v, want %#v", gotPrivate, tt.wantPrivate)
+			}
+		})
+	}
+}
+
+func getDataResourceModeInstance() addrs.AbsResourceInstance {
+	addr := mustResourceInstanceAddr("foo_instance.cur")
+	addr.Resource.Resource.Mode = addrs.DataResourceMode
+	return addr
+}
+
+func TestTransformResourceState(t *testing.T) {
+	tests := []struct {
+		name           string
+		args           stateTransformArgs
+		stateTransform providerStateTransform
+		wantErr        string
+		// Callback to validate the object source after transformation.
+		objectSrcValidator func(*testing.T, *states.ResourceInstanceObjectSrc)
+	}{
+		// We should never have flatmap state in the returned ObjectSrc from transformResourceState (Ensured by objectSrc.CompleteUpgrade)
+		{
+			name: "flatmap state should be removed after transformation",
+			args: stateTransformArgs{
+				currentAddr: mustResourceInstanceAddr("test_instance.foo"),
+				provider: &MockProvider{
+					GetProviderSchemaResponse: testProviderSchema("test"),
+				},
+				objectSrc: &states.ResourceInstanceObjectSrc{
+					AttrsFlat: map[string]string{"foo": "bar"},
+					AttrsJSON: []byte(`{"foo":"bar"}`),
+				},
+				currentSchema: &configschema.Block{
+					Attributes: map[string]*configschema.Attribute{
+						"foo": {Type: cty.String},
+					},
+				},
+			},
+			stateTransform: func(args stateTransformArgs) (cty.Value, []byte, tfdiags.Diagnostics) {
+				return cty.ObjectVal(map[string]cty.Value{
+					"foo": cty.StringVal("bar"),
+				}), nil, nil
+			},
+			objectSrcValidator: func(t *testing.T, got *states.ResourceInstanceObjectSrc) {
+				if got.AttrsFlat != nil {
+					t.Error("AttrsFlat should be nil after transformation")
+				}
+			},
+		},
+		// State must pass TestConformance, i.e. be the same type as the current schema, otherwise expect error
+		{
+			name: "state must conform to schema",
+			args: stateTransformArgs{
+				currentAddr: mustResourceInstanceAddr("test_instance.foo"),
+				provider: &MockProvider{
+					GetProviderSchemaResponse: testProviderSchema("test"),
+				},
+				objectSrc: &states.ResourceInstanceObjectSrc{
+					AttrsJSON: []byte(`{"foo":"bar"}`),
+				},
+				currentSchema: &configschema.Block{
+					Attributes: map[string]*configschema.Attribute{
+						"baz": {Type: cty.String}, // different attribute
+					},
+				},
+			},
+			stateTransform: func(args stateTransformArgs) (cty.Value, []byte, tfdiags.Diagnostics) {
+				return cty.ObjectVal(map[string]cty.Value{
+					"foo": cty.StringVal("bar"), // doesn't match schema
+				}), nil, nil
+			},
+			wantErr: "Invalid resource state transformation",
+		},
+		// Non-Managed resources should not be upgraded and should return the same object source without errors
+		{
+			name: "non-managed resource should not be upgraded",
+			args: stateTransformArgs{
+				currentAddr: getDataResourceModeInstance(),
+				objectSrc: &states.ResourceInstanceObjectSrc{
+					AttrsJSON: []byte(`{"foo":"bar"}`),
+				},
+			},
+			stateTransform: func(args stateTransformArgs) (cty.Value, []byte, tfdiags.Diagnostics) {
+				t.Error("stateTransform should not be called for data sources")
+				return cty.NilVal, nil, nil
+			},
+			objectSrcValidator: func(t *testing.T, got *states.ResourceInstanceObjectSrc) {
+				if !bytes.Equal(got.AttrsJSON, []byte(`{"foo":"bar"}`)) {
+					t.Error("state should remain unchanged")
+				}
+			},
+		},
+		// The new private state (returned from the provider/state transform call) should be set in the object source
+		{
+			name: "private state should be updated",
+			args: stateTransformArgs{
+				currentAddr: mustResourceInstanceAddr("test_instance.foo"),
+				provider: &MockProvider{
+					GetProviderSchemaResponse: testProviderSchema("test"),
+				},
+				objectSrc: &states.ResourceInstanceObjectSrc{
+					AttrsJSON: []byte(`{"foo":"bar"}`),
+					Private:   []byte("old-private"),
+				},
+				currentSchema: &configschema.Block{
+					Attributes: map[string]*configschema.Attribute{
+						"foo": {Type: cty.String},
+					},
+				},
+			},
+			stateTransform: func(args stateTransformArgs) (cty.Value, []byte, tfdiags.Diagnostics) {
+				return cty.ObjectVal(map[string]cty.Value{
+					"foo": cty.StringVal("bar"),
+				}), []byte("new-private"), nil
+			},
+			objectSrcValidator: func(t *testing.T, got *states.ResourceInstanceObjectSrc) {
+				if !bytes.Equal(got.Private, []byte("new-private")) {
+					t.Error("private state not updated")
+				}
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, diags := transformResourceState(tt.args, tt.stateTransform)
+
+			if tt.wantErr != "" {
+				if !diags.HasErrors() {
+					t.Fatal("expected error diagnostics")
+				}
+				if !strings.Contains(diags.Err().Error(), tt.wantErr) {
+					t.Errorf("expected error containing %q, got %q", tt.wantErr, diags.Err())
+				}
+				return
+			}
+
+			if diags.HasErrors() {
+				t.Fatalf("unexpected errors: %s", diags.Err())
+			}
+
+			if tt.objectSrcValidator != nil {
+				tt.objectSrcValidator(t, got)
+			}
+		})
+	}
+}

website/docs/language/modules/develop/refactoring.mdx

@@ -45,6 +45,7 @@ describe several refactoring use-cases and the appropriate addressing syntax
 for each situation.
 
 * [Renaming a Resource](#renaming-a-resource)
+* [Changing a Resource Type](#changing-a-resource-type)
 * [Enabling `count` or `for_each` For a Resource](#enabling-count-or-for_each-for-a-resource)
 * [Renaming a Module Call](#renaming-a-module-call)
 * [Enabling `count` or `for_each` For a Module Call](#enabling-count-or-for_each-for-a-module-call)
@@ -96,11 +97,15 @@ so OpenTofu recognizes the move for all instances of the resource. That is,
 it covers both `aws_instance.a[0]` and `aws_instance.a[1]` without the need
 to identify each one separately.
 
-Each resource type has a separate schema and so objects of different types
-are not compatible. Therefore, although you can use `moved` to change the name
-of a resource, you _cannot_ use `moved` to change to a different resource type
-or to change a managed resource (a `resource` block) into a data resource
-(a `data` block).
+## Changing a Resource Type
+
+Each resource type has a separate schema, so objects of different types are not generally compatible.
+However, some providers will let you change an object from one resource type to another.
+It is also possible to change one provider's resource to another provider's resource, as long as the new provider supports migration from the old resource.
+Refer to the provider documentation for more information about resource compatibility.
+
+You can use `moved` to change the name (and sometimes type) of a resource,
+but you _cannot_ use `moved` to change a managed resource (a `resource` block) into a data resource (a `data` block).
 
 ## Enabling `count` or `for_each` For a Resource