diff --git a/internal/terraform/context_import.go b/internal/terraform/context_import.go
index d809d6bb9d..91ea372ba7 100644
--- a/internal/terraform/context_import.go
+++ b/internal/terraform/context_import.go
@@ -56,11 +56,14 @@ func (c *Context) Import(config *configs.Config, prevRunState *states.State, opt
 	variables := opts.SetVariables
 
 	// Initialize our graph builder
-	builder := &ImportGraphBuilder{
+	builder := &PlanGraphBuilder{
 		ImportTargets:      opts.Targets,
 		Config:             config,
+		State:              state,
 		RootVariableValues: variables,
 		Plugins:            c.plugins,
+		skipRefresh:        true,
+		Operation:          walkImport,
 	}
 
 	// Build the graph
diff --git a/internal/terraform/context_plan.go b/internal/terraform/context_plan.go
index 3364bca019..c832de529f 100644
--- a/internal/terraform/context_plan.go
+++ b/internal/terraform/context_plan.go
@@ -560,6 +560,7 @@ func (c *Context) planGraph(config *configs.Config, prevRunState *states.State,
 			Targets:            opts.Targets,
 			ForceReplace:       opts.ForceReplace,
 			skipRefresh:        opts.SkipRefresh,
+			Operation:          walkPlan,
 		}).Build(addrs.RootModuleInstance)
 		return graph, walkPlan, diags
 	case plans.RefreshOnlyMode:
@@ -571,16 +572,18 @@ func (c *Context) planGraph(config *configs.Config, prevRunState *states.State,
 			Targets:            opts.Targets,
 			skipRefresh:        opts.SkipRefresh,
 			skipPlanChanges:    true, // this activates "refresh only" mode.
+			Operation:          walkPlan,
 		}).Build(addrs.RootModuleInstance)
 		return graph, walkPlan, diags
 	case plans.DestroyMode:
