f/dynamodb_table: add restore_source_table_arn to support cross region table restore

.changelog/38953.txt

@@ -0,0 +1,3 @@
+```release-note:enhancement
+resource/aws_dynamodb_table: Add `restore_source_table_arn` attribute
+```

internal/service/dynamodb/table.go

@@ -103,6 +103,11 @@ func resourceTable() *schema.Resource {
 					return nil
 				}
 
+				if !diff.GetRawPlan().GetAttr("restore_source_table_arn").IsWhollyKnown() ||
+					diff.Get("restore_source_table_arn") != "" {
+					return nil
+				}
+
 				var errs []error
 				if err := validateProvisionedThroughputField(diff, "read_capacity"); err != nil {
 					errs = append(errs, err)
@@ -117,6 +122,9 @@ func resourceTable() *schema.Resource {
 				// https://github.com/hashicorp/terraform-provider-aws/issues/25214
 				return old.(string) != new.(string) && new.(string) != ""
 			}),
+			customdiff.ForceNewIfChange("restore_source_table_arn", func(_ context.Context, old, new, meta interface{}) bool {
+				return old.(string) != new.(string) && new.(string) != ""
+			}),
 			validateTTLCustomDiff,
 			verify.SetTagsDiff,
 		),
@@ -309,7 +317,7 @@ func resourceTable() *schema.Resource {
 				Type:          schema.TypeList,
 				Optional:      true,
 				MaxItems:      1,
-				ConflictsWith: []string{"restore_source_name"},
+				ConflictsWith: []string{"restore_source_name", "restore_source_table_arn"},
 				Elem: &schema.Resource{
 					Schema: map[string]*schema.Schema{
 						"input_compression_type": {
@@ -380,10 +388,16 @@ func resourceTable() *schema.Resource {
 				ForceNew:     true,
 				ValidateFunc: verify.ValidUTCTimestamp,
 			},
+			"restore_source_table_arn": {
+				Type:          schema.TypeString,
+				Optional:      true,
+				ValidateFunc:  verify.ValidARN,
+				ConflictsWith: []string{"import_table", "restore_source_name"},
+			},
 			"restore_source_name": {
 				Type:          schema.TypeString,
 				Optional:      true,
-				ConflictsWith: []string{"import_table"},
+				ConflictsWith: []string{"import_table", "restore_source_table_arn"},
 			},
 			"restore_to_latest_time": {
 				Type:     schema.TypeBool,
@@ -484,12 +498,22 @@ func resourceTableCreate(ctx context.Context, d *schema.ResourceData, meta inter
 		keySchemaMap["range_key"] = v.(string)
 	}
 
-	if v, ok := d.GetOk("restore_source_name"); ok {
+	sourceName, nameOk := d.GetOk("restore_source_name")
+	sourceArn, arnOk := d.GetOk("restore_source_table_arn")
+
+	if nameOk || arnOk {
 		input := &dynamodb.RestoreTableToPointInTimeInput{
-			SourceTableName: aws.String(v.(string)),
 			TargetTableName: aws.String(tableName),
 		}
 
+		if nameOk {
+			input.SourceTableName = aws.String(sourceName.(string))
+		}
+
+		if arnOk {
+			input.SourceTableArn = aws.String(sourceArn.(string))
+		}
+
 		if v, ok := d.GetOk("restore_date_time"); ok {
 			t, _ := time.Parse(time.RFC3339, v.(string))
 			input.RestoreDateTime = aws.Time(t)

internal/service/dynamodb/table_test.go

@@ -1678,6 +1678,46 @@ func TestAccDynamoDBTable_encryption(t *testing.T) {
 	})
 }
 
+func TestAccDynamoDBTable_restoreCrossRegion(t *testing.T) {
+	ctx := acctest.Context(t)
+	if testing.Short() {
+		t.Skip("skipping long-running test in short mode")
+	}
+
+	var conf awstypes.TableDescription
+	resourceName := "aws_dynamodb_table.test"
+	resourceNameRestore := "aws_dynamodb_table.test_restore"
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	rNameRestore := fmt.Sprintf("%s-restore", rName)
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck: func() {
+			acctest.PreCheck(ctx, t)
+			acctest.PreCheckMultipleRegion(t, 2)
+		},
+		ErrorCheck:               acctest.ErrorCheck(t, names.DynamoDBServiceID),
+		ProtoV5ProviderFactories: acctest.ProtoV5FactoriesMultipleRegions(ctx, t, 2),
+		CheckDestroy:             testAccCheckTableDestroy(ctx),
+		Steps: []resource.TestStep{
+			{
+				Config: testAccTableConfig_restoreCrossRegion(rName),
+				Check: resource.ComposeAggregateTestCheckFunc(
+					testAccCheckInitialTableExists(ctx, resourceName, &conf),
+					resource.TestCheckResourceAttr(resourceName, names.AttrName, rName),
+					resource.TestCheckResourceAttr(resourceNameRestore, names.AttrName, rNameRestore),
+					acctest.MatchResourceAttrRegionalARNRegion(resourceName, names.AttrARN, "dynamodb", acctest.Region(), regexache.MustCompile(`table/+.`)),
+					acctest.MatchResourceAttrRegionalARNRegion(resourceNameRestore, names.AttrARN, "dynamodb", acctest.AlternateRegion(), regexache.MustCompile(`table/+.`)),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+		},
+	})
+}
+
 func TestAccDynamoDBTable_Replica_multiple(t *testing.T) {
 	ctx := acctest.Context(t)
 	if testing.Short() {
@@ -4613,3 +4653,59 @@ resource "aws_dynamodb_table" "test" {
 }
 `, rName)
 }
+
+func testAccTableConfig_restoreCrossRegion(rName string) string {
+	return acctest.ConfigCompose(
+		acctest.ConfigMultipleRegionProvider(2),
+		fmt.Sprintf(`
+resource "aws_kms_key" "test" {
+  provider = "aws"
+
+  description             = %[1]q
+  deletion_window_in_days = 7
+}
+
+resource "aws_dynamodb_table" "test" {
+  provider = "aws"
+
+  name           = %[1]q
+  read_capacity  = 2
+  write_capacity = 2
+  hash_key       = "TestTableHashKey"
+
+  attribute {
+    name = "TestTableHashKey"
+    type = "S"
+  }
+
+  point_in_time_recovery {
+    enabled = true
+  }
+
+  server_side_encryption {
+    enabled     = true
+    kms_key_arn = aws_kms_key.test.arn
+  }
+}
+
+resource "aws_kms_key" "test_restore" {
+  provider = "awsalternate"
+
+  description             = "%[1]s-restore"
+  deletion_window_in_days = 7
+}
+
+resource "aws_dynamodb_table" "test_restore" {
+  provider = "awsalternate"
+
+  name                     = "%[1]s-restore"
+  restore_source_table_arn = aws_dynamodb_table.test.arn
+  restore_to_latest_time   = true
+
+  server_side_encryption {
+    enabled     = true
+    kms_key_arn = aws_kms_key.test_restore.arn
+  }
+}
+`, rName))
+}

website/docs/r/dynamodb_table.html.markdown

@@ -186,8 +186,9 @@ Optional arguments:
 * `replica` - (Optional) Configuration block(s) with [DynamoDB Global Tables V2 (version 2019.11.21)](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/globaltables.V2.html) replication configurations. See below.
 * `restore_date_time` - (Optional) Time of the point-in-time recovery point to restore.
 * `restore_source_name` - (Optional) Name of the table to restore. Must match the name of an existing table.
+* `restore_source_table_arn` - (Optional) ARN of the source table to restore. Must be supplied for cross-region restores.
 * `restore_to_latest_time` - (Optional) If set, restores table to the most recent point-in-time recovery point.
-* `server_side_encryption` - (Optional) Encryption at rest options. AWS DynamoDB tables are automatically encrypted at rest with an AWS-owned Customer Master Key if this argument isn't specified. See below.
+* `server_side_encryption` - (Optional) Encryption at rest options. AWS DynamoDB tables are automatically encrypted at rest with an AWS-owned Customer Master Key if this argument isn't specified. Must be supplied for cross-region restores. See below.
 * `stream_enabled` - (Optional) Whether Streams are enabled.
 * `stream_view_type` - (Optional) When an item in the table is modified, StreamViewType determines what information is written to the table's stream. Valid values are `KEYS_ONLY`, `NEW_IMAGE`, `OLD_IMAGE`, `NEW_AND_OLD_IMAGES`.
 * `table_class` - (Optional) Storage class of the table.