diff --git a/third_party/terraform/resources/resource_dataproc_cluster.go.erb b/third_party/terraform/resources/resource_dataproc_cluster.go.erb
index 75dcbceee89f..6e33a85342a5 100644
--- a/third_party/terraform/resources/resource_dataproc_cluster.go.erb
+++ b/third_party/terraform/resources/resource_dataproc_cluster.go.erb
@@ -496,7 +496,7 @@ func resourceDataprocClusterCreate(d *schema.ResourceData, meta interface{}) err
 		return fmt.Errorf("Error creating Dataproc cluster: %s", err)
 	}
 
-	d.SetId(cluster.ClusterName)
+	d.SetId(fmt.Sprintf("projects/%s/regions/%s/clusters/%s", project, region, cluster.ClusterName))
 
 	// Wait until it's created
 	timeoutInMinutes := int(d.Timeout(schema.TimeoutCreate).Minutes())
diff --git a/third_party/terraform/resources/resource_dataproc_job.go b/third_party/terraform/resources/resource_dataproc_job.go
index ffc9658738f6..d7e2ccdf23e4 100644
--- a/third_party/terraform/resources/resource_dataproc_job.go
+++ b/third_party/terraform/resources/resource_dataproc_job.go
@@ -3,6 +3,7 @@ package google
 import (
 	"fmt"
 	"log"
+	"strings"
 	"time"
 
 	"github.com/hashicorp/terraform-plugin-sdk/helper/schema"
@@ -249,7 +250,7 @@ func resourceDataprocJobCreate(d *schema.ResourceData, meta interface{}) error {
 	if err != nil {
 		return err
 	}
-	d.SetId(job.Reference.JobId)
+	d.SetId(fmt.Sprintf("projects/%s/regions/%s/jobs/%s", project, region, job.Reference.JobId))
 
 	timeoutInMinutes := int(d.Timeout(schema.TimeoutCreate).Minutes())
 	waitErr := dataprocJobOperationWait(config, region, project, job.Reference.JobId,
@@ -271,10 +272,12 @@ func resourceDataprocJobRead(d *schema.ResourceData, meta interface{}) error {
 		return err
 	}
 
+	parts := strings.Split(d.Id(), "/")
+	jobId := parts[len(parts)-1]
 	job, err := config.clientDataproc.Projects.Regions.Jobs.Get(
-		project, region, d.Id()).Do()
+		project, region, jobId).Do()
 	if err != nil {
-		return handleNotFoundError(err, d, fmt.Sprintf("Dataproc Job %q", d.Id()))
+		return handleNotFoundError(err, d, fmt.Sprintf("Dataproc Job %q", jobId))
 	}
 
 	d.Set("force_delete", d.Get("force_delete"))
@@ -320,15 +323,17 @@ func resourceDataprocJobDelete(d *schema.ResourceData, meta interface{}) error {
 	forceDelete := d.Get("force_delete").(bool)
 	timeoutInMinutes := int(d.Timeout(schema.TimeoutDelete).Minutes())
 
+	parts := strings.Split(d.Id(), "/")
+	jobId := parts[len(parts)-1]
 	if forceDelete {
 		log.Printf("[DEBUG] Attempting to first cancel Dataproc job %s if it's still running ...", d.Id())
 
 		// ignore error if we get one - job may be finished already and not need to
 		// be cancelled. We do however wait for the state to be one that is
 		// at least not active
-		_, _ = config.clientDataproc.Projects.Regions.Jobs.Cancel(project, region, d.Id(), &dataproc.CancelJobRequest{}).Do()
+		_, _ = config.clientDataproc.Projects.Regions.Jobs.Cancel(project, region, jobId, &dataproc.CancelJobRequest{}).Do()
 
-		waitErr := dataprocJobOperationWait(config, region, project, d.Id(),
+		waitErr := dataprocJobOperationWait(config, region, project, jobId,
 			"Cancelling Dataproc job", timeoutInMinutes, 1)
 		if waitErr != nil {
 			return waitErr
@@ -338,12 +343,12 @@ func resourceDataprocJobDelete(d *schema.ResourceData, meta interface{}) error {
 
 	log.Printf("[DEBUG] Deleting Dataproc job %s", d.Id())
 	_, err = config.clientDataproc.Projects.Regions.Jobs.Delete(
-		project, region, d.Id()).Do()
+		project, region, jobId).Do()
 	if err != nil {
 		return err
 	}
 
-	waitErr := dataprocDeleteOperationWait(config, region, project, d.Id(),
+	waitErr := dataprocDeleteOperationWait(config, region, project, jobId,
 		"Deleting Dataproc job", timeoutInMinutes, 1)
 	if waitErr != nil {
 		return waitErr
diff --git a/third_party/terraform/resources/resource_google_project.go b/third_party/terraform/resources/resource_google_project.go
index 28a7a165ad8d..d6655836d178 100644
--- a/third_party/terraform/resources/resource_google_project.go
+++ b/third_party/terraform/resources/resource_google_project.go
@@ -124,7 +124,7 @@ func resourceGoogleProjectCreate(d *schema.ResourceData, meta interface{}) error
 			project.ProjectId, project.Name, err)
 	}
 
-	d.SetId(pid)
+	d.SetId(fmt.Sprintf("projects/%s", pid))
 
 	// Wait for the operation to complete
 	opAsMap, err := ConvertToMap(op)
@@ -175,7 +175,8 @@ func resourceGoogleProjectCreate(d *schema.ResourceData, meta interface{}) error
 
 func resourceGoogleProjectRead(d *schema.ResourceData, meta interface{}) error {
 	config := meta.(*Config)
-	pid := d.Id()
+	parts := strings.Split(d.Id(), "/")
+	pid := parts[len(parts)-1]
 
 	p, err := readGoogleProject(d, config)
 	if err != nil {
@@ -271,7 +272,8 @@ func parseFolderId(v interface{}) string {
 
 func resourceGoogleProjectUpdate(d *schema.ResourceData, meta interface{}) error {
 	config := meta.(*Config)
-	pid := d.Id()
+	parts := strings.Split(d.Id(), "/")
+	pid := parts[len(parts)-1]
 	project_name := d.Get("name").(string)
 
 	// Read the project
@@ -348,7 +350,8 @@ func resourceGoogleProjectDelete(d *schema.ResourceData, meta interface{}) error
 	config := meta.(*Config)
 	// Only delete projects if skip_delete isn't set
 	if !d.Get("skip_delete").(bool) {
-		pid := d.Id()
+		parts := strings.Split(d.Id(), "/")
+		pid := parts[len(parts)-1]
 		if err := retryTimeDuration(func() error {
 			_, delErr := config.clientResourceManager.Projects.Delete(pid).Do()
 			return delErr
@@ -361,7 +364,8 @@ func resourceGoogleProjectDelete(d *schema.ResourceData, meta interface{}) error
 }
 
 func resourceProjectImportState(d *schema.ResourceData, meta interface{}) ([]*schema.ResourceData, error) {
-	pid := d.Id()
+	parts := strings.Split(d.Id(), "/")
+	pid := parts[len(parts)-1]
 	// Prevent importing via project number, this will cause issues later
 	matched, err := regexp.MatchString("^\\d+$", pid)
 	if err != nil {
@@ -372,6 +376,9 @@ func resourceProjectImportState(d *schema.ResourceData, meta interface{}) ([]*sc
 		return nil, fmt.Errorf("Error importing project %q, please use project_id", pid)
 	}
 
+	// Ensure the id format includes projects/
+	d.SetId(fmt.Sprintf("projects/%s", pid))
+
 	// Explicitly set to default as a workaround for `ImportStateVerify` tests, and so that users
 	// don't see a diff immediately after import.
 	d.Set("auto_create_network", true)
@@ -414,7 +421,8 @@ func forceDeleteComputeNetwork(d *schema.ResourceData, config *Config, projectId
 }
 
 func updateProjectBillingAccount(d *schema.ResourceData, config *Config) error {
-	pid := d.Id()
+	parts := strings.Split(d.Id(), "/")
+	pid := parts[len(parts)-1]
 	name := d.Get("billing_account").(string)
 	ba := &cloudbilling.ProjectBillingInfo{}
 	// If we're unlinking an existing billing account, an empty request does that, not an empty-string billing account.
@@ -461,8 +469,10 @@ func deleteComputeNetwork(project, network string, config *Config) error {
 func readGoogleProject(d *schema.ResourceData, config *Config) (*cloudresourcemanager.Project, error) {
 	var p *cloudresourcemanager.Project
 	// Read the project
+	parts := strings.Split(d.Id(), "/")
+	pid := parts[len(parts)-1]
 	err := retryTimeDuration(func() (reqErr error) {
-		p, reqErr = config.clientResourceManager.Projects.Get(d.Id()).Do()
+		p, reqErr = config.clientResourceManager.Projects.Get(pid).Do()
 		return reqErr
 	}, d.Timeout(schema.TimeoutRead))
 	return p, err
diff --git a/third_party/terraform/resources/resource_sql_database_instance.go b/third_party/terraform/resources/resource_sql_database_instance.go
index dbf55492eecd..16966722af39 100644
--- a/third_party/terraform/resources/resource_sql_database_instance.go
+++ b/third_party/terraform/resources/resource_sql_database_instance.go
@@ -503,7 +503,11 @@ func resourceSqlDatabaseInstanceCreate(d *schema.ResourceData, meta interface{})
 		return fmt.Errorf("Error, failed to create instance %s: %s", instance.Name, err)
 	}
 
-	d.SetId(instance.Name)
+	id, err := replaceVars(d, config, "projects/{{project}}/instances/{{name}}")
+	if err != nil {
+		return fmt.Errorf("Error constructing id: %s", err)
+	}
+	d.SetId(id)
 
 	err = sqlAdminOperationWaitTime(config.clientSqlAdmin, op, project, "Create Instance", int(d.Timeout(schema.TimeoutCreate).Minutes()))
 	if err != nil {
@@ -708,7 +712,7 @@ func resourceSqlDatabaseInstanceRead(d *schema.ResourceData, meta interface{}) e
 
 	var instance *sqladmin.DatabaseInstance
 	err = retryTimeDuration(func() (rerr error) {
-		instance, rerr = config.clientSqlAdmin.Instances.Get(project, d.Id()).Do()
+		instance, rerr = config.clientSqlAdmin.Instances.Get(project, d.Get("name").(string)).Do()
 		return rerr
 	}, d.Timeout(schema.TimeoutRead), isSqlOperationInProgressError)
 	if err != nil {
@@ -843,7 +847,7 @@ func resourceSqlDatabaseInstanceImport(d *schema.ResourceData, meta interface{})
 	}
 
 	// Replace import id for the resource id
-	id, err := replaceVars(d, config, "{{name}}")
+	id, err := replaceVars(d, config, "projects/{{project}}/instances/{{name}}")
 	if err != nil {
 		return nil, fmt.Errorf("Error constructing id: %s", err)
 	}
diff --git a/third_party/terraform/resources/resource_sql_ssl_cert.go b/third_party/terraform/resources/resource_sql_ssl_cert.go
index edfbaba623ff..5afce1f0a18f 100644
--- a/third_party/terraform/resources/resource_sql_ssl_cert.go
+++ b/third_party/terraform/resources/resource_sql_ssl_cert.go
@@ -105,7 +105,7 @@ func resourceSqlSslCertCreate(d *schema.ResourceData, meta interface{}) error {
 	}
 
 	fingerprint := resp.ClientCert.CertInfo.Sha1Fingerprint
-	d.SetId(fmt.Sprintf("%s/%s", instance, fingerprint))
+	d.SetId(fmt.Sprintf("projects/%s/instances/%s/sslCerts/%s", project, instance, fingerprint))
 	d.Set("sha1_fingerprint", fingerprint)
 
 	// The private key is only returned on the initial insert so set it here.
@@ -148,7 +148,7 @@ func resourceSqlSslCertRead(d *schema.ResourceData, meta interface{}) error {
 	d.Set("create_time", sslCerts.CreateTime)
 	d.Set("expiration_time", sslCerts.ExpirationTime)
 
-	d.SetId(fmt.Sprintf("%s/%s", instance, fingerprint))
+	d.SetId(fmt.Sprintf("projects/%s/instances/%s/sslCerts/%s", project, instance, fingerprint))
 	return nil
 }
 
diff --git a/third_party/terraform/tests/resource_dataproc_cluster_test.go.erb b/third_party/terraform/tests/resource_dataproc_cluster_test.go.erb
index e5f2b2a184ee..0aeaa2a87291 100644
--- a/third_party/terraform/tests/resource_dataproc_cluster_test.go.erb
+++ b/third_party/terraform/tests/resource_dataproc_cluster_test.go.erb
@@ -624,8 +624,10 @@ func testAccCheckDataprocClusterDestroy() resource.TestCheckFunc {
 				return err
 			}
 
+			parts := strings.Split(rs.Primary.ID, "/")
+			clusterId := parts[len(parts)-1]
 			_, err = config.clientDataprocBeta.Projects.Regions.Clusters.Get(
-				project, attributes["region"], rs.Primary.ID).Do()
+				project, attributes["region"], clusterId).Do()
 
 			if err != nil {
 				if gerr, ok := err.(*googleapi.Error); ok && gerr.Code == http.StatusNotFound {
@@ -786,14 +788,16 @@ func testAccCheckDataprocClusterExists(n string, cluster *dataproc.Cluster) reso
 			return err
 		}
 
+		parts := strings.Split(rs.Primary.ID, "/")
+		clusterId := parts[len(parts)-1]
 		found, err := config.clientDataprocBeta.Projects.Regions.Clusters.Get(
-			project, rs.Primary.Attributes["region"], rs.Primary.ID).Do()
+			project, rs.Primary.Attributes["region"], clusterId).Do()
 		if err != nil {
 			return err
 		}
 
-		if found.ClusterName != rs.Primary.ID {
-			return fmt.Errorf("Dataproc cluster %s not found, found %s instead", rs.Primary.ID, cluster.ClusterName)
+		if found.ClusterName != clusterId {
+			return fmt.Errorf("Dataproc cluster %s not found, found %s instead", clusterId, cluster.ClusterName)
 		}
 
 		*cluster = *found
diff --git a/third_party/terraform/tests/resource_dataproc_job_test.go b/third_party/terraform/tests/resource_dataproc_job_test.go
index fcaa33bce425..4155c477a25e 100644
--- a/third_party/terraform/tests/resource_dataproc_job_test.go
+++ b/third_party/terraform/tests/resource_dataproc_job_test.go
@@ -283,8 +283,10 @@ func testAccCheckDataprocJobDestroy(s *terraform.State) error {
 			return err
 		}
 
+		parts := strings.Split(rs.Primary.ID, "/")
+		job_id := parts[len(parts)-1]
 		_, err = config.clientDataproc.Projects.Regions.Jobs.Get(
-			project, attributes["region"], rs.Primary.ID).Do()
+			project, attributes["region"], job_id).Do()
 		if err != nil {
 			if gerr, ok := err.(*googleapi.Error); ok && gerr.Code == 404 {
 				return nil
@@ -367,7 +369,8 @@ func testAccCheckDataprocJobExists(n string, job *dataproc.Job) resource.TestChe
 		}
 
 		config := testAccProvider.Meta().(*Config)
-		jobId := s.RootModule().Resources[n].Primary.ID
+		parts := strings.Split(s.RootModule().Resources[n].Primary.ID, "/")
+		jobId := parts[len(parts)-1]
 		project, err := getTestProject(s.RootModule().Resources[n].Primary, config)
 		if err != nil {
 			return err
diff --git a/third_party/terraform/tests/resource_google_project_test.go b/third_party/terraform/tests/resource_google_project_test.go
index 39ac099fe820..6df0cbcc17ce 100644
--- a/third_party/terraform/tests/resource_google_project_test.go
+++ b/third_party/terraform/tests/resource_google_project_test.go
@@ -200,8 +200,9 @@ func testAccCheckGoogleProjectExists(r, pid string) resource.TestCheckFunc {
 			return fmt.Errorf("No ID is set")
 		}
 
-		if rs.Primary.ID != pid {
-			return fmt.Errorf("Expected project %q to match ID %q in state", pid, rs.Primary.ID)
+		projectId := fmt.Sprintf("projects/%s", pid)
+		if rs.Primary.ID != projectId {
+			return fmt.Errorf("Expected project %q to match ID %q in state", projectId, rs.Primary.ID)
 		}
 
 		return nil