r/glue_crawler - support delta tables

.changelog/22472.txt

@@ -0,0 +1,3 @@
+```release-note:enhancement
+resource/aws_glue_crawler: add `delta_target` argument.
+```

internal/service/glue/crawler.go

@@ -112,7 +112,7 @@ func ResourceCrawler() *schema.Resource {
 				Type:         schema.TypeList,
 				Optional:     true,
 				MinItems:     1,
-				AtLeastOneOf: []string{"s3_target", "dynamodb_target", "mongodb_target", "jdbc_target", "catalog_target"},
+				AtLeastOneOf: []string{"s3_target", "dynamodb_target", "mongodb_target", "jdbc_target", "catalog_target", "delta_target"},
 				Elem: &schema.Resource{
 					Schema: map[string]*schema.Schema{
 						"connection_name": {
@@ -150,7 +150,7 @@ func ResourceCrawler() *schema.Resource {
 				Type:         schema.TypeList,
 				Optional:     true,
 				MinItems:     1,
-				AtLeastOneOf: []string{"s3_target", "dynamodb_target", "mongodb_target", "jdbc_target", "catalog_target"},
+				AtLeastOneOf: []string{"s3_target", "dynamodb_target", "mongodb_target", "jdbc_target", "catalog_target", "delta_target"},
 				Elem: &schema.Resource{
 					Schema: map[string]*schema.Schema{
 						"path": {
@@ -174,7 +174,7 @@ func ResourceCrawler() *schema.Resource {
 				Type:         schema.TypeList,
 				Optional:     true,
 				MinItems:     1,
-				AtLeastOneOf: []string{"s3_target", "dynamodb_target", "mongodb_target", "jdbc_target", "catalog_target"},
+				AtLeastOneOf: []string{"s3_target", "dynamodb_target", "mongodb_target", "jdbc_target", "catalog_target", "delta_target"},
 				Elem: &schema.Resource{
 					Schema: map[string]*schema.Schema{
 						"connection_name": {
@@ -197,7 +197,7 @@ func ResourceCrawler() *schema.Resource {
 				Type:         schema.TypeList,
 				Optional:     true,
 				MinItems:     1,
-				AtLeastOneOf: []string{"s3_target", "dynamodb_target", "mongodb_target", "jdbc_target", "catalog_target"},
+				AtLeastOneOf: []string{"s3_target", "dynamodb_target", "mongodb_target", "jdbc_target", "catalog_target", "delta_target"},
 				Elem: &schema.Resource{
 					Schema: map[string]*schema.Schema{
 						"connection_name": {
@@ -216,11 +216,34 @@ func ResourceCrawler() *schema.Resource {
 					},
 				},
 			},
+			"delta_target": {
+				Type:         schema.TypeList,
+				Optional:     true,
+				MinItems:     1,
+				AtLeastOneOf: []string{"s3_target", "dynamodb_target", "mongodb_target", "jdbc_target", "catalog_target", "delta_target"},
+				Elem: &schema.Resource{
+					Schema: map[string]*schema.Schema{
+						"connection_name": {
+							Type:     schema.TypeString,
+							Required: true,
+						},
+						"delta_tables": {
+							Type:     schema.TypeSet,
+							Required: true,
+							Elem:     &schema.Schema{Type: schema.TypeString},
+						},
+						"write_manifest": {
+							Type:     schema.TypeBool,
+							Required: true,
+						},
+					},
+				},
+			},
 			"catalog_target": {
 				Type:         schema.TypeList,
 				Optional:     true,
 				MinItems:     1,
-				AtLeastOneOf: []string{"s3_target", "dynamodb_target", "mongodb_target", "jdbc_target", "catalog_target"},
+				AtLeastOneOf: []string{"s3_target", "dynamodb_target", "mongodb_target", "jdbc_target", "catalog_target", "delta_target"},
 				Elem: &schema.Resource{
 					Schema: map[string]*schema.Schema{
 						"database_name": {
@@ -480,6 +503,10 @@ func expandGlueCrawlerTargets(d *schema.ResourceData) *glue.CrawlerTargets {
 		crawlerTargets.MongoDBTargets = expandGlueMongoDBTargets(v.([]interface{}))
 	}
 
+	if v, ok := d.GetOk("delta_target"); ok {
+		crawlerTargets.DeltaTargets = expandGlueDeltaTargets(v.([]interface{}))
+	}
+
 	return crawlerTargets
 }
 
@@ -620,6 +647,29 @@ func expandGlueMongoDBTarget(cfg map[string]interface{}) *glue.MongoDBTarget {
 	return target
 }
 
+func expandGlueDeltaTargets(targets []interface{}) []*glue.DeltaTarget {
+	if len(targets) < 1 {
+		return []*glue.DeltaTarget{}
+	}
+
+	perms := make([]*glue.DeltaTarget, len(targets))
+	for i, rawCfg := range targets {
+		cfg := rawCfg.(map[string]interface{})
+		perms[i] = expandGlueDeltaTarget(cfg)
+	}
+	return perms
+}
+
+func expandGlueDeltaTarget(cfg map[string]interface{}) *glue.DeltaTarget {
+	target := &glue.DeltaTarget{
+		ConnectionName: aws.String(cfg["connection_name"].(string)),
+		DeltaTables:    flex.ExpandStringSet(cfg["delta_tables"].(*schema.Set)),
+		WriteManifest:  aws.Bool(cfg["write_manifest"].(bool)),
+	}
+
+	return target
+}
+
 func resourceCrawlerUpdate(d *schema.ResourceData, meta interface{}) error {
 	glueConn := meta.(*conns.AWSClient).GlueConn
 	name := d.Get("name").(string)
@@ -753,6 +803,10 @@ func resourceCrawlerRead(d *schema.ResourceData, meta interface{}) error {
 		if err := d.Set("mongodb_target", flattenGlueMongoDBTargets(crawler.Targets.MongoDBTargets)); err != nil {
 			return fmt.Errorf("error setting mongodb_target: %w", err)
 		}
+
+		if err := d.Set("delta_target", flattenGlueDeltaTargets(crawler.Targets.DeltaTargets)); err != nil {
+			return fmt.Errorf("error setting delta_target: %w", err)
+		}
 	}
 
 	tags, err := ListTags(glueConn, crawlerARN)
@@ -859,6 +913,20 @@ func flattenGlueMongoDBTargets(mongoDBTargets []*glue.MongoDBTarget) []map[strin
 	return result
 }
 
+func flattenGlueDeltaTargets(deltaTargets []*glue.DeltaTarget) []map[string]interface{} {
+	result := make([]map[string]interface{}, 0)
+
+	for _, deltaTarget := range deltaTargets {
+		attrs := make(map[string]interface{})
+		attrs["connection_name"] = aws.StringValue(deltaTarget.ConnectionName)
+		attrs["delta_tables"] = flex.FlattenStringSet(deltaTarget.DeltaTables)
+		attrs["write_manifest"] = aws.BoolValue(deltaTarget.WriteManifest)
+
+		result = append(result, attrs)
+	}
+	return result
+}
+
 func resourceCrawlerDelete(d *schema.ResourceData, meta interface{}) error {
 	glueConn := meta.(*conns.AWSClient).GlueConn
 

internal/service/glue/crawler_test.go

@@ -503,6 +503,50 @@ func TestAccGlueCrawler_MongoDBTarget_multiple(t *testing.T) {
 	})
 }
 
+func TestAccGlueCrawler_deltaTarget(t *testing.T) {
+	var crawler glue.Crawler
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	resourceName := "aws_glue_crawler.test"
+
+	connectionUrl := fmt.Sprintf("mongodb://%s:27017/testdatabase", acctest.RandomDomainName())
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { acctest.PreCheck(t) },
+		ErrorCheck:   acctest.ErrorCheck(t, glue.EndpointsID),
+		Providers:    acctest.Providers,
+		CheckDestroy: testAccCheckCrawlerDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccGlueCrawlerConfigDeltaTarget(rName, connectionUrl, "s3://table1"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckCrawlerExists(resourceName, &crawler),
+					resource.TestCheckResourceAttr(resourceName, "delta_target.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "delta_target.0.connection_name", rName),
+					resource.TestCheckResourceAttr(resourceName, "delta_target.0.delta_tables.#", "1"),
+					resource.TestCheckTypeSetElemAttr(resourceName, "delta_target.0.delta_tables.*", "s3://table1"),
+					resource.TestCheckResourceAttr(resourceName, "delta_target.0.write_manifest", "false"),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+			{
+				Config: testAccGlueCrawlerConfigDeltaTarget(rName, connectionUrl, "s3://table2"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckCrawlerExists(resourceName, &crawler),
+					resource.TestCheckResourceAttr(resourceName, "delta_target.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "delta_target.0.connection_name", rName),
+					resource.TestCheckResourceAttr(resourceName, "delta_target.0.delta_tables.#", "1"),
+					resource.TestCheckTypeSetElemAttr(resourceName, "delta_target.0.delta_tables.*", "s3://table2"),
+					resource.TestCheckResourceAttr(resourceName, "delta_target.0.write_manifest", "false"),
+				),
+			},
+		},
+	})
+}
+
 func TestAccGlueCrawler_s3Target(t *testing.T) {
 	var crawler glue.Crawler
 	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
@@ -2718,6 +2762,39 @@ resource "aws_glue_crawler" "test" {
 `, rName, connectionUrl, path1, path2)
 }
 
+func testAccGlueCrawlerConfigDeltaTarget(rName, connectionUrl, tableName string) string {
+	return testAccGlueCrawlerConfig_Base(rName) + fmt.Sprintf(`
+resource "aws_glue_catalog_database" "test" {
+  name = %[1]q
+}
+
+resource "aws_glue_connection" "test" {
+  name            = %[1]q
+  connection_type = "MONGODB"
+
+  connection_properties = {
+    CONNECTION_URL = %[2]q
+    PASSWORD       = "testpassword"
+    USERNAME       = "testusername"
+  }
+}
+
+resource "aws_glue_crawler" "test" {
+  depends_on = [aws_iam_role_policy_attachment.test-AWSGlueServiceRole]
+
+  database_name = aws_glue_catalog_database.test.name
+  name          = %[1]q
+  role          = aws_iam_role.test.name
+
+  delta_target {
+    connection_name = aws_glue_connection.test.name
+    delta_tables    = [%[3]q]
+    write_manifest  = false
+  }
+}
+`, rName, connectionUrl, tableName)
+}
+
 func testAccGlueCrawlerLineageConfig(rName, lineageConfig string) string {
 	return testAccGlueCrawlerConfig_Base(rName) + fmt.Sprintf(`
 resource "aws_glue_catalog_database" "test" {

website/docs/r/glue_crawler.html.markdown

@@ -188,6 +188,12 @@ The following arguments are supported:
 * `path` - (Required) The path of the Amazon DocumentDB or MongoDB target (database/collection).
 * `scan_all` - (Optional) Indicates whether to scan all the records, or to sample rows from the table. Scanning all the records can take a long time when the table is not a high throughput table. Default value is `true`.
 
+### Delta Target
+
+* `connection_name` - (Required) The name of the connection to use to connect to the Delta table target.
+* `delta_tables` - (Required) A list of the Amazon S3 paths to the Delta tables.
+* `write_manifest` - (Required) Specifies whether to write the manifest files to the Delta table path.
+
 ### Schema Change Policy
 
 * `delete_behavior` - (Optional) The deletion behavior when the crawler finds a deleted object. Valid values: `LOG`, `DELETE_FROM_DATABASE`, or `DEPRECATE_IN_DATABASE`. Defaults to `DEPRECATE_IN_DATABASE`.