-		graph, diags := DestroyPlanGraphBuilder(&PlanGraphBuilder{
+		graph, diags := (&PlanGraphBuilder{
 			Config:             config,
 			State:              prevRunState,
 			RootVariableValues: opts.SetVariables,
 			Plugins:            c.plugins,
 			Targets:            opts.Targets,
 			skipRefresh:        opts.SkipRefresh,
+			Operation:          walkPlanDestroy,
 		}).Build(addrs.RootModuleInstance)
 		return graph, walkPlanDestroy, diags
 	default:
diff --git a/internal/terraform/context_validate.go b/internal/terraform/context_validate.go
index 070694920c..aad884442a 100644
--- a/internal/terraform/context_validate.go
+++ b/internal/terraform/context_validate.go
@@ -55,11 +55,12 @@ func (c *Context) Validate(config *configs.Config) tfdiags.Diagnostics {
 		}
 	}
 
-	graph, moreDiags := ValidateGraphBuilder(&PlanGraphBuilder{
+	graph, moreDiags := (&PlanGraphBuilder{
 		Config:             config,
 		Plugins:            c.plugins,
 		State:              states.NewState(),
 		RootVariableValues: varValues,
+		Operation:          walkValidate,
 	}).Build(addrs.RootModuleInstance)
 	diags = diags.Append(moreDiags)
 	if moreDiags.HasErrors() {
diff --git a/internal/terraform/context_walk.go b/internal/terraform/context_walk.go
index 1bc7491d06..cc61331cbb 100644
--- a/internal/terraform/context_walk.go
+++ b/internal/terraform/context_walk.go
@@ -73,7 +73,7 @@ func (c *Context) graphWalker(operation walkOperation, opts *graphWalkOpts) *Con
 		refreshState = states.NewState().SyncWrapper()
 		prevRunState = states.NewState().SyncWrapper()
 
-	case walkPlan, walkPlanDestroy:
+	case walkPlan, walkPlanDestroy, walkImport:
 		state = inputState.DeepCopy().SyncWrapper()
 		refreshState = inputState.DeepCopy().SyncWrapper()
 		prevRunState = inputState.DeepCopy().SyncWrapper()
diff --git a/internal/terraform/graph_builder_destroy_plan.go b/internal/terraform/graph_builder_destroy_plan.go
deleted file mode 100644
index 9e29d7ceb6..0000000000
--- a/internal/terraform/graph_builder_destroy_plan.go
+++ /dev/null
@@ -1,17 +0,0 @@
-package terraform
-
-import (
-	"github.com/hashicorp/terraform/internal/dag"
-)
-
-func DestroyPlanGraphBuilder(p *PlanGraphBuilder) GraphBuilder {
-	p.ConcreteResourceInstance = func(a *NodeAbstractResourceInstance) dag.Vertex {
-		return &NodePlanDestroyableResourceInstance{
-			NodeAbstractResourceInstance: a,
-			skipRefresh:                  p.skipRefresh,
-		}
-	}
-	p.destroy = true
-
-	return p
-}
diff --git a/internal/terraform/graph_builder_import.go b/internal/terraform/graph_builder_import.go
deleted file mode 100644
index 79c0724f76..0000000000
--- a/internal/terraform/graph_builder_import.go
+++ /dev/null
@@ -1,101 +0,0 @@
-package terraform
-
-import (
-	"github.com/hashicorp/terraform/internal/addrs"
-	"github.com/hashicorp/terraform/internal/configs"
-	"github.com/hashicorp/terraform/internal/dag"
-	"github.com/hashicorp/terraform/internal/tfdiags"
-)
-
-// ImportGraphBuilder implements GraphBuilder and is responsible for building
-// a graph for importing resources into Terraform. This is a much, much
-// simpler graph than a normal configuration graph.
-type ImportGraphBuilder struct {
-	// ImportTargets are the list of resources to import.
-	ImportTargets []*ImportTarget
-
-	// Module is a configuration to build the graph from. See ImportOpts.Config.
-	Config *configs.Config
-
-	// RootVariableValues are the raw input values for root input variables
-	// given by the caller, which we'll resolve into final values as part
-	// of the plan walk.
-	RootVariableValues InputValues
-
-	// Plugins is a library of plug-in components (providers and
-	// provisioners) available for use.
-	Plugins *contextPlugins
-}
-
-// Build builds the graph according to the steps returned by Steps.
-func (b *ImportGraphBuilder) Build(path addrs.ModuleInstance) (*Graph, tfdiags.Diagnostics) {
-	return (&BasicGraphBuilder{
-		Steps: b.Steps(),
-		Name:  "ImportGraphBuilder",
-	}).Build(path)
-}
-
-// Steps returns the ordered list of GraphTransformers that must be executed
-// to build a complete graph.
-func (b *ImportGraphBuilder) Steps() []GraphTransformer {
-	// Get the module. If we don't have one, we just use an empty tree
-	// so that the transform still works but does nothing.
-	config := b.Config
-	if config == nil {
-		config = configs.NewEmptyConfig()
-	}
-
-	// Custom factory for creating providers.
-	concreteProvider := func(a *NodeAbstractProvider) dag.Vertex {
-		return &NodeApplyableProvider{
-			NodeAbstractProvider: a,
-		}
-	}
-
-	steps := []GraphTransformer{
-		// Create all our resources from the configuration and state
-		&ConfigTransformer{Config: config},
-
-		// Add dynamic values
-		&RootVariableTransformer{Config: b.Config, RawValues: b.RootVariableValues},
-		&ModuleVariableTransformer{Config: b.Config},
-		&LocalTransformer{Config: b.Config},
-		&OutputTransformer{Config: b.Config},
-
-		// Attach the configuration to any resources
-		&AttachResourceConfigTransformer{Config: b.Config},
-
-		// Add the import steps
-		&ImportStateTransformer{Targets: b.ImportTargets, Config: b.Config},
-
-		transformProviders(concreteProvider, config),
-
-		// Must attach schemas before ReferenceTransformer so that we can
-		// analyze the configuration to find references.
-		&AttachSchemaTransformer{Plugins: b.Plugins, Config: b.Config},
-
-		// Create expansion nodes for all of the module calls. This must
-		// come after all other transformers that create nodes representing
-		// objects that can belong to modules.
-		&ModuleExpansionTransformer{Config: b.Config},
-
-		// Connect so that the references are ready for targeting. We'll
-		// have to connect again later for providers and so on.
-		&ReferenceTransformer{},
-
-		// Make sure data sources are aware of any depends_on from the
-		// configuration
-		&attachDataResourceDependsOnTransformer{},
-
-		// Close opened plugin connections
-		&CloseProviderTransformer{},
-
-		// Close root module
-		&CloseRootModuleTransformer{},
-
-		// Optimize
-		&TransitiveReductionTransformer{},
-	}
-
-	return steps
-}
diff --git a/internal/terraform/graph_builder_plan.go b/internal/terraform/graph_builder_plan.go
index ed63675d22..bcd0102d3e 100644
--- a/internal/terraform/graph_builder_plan.go
+++ b/internal/terraform/graph_builder_plan.go
@@ -1,6 +1,8 @@
 package terraform
 
 import (
+	"log"
+
 	"github.com/hashicorp/terraform/internal/addrs"
 	"github.com/hashicorp/terraform/internal/configs"
 	"github.com/hashicorp/terraform/internal/dag"
@@ -63,12 +65,16 @@ type PlanGraphBuilder struct {
 	ConcreteResourceInstanceDeposed ConcreteResourceInstanceDeposedNodeFunc
 	ConcreteModule                  ConcreteModuleNodeFunc
 
-	// destroy is set to true when create a full destroy plan.
-	destroy bool
+	// Plan Operation this graph will be used for.
+	Operation walkOperation
+
+	// ImportTargets are the list of resources to import.
+	ImportTargets []*ImportTarget
 }
 
 // See GraphBuilder
 func (b *PlanGraphBuilder) Build(path addrs.ModuleInstance) (*Graph, tfdiags.Diagnostics) {
+	log.Printf("[TRACE] building graph for %s", b.Operation)
 	return (&BasicGraphBuilder{
 		Steps: b.Steps(),
 		Name:  "PlanGraphBuilder",
@@ -77,14 +83,29 @@ func (b *PlanGraphBuilder) Build(path addrs.ModuleInstance) (*Graph, tfdiags.Dia
 
 // See GraphBuilder
 func (b *PlanGraphBuilder) Steps() []GraphTransformer {
-	b.init()
+	switch b.Operation {
+	case walkPlan:
+		b.initPlan()
+	case walkPlanDestroy:
+		b.initDestroy()
+	case walkValidate:
+		b.initValidate()
+	case walkImport:
+		b.initImport()
+	default:
+		panic("invalid plan operation: " + b.Operation.String())
+	}
 
 	steps := []GraphTransformer{
 		// Creates all the resources represented in the config
 		&ConfigTransformer{
 			Concrete: b.ConcreteResource,
 			Config:   b.Config,
-			skip:     b.destroy,
+
+			// Resources are not added from the config on destroy.
+			skip: b.Operation == walkPlanDestroy,
+
+			importTargets: b.ImportTargets,
 		},
 
 		// Add dynamic values
@@ -94,7 +115,7 @@ func (b *PlanGraphBuilder) Steps() []GraphTransformer {
 		&OutputTransformer{
 			Config:            b.Config,
 			RefreshOnly:       b.skipPlanChanges,
-			removeRootOutputs: b.destroy,
+			removeRootOutputs: b.Operation == walkPlanDestroy,
 		},
 
 		// Add orphan resources
@@ -102,13 +123,14 @@ func (b *PlanGraphBuilder) Steps() []GraphTransformer {
 			Concrete: b.ConcreteResourceOrphan,
 			State:    b.State,
 			Config:   b.Config,
-			skip:     b.destroy,
+			skip:     b.Operation == walkPlanDestroy,
 		},
 
 		// We also need nodes for any deposed instance objects present in the
-		// state, so we can plan to destroy them. (This intentionally
-		// skips creating nodes for _current_ objects, since ConfigTransformer
-		// created nodes that will do that during DynamicExpand.)
+		// state, so we can plan to destroy them. (During plan this will
+		// intentionally skips creating nodes for _current_ objects, since
+		// ConfigTransformer created nodes that will do that during
+		// DynamicExpand.)
 		&StateTransformer{
 			ConcreteCurrent: b.ConcreteResourceInstance,
 			ConcreteDeposed: b.ConcreteResourceInstanceDeposed,
@@ -172,12 +194,7 @@ func (b *PlanGraphBuilder) Steps() []GraphTransformer {
 	return steps
 }
 
-func (b *PlanGraphBuilder) init() {
-	// Do nothing if the user requests customizing the fields
-	if b.CustomConcrete {
-		return
-	}
-
+func (b *PlanGraphBuilder) initPlan() {
 	b.ConcreteProvider = func(a *NodeAbstractProvider) dag.Vertex {
 		return &NodeApplyableProvider{
 			NodeAbstractProvider: a,
@@ -210,5 +227,61 @@ func (b *PlanGraphBuilder) init() {
 			skipPlanChanges: b.skipPlanChanges,
 		}
 	}
-
+}
+
+func (b *PlanGraphBuilder) initDestroy() {
+	b.initPlan()
+
+	b.ConcreteResourceInstance = func(a *NodeAbstractResourceInstance) dag.Vertex {
+		return &NodePlanDestroyableResourceInstance{
+			NodeAbstractResourceInstance: a,
+			skipRefresh:                  b.skipRefresh,
+		}
+	}
+}
+
+func (b *PlanGraphBuilder) initValidate() {
+	// Set the provider to the normal provider. This will ask for input.
+	b.ConcreteProvider = func(a *NodeAbstractProvider) dag.Vertex {
+		return &NodeApplyableProvider{
+			NodeAbstractProvider: a,
+		}
+	}
+
+	b.ConcreteResource = func(a *NodeAbstractResource) dag.Vertex {
+		return &NodeValidatableResource{
+			NodeAbstractResource: a,
+		}
+	}
+
+	b.ConcreteModule = func(n *nodeExpandModule) dag.Vertex {
+		return &nodeValidateModule{
+			nodeExpandModule: *n,
+		}
+	}
+}
+
+func (b *PlanGraphBuilder) initImport() {
+	b.ConcreteProvider = func(a *NodeAbstractProvider) dag.Vertex {
+		return &NodeApplyableProvider{
+			NodeAbstractProvider: a,
+		}
+	}
+
+	b.ConcreteResource = func(a *NodeAbstractResource) dag.Vertex {
+		return &nodeExpandPlannableResource{
+			NodeAbstractResource: a,
+
+			// For now we always skip planning changes for import, since we are
+			// not going to combine importing with other changes. This is
+			// temporary to try and maintain existing import behaviors, but
+			// planning will need to be allowed for more complex configurations.
+			skipPlanChanges: true,
+
+			// We also skip refresh for now, since the plan output is written
+			// as the new state, and users are not expecting the import process
+			// to update any other instances in state.
+			skipRefresh: true,
+		}
+	}
 }
diff --git a/internal/terraform/graph_builder_plan_test.go b/internal/terraform/graph_builder_plan_test.go
index 9ec16c6ed7..8e6f5cdaf6 100644
--- a/internal/terraform/graph_builder_plan_test.go
+++ b/internal/terraform/graph_builder_plan_test.go
@@ -34,8 +34,9 @@ func TestPlanGraphBuilder(t *testing.T) {
 	}, nil)
 
 	b := &PlanGraphBuilder{
-		Config:  testModule(t, "graph-builder-plan-basic"),
-		Plugins: plugins,
+		Config:    testModule(t, "graph-builder-plan-basic"),
+		Plugins:   plugins,
+		Operation: walkPlan,
 	}
 
 	g, err := b.Build(addrs.RootModuleInstance)
@@ -76,8 +77,9 @@ func TestPlanGraphBuilder_dynamicBlock(t *testing.T) {
 	}, nil)
 
 	b := &PlanGraphBuilder{
-		Config:  testModule(t, "graph-builder-plan-dynblock"),
-		Plugins: plugins,
+		Config:    testModule(t, "graph-builder-plan-dynblock"),
+		Plugins:   plugins,
+		Operation: walkPlan,
 	}
 
 	g, err := b.Build(addrs.RootModuleInstance)
@@ -131,8 +133,9 @@ func TestPlanGraphBuilder_attrAsBlocks(t *testing.T) {
 	}, nil)
 
 	b := &PlanGraphBuilder{
-		Config:  testModule(t, "graph-builder-plan-attr-as-blocks"),
-		Plugins: plugins,
+		Config:    testModule(t, "graph-builder-plan-attr-as-blocks"),
+		Plugins:   plugins,
+		Operation: walkPlan,
 	}
 
 	g, err := b.Build(addrs.RootModuleInstance)
@@ -173,6 +176,7 @@ func TestPlanGraphBuilder_targetModule(t *testing.T) {
 		Targets: []addrs.Targetable{
 			addrs.RootModuleInstance.Child("child2", addrs.NoKey),
 		},
+		Operation: walkPlan,
 	}
 
 	g, err := b.Build(addrs.RootModuleInstance)
@@ -194,8 +198,9 @@ func TestPlanGraphBuilder_forEach(t *testing.T) {
 	}, nil)
 
 	b := &PlanGraphBuilder{
-		Config:  testModule(t, "plan-for-each"),
-		Plugins: plugins,
+		Config:    testModule(t, "plan-for-each"),
+		Plugins:   plugins,
+		Operation: walkPlan,
 	}
 
 	g, err := b.Build(addrs.RootModuleInstance)
diff --git a/internal/terraform/graph_builder_validate.go b/internal/terraform/graph_builder_validate.go
deleted file mode 100644
index ff5148216c..0000000000
--- a/internal/terraform/graph_builder_validate.go
+++ /dev/null
@@ -1,40 +0,0 @@
-package terraform
-
-import (
-	"github.com/hashicorp/terraform/internal/dag"
-)
-
-// ValidateGraphBuilder creates the graph for the validate operation.
-//
-// ValidateGraphBuilder is based on the PlanGraphBuilder. We do this so that
-// we only have to validate what we'd normally plan anyways. The
-// PlanGraphBuilder given will be modified so it shouldn't be used for anything
-// else after calling this function.
-func ValidateGraphBuilder(p *PlanGraphBuilder) GraphBuilder {
-	// We're going to customize the concrete functions
-	p.CustomConcrete = true
-
-	// Set the provider to the normal provider. This will ask for input.
-	p.ConcreteProvider = func(a *NodeAbstractProvider) dag.Vertex {
-		return &NodeApplyableProvider{
-			NodeAbstractProvider: a,
-		}
-	}
-
-	p.ConcreteResource = func(a *NodeAbstractResource) dag.Vertex {
-		return &NodeValidatableResource{
-			NodeAbstractResource: a,
-		}
-	}
-
-	p.ConcreteModule = func(n *nodeExpandModule) dag.Vertex {
-		return &nodeValidateModule{
-			nodeExpandModule: *n,
-		}
-	}
-
-	// We purposely don't set any other concrete types since they don't
-	// require validation.
-
-	return p
-}
diff --git a/internal/terraform/node_resource_abstract.go b/internal/terraform/node_resource_abstract.go
index 1a5bd1ff5c..45941f13a1 100644
--- a/internal/terraform/node_resource_abstract.go
+++ b/internal/terraform/node_resource_abstract.go
@@ -68,6 +68,9 @@ type NodeAbstractResource struct {
 
 	// The address of the provider this resource will use
 	ResolvedProvider addrs.AbsProviderConfig
+
+	// This resource may expand into instances which need to be imported.
+	importTargets []*ImportTarget
 }
 
 var (
@@ -124,6 +127,10 @@ func (n *NodeAbstractResource) ReferenceableAddrs() []addrs.Referenceable {
 	return []addrs.Referenceable{n.Addr.Resource}
 }
 
+func (n *NodeAbstractResource) Import(addr *ImportTarget) {
+
+}
+
 // GraphNodeReferencer
 func (n *NodeAbstractResource) References() []*addrs.Reference {
 	// If we have a config then we prefer to use that.
diff --git a/internal/terraform/node_resource_plan.go b/internal/terraform/node_resource_plan.go
index 5a01d9337d..8314b4cc11 100644
--- a/internal/terraform/node_resource_plan.go
+++ b/internal/terraform/node_resource_plan.go
@@ -319,6 +319,17 @@ func (n *NodePlannableResource) DynamicExpand(ctx EvalContext) (*Graph, error) {
 
 	// The concrete resource factory we'll use
 	concreteResource := func(a *NodeAbstractResourceInstance) dag.Vertex {
+		// check if this node is being imported first
+		for _, importTarget := range n.importTargets {
+			if importTarget.Addr.Equal(a.Addr) {
+				return &graphNodeImportState{
+					Addr:             importTarget.Addr,
+					ID:               importTarget.ID,
+					ResolvedProvider: n.ResolvedProvider,
+				}
+			}
+		}
+
 		// Add the config and state since we don't do that via transforms
 		a.Config = n.Config
 		a.ResolvedProvider = n.ResolvedProvider
diff --git a/internal/terraform/testdata/import-provider/main.tf b/internal/terraform/testdata/import-provider/main.tf
index ed8e3fe9fd..328624e0d9 100644
--- a/internal/terraform/testdata/import-provider/main.tf
+++ b/internal/terraform/testdata/import-provider/main.tf
@@ -3,5 +3,5 @@ provider "aws" {
 }
 
 resource "aws_instance" "foo" {
-  id = "bar"
+  #id = "bar"
 }
diff --git a/internal/terraform/transform_config.go b/internal/terraform/transform_config.go
index 3895efebb2..59fa1eeea9 100644
--- a/internal/terraform/transform_config.go
+++ b/internal/terraform/transform_config.go
@@ -31,6 +31,9 @@ type ConfigTransformer struct {
 
 	// Do not apply this transformer.
 	skip bool
+
+	// configuration resources that are to be imported
+	importTargets []*ImportTarget
 }
 
 func (t *ConfigTransformer) Transform(g *Graph) error {
@@ -89,11 +92,22 @@ func (t *ConfigTransformer) transformSingle(g *Graph, config *configs.Config) er
 			continue
 		}
 
+		// If any of the import targets can apply to this node's instances,
+		// filter them down to the applicable addresses.
+		var imports []*ImportTarget
+		configAddr := relAddr.InModule(path)
+		for _, i := range t.importTargets {
+			if target := i.Addr.ContainingResource().Config(); target.Equal(configAddr) {
+				imports = append(imports, i)
+			}
+		}
+
 		abstract := &NodeAbstractResource{
 			Addr: addrs.ConfigResource{
 				Resource: relAddr,
 				Module:   path,
 			},
+			importTargets: imports,
 		}
 
 		var node dag.Vertex = abstract
diff --git a/internal/terraform/transform_import_state.go b/internal/terraform/transform_import_state.go
index 3aa53e22d7..a7629dbee0 100644
--- a/internal/terraform/transform_import_state.go
+++ b/internal/terraform/transform_import_state.go
@@ -16,11 +16,15 @@ import (
 type ImportStateTransformer struct {
 	Targets []*ImportTarget
 	Config  *configs.Config
+	skip    bool
 }
 
 func (t *ImportStateTransformer) Transform(g *Graph) error {
-	for _, target := range t.Targets {
+	if t.skip {
+		return nil
+	}
 
+	for _, target := range t.Targets {
 		// This is only likely to happen in misconfigured tests
 		if t.Config == nil {
 			return fmt.Errorf("cannot import into an empty configuration")