diff --git a/builtin/providers/test/data_source_test.go b/builtin/providers/test/data_source_test.go
index 3f4e5ada6..77e235f3d 100644
--- a/builtin/providers/test/data_source_test.go
+++ b/builtin/providers/test/data_source_test.go
@@ -99,3 +99,59 @@ resource "test_resource" "foo" {
 		},
 	})
 }
+
+// TestDataSource_dataSourceCountGrandChild tests that a grandchild data source
+// that is based off of count works, ie: dependency chain foo -> bar -> baz.
+// This was failing because CountBoundaryTransformer is being run during apply
+// instead of plan, which meant that it wasn't firing after data sources were
+// potentially changing state and causing diff/interpolation issues.
+//
+// This happens after the initial apply, after state is saved.
+func TestDataSource_dataSourceCountGrandChild(t *testing.T) {
+	resource.UnitTest(t, resource.TestCase{
+		Providers: testAccProviders,
+		CheckDestroy: func(s *terraform.State) error {
+			return nil
+		},
+		Steps: []resource.TestStep{
+			{
+				Config: dataSourceCountGrandChildConfig,
+			},
+			{
+				Config: dataSourceCountGrandChildConfig,
+				Check: func(s *terraform.State) error {
+					for _, v := range []string{"foo", "bar", "baz"} {
+						count := 0
+						for k := range s.RootModule().Resources {
+							if strings.HasPrefix(k, fmt.Sprintf("data.test_data_source.%s.", v)) {
+								count++
+							}
+						}
+
+						if count != 2 {
+							return fmt.Errorf("bad count for data.test_data_source.%s: %d", v, count)
+						}
+					}
+					return nil
+				},
+			},
+		},
+	})
+}
+
+const dataSourceCountGrandChildConfig = `
+data "test_data_source" "foo" {
+  count = 2
+  input = "one"
+}
+
+data "test_data_source" "bar" {
+  count = "${length(data.test_data_source.foo.*.id)}"
+  input = "${data.test_data_source.foo.*.output[count.index]}"
+}
+
+data "test_data_source" "baz" {
+  count = "${length(data.test_data_source.bar.*.id)}"
+  input = "${data.test_data_source.bar.*.output[count.index]}"
+}
+`
diff --git a/terraform/context_plan_test.go b/terraform/context_plan_test.go
index 3e8c54190..3851e406b 100644
--- a/terraform/context_plan_test.go
+++ b/terraform/context_plan_test.go
@@ -3154,3 +3154,146 @@ func TestContext2Plan_ignoreChangesWithFlatmaps(t *testing.T) {
 		t.Fatalf("bad:\n%s\n\nexpected\n\n%s", actual, expected)
 	}
 }
+
+// TestContext2Plan_resourceNestedCount ensures resource sets that depend on
+// the count of another resource set (ie: count of a data source that depends
+// on another data source's instance count - data.x.foo.*.id) get properly
+// normalized to the indexes they should be. This case comes up when there is
+// an existing state (after an initial apply).
+func TestContext2Plan_resourceNestedCount(t *testing.T) {
+	m := testModule(t, "nested-resource-count-plan")
+	p := testProvider("aws")
+	p.DiffFn = testDiffFn
+	p.RefreshFn = func(i *InstanceInfo, is *InstanceState) (*InstanceState, error) {
+		return is, nil
+	}
+	s := &State{
+		Modules: []*ModuleState{
+			&ModuleState{
+				Path: rootModulePath,
+				Resources: map[string]*ResourceState{
+					"aws_instance.foo.0": &ResourceState{
+						Type: "aws_instance",
+						Primary: &InstanceState{
+							ID: "foo0",
+							Attributes: map[string]string{
+								"id": "foo0",
+							},
+						},
+					},
+					"aws_instance.foo.1": &ResourceState{
+						Type: "aws_instance",
+						Primary: &InstanceState{
+							ID: "foo1",
+							Attributes: map[string]string{
+								"id": "foo1",
+							},
+						},
+					},
+					"aws_instance.bar.0": &ResourceState{
+						Type:         "aws_instance",
+						Dependencies: []string{"aws_instance.foo.*"},
+						Primary: &InstanceState{
+							ID: "bar0",
+							Attributes: map[string]string{
+								"id": "bar0",
+							},
+						},
+					},
+					"aws_instance.bar.1": &ResourceState{
+						Type:         "aws_instance",
+						Dependencies: []string{"aws_instance.foo.*"},
+						Primary: &InstanceState{
+							ID: "bar1",
+							Attributes: map[string]string{
+								"id": "bar1",
+							},
+						},
+					},
+					"aws_instance.baz.0": &ResourceState{
+						Type:         "aws_instance",
+						Dependencies: []string{"aws_instance.bar.*"},
+						Primary: &InstanceState{
+							ID: "baz0",
+							Attributes: map[string]string{
+								"id": "baz0",
+							},
+						},
+					},
+					"aws_instance.baz.1": &ResourceState{
+						Type:         "aws_instance",
+						Dependencies: []string{"aws_instance.bar.*"},
+						Primary: &InstanceState{
+							ID: "baz1",
+							Attributes: map[string]string{
+								"id": "baz1",
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	ctx := testContext2(t, &ContextOpts{
+		Module: m,
+		Providers: map[string]ResourceProviderFactory{
+			"aws": testProviderFuncFixed(p),
+		},
+		State: s,
+	})
+
+	w, e := ctx.Validate()
+	if len(w) > 0 {
+		t.Fatalf("warnings generated on validate: %#v", w)
+	}
+	if len(e) > 0 {
+		t.Fatalf("errors generated on validate: %#v", e)
+	}
+
+	_, err := ctx.Refresh()
+	if err != nil {
+		t.Fatalf("refresh err: %s", err)
+	}
+
+	plan, err := ctx.Plan()
+	if err != nil {
+		t.Fatalf("plan err: %s", err)
+	}
+
+	actual := strings.TrimSpace(plan.String())
+	expected := strings.TrimSpace(`
+DIFF:
+
+
+
+STATE:
+
+aws_instance.bar.0:
+  ID = bar0
+
+  Dependencies:
+    aws_instance.foo.*
+aws_instance.bar.1:
+  ID = bar1
+
+  Dependencies:
+    aws_instance.foo.*
+aws_instance.baz.0:
+  ID = baz0
+
+  Dependencies:
+    aws_instance.bar.*
+aws_instance.baz.1:
+  ID = baz1
+
+  Dependencies:
+    aws_instance.bar.*
+aws_instance.foo.0:
+  ID = foo0
+aws_instance.foo.1:
+  ID = foo1
+`)
+	if actual != expected {
+		t.Fatalf("bad:\n%s\n\nexpected\n\n%s", actual, expected)
+	}
+}
diff --git a/terraform/graph_builder_plan.go b/terraform/graph_builder_plan.go
index 02d869700..a6a3a90d4 100644
--- a/terraform/graph_builder_plan.go
+++ b/terraform/graph_builder_plan.go
@@ -113,6 +113,9 @@ func (b *PlanGraphBuilder) Steps() []GraphTransformer {
 		// have to connect again later for providers and so on.
 		&ReferenceTransformer{},
 
+		// Add the node to fix the state count boundaries
+		&CountBoundaryTransformer{},
+
 		// Target
 		&TargetsTransformer{Targets: b.Targets},
 
diff --git a/terraform/graph_builder_plan_test.go b/terraform/graph_builder_plan_test.go
index 23526a9ac..25578ebaf 100644
--- a/terraform/graph_builder_plan_test.go
+++ b/terraform/graph_builder_plan_test.go
@@ -29,7 +29,7 @@ func TestPlanGraphBuilder(t *testing.T) {
 	actual := strings.TrimSpace(g.String())
 	expected := strings.TrimSpace(testPlanGraphBuilderStr)
 	if actual != expected {
-		t.Fatalf("bad: %s", actual)
+		t.Fatalf("expected:\n%s\n\ngot:\n%s", expected, actual)
 	}
 }
 
@@ -61,6 +61,14 @@ aws_load_balancer.weblb
   provider.aws
 aws_security_group.firewall
   provider.aws
+meta.count-boundary (count boundary fixup)
+  aws_instance.web
+  aws_load_balancer.weblb
+  aws_security_group.firewall
+  openstack_floating_ip.random
+  provider.aws
+  provider.openstack
+  var.foo
 openstack_floating_ip.random
   provider.openstack
 provider.aws
@@ -75,6 +83,7 @@ provider.openstack (close)
   openstack_floating_ip.random
   provider.openstack
 root
+  meta.count-boundary (count boundary fixup)
   provider.aws (close)
   provider.openstack (close)
 var.foo
diff --git a/terraform/test-fixtures/nested-resource-count-plan/main.tf b/terraform/test-fixtures/nested-resource-count-plan/main.tf
new file mode 100644
index 000000000..f803fd1f6
--- /dev/null
+++ b/terraform/test-fixtures/nested-resource-count-plan/main.tf
@@ -0,0 +1,11 @@
+resource "aws_instance" "foo" {
+  count = 2
+}
+
+resource "aws_instance" "bar" {
+  count = "${length(aws_instance.foo.*.id)}"
+}
+
+resource "aws_instance" "baz" {
+  count = "${length(aws_instance.bar.*.id)}"
+}