🐛 Source Dynamodb: Fix reserved words in expression (#20172)

itaseskii · vincentkoc · octavia-squidington-iii · web-flow · commit b9b8cb063858 · 2023-02-27T21:13:03.000Z
* fix reserved words in expression

* chore: bump version

* fix state message

* auto-bump connector version

* fix reserved words in projection expression

* bump dockerfile version

* auto-bump connector version

---------

Co-authored-by: Vincent Koc &lt;vincentkoc@ieee.org&gt;
Co-authored-by: Octavia Squidington III &lt;octavia-squidington-iii@users.noreply.github.com&gt;
Co-authored-by: Sajarin &lt;sajarindider@gmail.com&gt;
Co-authored-by: Sunny &lt;6833405+sh4sh@users.noreply.github.com&gt;
diff --git a/airbyte-config/init/src/main/resources/seed/source_definitions.yaml b/airbyte-config/init/src/main/resources/seed/source_definitions.yaml
@@ -474,7 +474,7 @@
 - name: DynamoDB
   sourceDefinitionId: 50401137-8871-4c5a-abb7-1f5fda35545a
   dockerRepository: airbyte/source-dynamodb
-  dockerImageTag: 0.1.1
+  dockerImageTag: 0.1.2
   documentationUrl: https://docs.airbyte.com/integrations/sources/dynamodb
   icon: dynamodb.svg
   sourceType: api
diff --git a/airbyte-config/init/src/main/resources/seed/source_specs.yaml b/airbyte-config/init/src/main/resources/seed/source_specs.yaml
@@ -3398,7 +3398,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-dynamodb:0.1.1"
+- dockerImage: "airbyte/source-dynamodb:0.1.2"
   spec:
     documentationUrl: "https://docs.airbyte.com/integrations/sources/dynamodb"
     connectionSpecification:
@@ -3464,6 +3464,13 @@
           airbyte_secret: true
           examples:
           - "a012345678910ABCDEFGH/AbCdEfGhEXAMPLEKEY"
+        reserved_attribute_names:
+          title: "Reserved attribute names"
+          type: "string"
+          description: "Comma separated reserved attribute names present in your tables"
+          airbyte_secret: true
+          examples:
+          - "name, field_name, field-name"
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
diff --git a/airbyte-integrations/connectors/source-dynamodb/Dockerfile b/airbyte-integrations/connectors/source-dynamodb/Dockerfile
@@ -17,5 +17,5 @@ ENV APPLICATION source-dynamodb
 COPY --from=build /airbyte /airbyte
 
 # Airbyte's build system uses these labels to know what to name and tag the docker images produced by this Dockerfile.
-LABEL io.airbyte.version=0.1.1
+LABEL io.airbyte.version=0.1.2
 LABEL io.airbyte.name=airbyte/source-dynamodb
diff --git a/airbyte-integrations/connectors/source-dynamodb/src/main/java/io/airbyte/integrations/source/dynamodb/DynamodbConfig.java b/airbyte-integrations/connectors/source-dynamodb/src/main/java/io/airbyte/integrations/source/dynamodb/DynamodbConfig.java
@@ -6,6 +6,8 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import java.net.URI;
+import java.util.Arrays;
+import java.util.List;
 import software.amazon.awssdk.regions.Region;
 
 public record DynamodbConfig(
@@ -16,18 +18,22 @@ public record DynamodbConfig(
 
                              String accessKey,
 
-                             String secretKey
+                             String secretKey,
+
+                             List<String> reservedAttributeNames
 
 ) {
 
   public static DynamodbConfig createDynamodbConfig(JsonNode jsonNode) {
     JsonNode endpoint = jsonNode.get("endpoint");
     JsonNode region = jsonNode.get("region");
+    JsonNode attributeNames = jsonNode.get("reserved_attribute_names");
     return new DynamodbConfig(
         endpoint != null && !endpoint.asText().isBlank() ? URI.create(endpoint.asText()) : null,
         region != null && !region.asText().isBlank() ? Region.of(region.asText()) : null,
         jsonNode.get("access_key_id").asText(),
-        jsonNode.get("secret_access_key").asText());
+        jsonNode.get("secret_access_key").asText(),
+        attributeNames != null ? Arrays.asList(attributeNames.asText().split("\\s*,\\s*")) : List.of());
   }
 
 }
diff --git a/airbyte-integrations/connectors/source-dynamodb/src/main/java/io/airbyte/integrations/source/dynamodb/DynamodbOperations.java b/airbyte-integrations/connectors/source-dynamodb/src/main/java/io/airbyte/integrations/source/dynamodb/DynamodbOperations.java
@@ -13,9 +13,11 @@
 import java.time.format.DateTimeParseException;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
 import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
 import software.amazon.awssdk.services.dynamodb.model.AttributeDefinition;
 import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
@@ -30,7 +32,10 @@ public class DynamodbOperations extends AbstractDatabase implements Closeable {
 
   private ObjectMapper schemaObjectMapper;
 
+  private DynamodbConfig dynamodbConfig;
+
   public DynamodbOperations(DynamodbConfig dynamodbConfig) {
+    this.dynamodbConfig = dynamodbConfig;
     this.dynamoDbClient = DynamodbUtils.createDynamoDbClient(dynamodbConfig);
     initMappers();
   }
@@ -105,12 +110,31 @@ public JsonNode inferSchema(String tableName, int sampleSize) {
   public List<JsonNode> scanTable(String tableName, Set<String> attributes, FilterAttribute filterAttribute) {
     List<JsonNode> items = new ArrayList<>();
 
-    var projectionAttributes = String.join(", ", attributes);
+    String prefix = "dyndb";
+    // remove and replace reserved attribute names
+    Set<String> copyAttributes = new HashSet<>(attributes);
+    dynamodbConfig.reservedAttributeNames().forEach(copyAttributes::remove);
+    dynamodbConfig.reservedAttributeNames().stream()
+        .filter(attributes::contains)
+        .map(str -> str.replaceAll("[-.]", ""))
+        .forEach(attr -> copyAttributes.add("#" + prefix + "_" + attr));
+
+    Map<String, String> mappingAttributes = dynamodbConfig.reservedAttributeNames().stream()
+        .filter(attributes::contains)
+        .collect(Collectors.toUnmodifiableMap(k -> "#" + prefix + "_" + k.replaceAll("[-.]", ""), k -> k));
+
+    var projectionAttributes = String.join(", ", copyAttributes);
+
 
     ScanRequest.Builder scanRequestBuilder = ScanRequest.builder()
         .tableName(tableName)
         .projectionExpression(projectionAttributes);
 
+    if (!mappingAttributes.isEmpty()) {
+      scanRequestBuilder
+          .expressionAttributeNames(mappingAttributes);
+    }
+
     if (filterAttribute != null && filterAttribute.name() != null &&
         filterAttribute.value() != null && filterAttribute.type() != null) {
 
@@ -134,8 +158,10 @@ public List<JsonNode> scanTable(String tableName, Set<String> attributes, Filter
         comparator = ">";
       }
 
+      String filterPlaceholder = dynamodbConfig.reservedAttributeNames().contains(filterName) ?
+          "#" + prefix + "_" + filterName.replaceAll("[-.]", "") : filterName;
       scanRequestBuilder
-          .filterExpression(filterName + " " + comparator + " :timestamp")
+          .filterExpression(filterPlaceholder + " " + comparator + " :timestamp")
           .expressionAttributeValues(Map.of(":timestamp", attributeValue));
 
     }
diff --git a/airbyte-integrations/connectors/source-dynamodb/src/main/java/io/airbyte/integrations/source/dynamodb/DynamodbUtils.java b/airbyte-integrations/connectors/source-dynamodb/src/main/java/io/airbyte/integrations/source/dynamodb/DynamodbUtils.java
@@ -86,9 +86,9 @@ private static AirbyteStateMessage convertStateMessage(final io.airbyte.protocol
 
   record StreamState(
 
-                     AirbyteStateMessage.AirbyteStateType airbyteStateType,
+      AirbyteStateMessage.AirbyteStateType airbyteStateType,
 
-                     List<AirbyteStateMessage> airbyteStateMessages) {
+      List<AirbyteStateMessage> airbyteStateMessages) {
 
   }
 
diff --git a/airbyte-integrations/connectors/source-dynamodb/src/main/resources/spec.json b/airbyte-integrations/connectors/source-dynamodb/src/main/resources/spec.json
@@ -61,6 +61,13 @@
         "description": "The corresponding secret to the access key id.",
         "airbyte_secret": true,
         "examples": ["a012345678910ABCDEFGH/AbCdEfGhEXAMPLEKEY"]
+      },
+      "reserved_attribute_names": {
+        "title": "Reserved attribute names",
+        "type": "string",
+        "description": "Comma separated reserved attribute names present in your tables",
+        "airbyte_secret": true,
+        "examples": ["name, field_name, field-name"]
       }
     }
   }
diff --git a/airbyte-integrations/connectors/source-dynamodb/src/test-integration/java/io/airbyte/integrations/source/dynamodb/DynamodbDataFactory.java b/airbyte-integrations/connectors/source-dynamodb/src/test-integration/java/io/airbyte/integrations/source/dynamodb/DynamodbDataFactory.java
@@ -80,6 +80,7 @@ public static JsonNode createJsonConfig(DynamodbContainer dynamodbContainer) {
         .put("region", dynamodbContainer.getRegion())
         .put("access_key_id", dynamodbContainer.getAccessKey())
         .put("secret_access_key", dynamodbContainer.getSecretKey())
+        .put("reserved_attribute_names", "name, field.name, field-name")
         .build());
   }
 
diff --git a/airbyte-integrations/connectors/source-dynamodb/src/test-integration/java/io/airbyte/integrations/source/dynamodb/DynamodbOperationsTest.java b/airbyte-integrations/connectors/source-dynamodb/src/test-integration/java/io/airbyte/integrations/source/dynamodb/DynamodbOperationsTest.java
@@ -153,29 +153,29 @@ void testScanTable() throws JsonProcessingException, JSONException {
     PutItemRequest putItemRequest1 = DynamodbDataFactory.putItemRequest(tableName, Map.of(
         "attr_1", AttributeValue.builder().s("str_4").build(),
         "attr_2", AttributeValue.builder().s("str_5").build(),
-        "attr_3", AttributeValue.builder().s("2017-12-21T17:42:34Z").build(),
+        "name", AttributeValue.builder().s("2017-12-21T17:42:34Z").build(),
         "attr_4", AttributeValue.builder().ns("12.5", "74.5").build()));
 
     dynamoDbClient.putItem(putItemRequest1);
 
     PutItemRequest putItemRequest2 = DynamodbDataFactory.putItemRequest(tableName, Map.of(
         "attr_1", AttributeValue.builder().s("str_6").build(),
         "attr_2", AttributeValue.builder().s("str_7").build(),
-        "attr_3", AttributeValue.builder().s("2019-12-21T17:42:34Z").build(),
+        "name", AttributeValue.builder().s("2019-12-21T17:42:34Z").build(),
         "attr_6", AttributeValue.builder().ss("str_1", "str_2").build()));
 
     dynamoDbClient.putItem(putItemRequest2);
 
-    var response = dynamodbOperations.scanTable(tableName, Set.of("attr_1", "attr_2", "attr_3"),
-        new DynamodbOperations.FilterAttribute("attr_3", "2018-12-21T17:42:34Z",
+    var response = dynamodbOperations.scanTable(tableName, Set.of("attr_1", "attr_2", "name"),
+        new DynamodbOperations.FilterAttribute("name", "2018-12-21T17:42:34Z",
             DynamodbOperations.FilterAttribute.FilterType.S));
 
     assertThat(response)
         .hasSize(1);
 
     JSONAssert.assertEquals(objectMapper.writeValueAsString(response.get(0)), """
                                                                               {
-                                                                              	"attr_3": "2019-12-21T17:42:34Z",
+                                                                              	"name": "2019-12-21T17:42:34Z",
                                                                               	"attr_2": "str_7",
                                                                               	"attr_1": "str_6"
                                                                               }
diff --git a/docs/integrations/sources/dynamodb.md b/docs/integrations/sources/dynamodb.md
@@ -51,15 +51,18 @@ This guide describes in details how you can configure the connector to connect w
 
 ### Сonfiguration Parameters
 
-* endpoint: aws endpoint of the dynamodb instance
-* region: the region code of the dynamodb instance
-* access_key_id: the access key for the IAM user with the required permissions
-* secret_access_key: the secret key for the IAM user with the required permissions
-
+* **_endpoint_**: aws endpoint of the dynamodb instance
+* **_region_**: the region code of the dynamodb instance
+* **_access_key_id_**: the access key for the IAM user with the required permissions
+* **_secret_access_key_**: the secret key for the IAM user with the required permissions
+* **_reserved_attribute_names_**: comma separated list of attribute names present in the replication tables which contain reserved words or special characters. https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Expressions.ExpressionAttributeNames.html
 
 ## Changelog
 
-| Version | Date       | Pull Request | Subject         |
-|:--------|:-----------|:-------------|:----------------|
-| 0.1.1   | 02-09-2023 | https://github.com/airbytehq/airbyte/pull/22682             | Fix build     |
-| 0.1.0   | 11-14-2022 | https://github.com/airbytehq/airbyte/pull/18750             | Initial version |
+
+| Version | Date       | Pull Request                                    | Subject                                                              |
+|:--------|:-----------|:------------------------------------------------|:---------------------------------------------------------------------|
+| 0.1.2   | 01-19-2023 | https://github.com/airbytehq/airbyte/pull/20172 | Fix reserved words in projection expression & make them configurable |
+| 0.1.1   | 02-09-2023 | https://github.com/airbytehq/airbyte/pull/22682 | Fix build                                                            |
+| 0.1.0   | 11-14-2022 | https://github.com/airbytehq/airbyte/pull/18750 | Initial version                                                      |
+

Original file line number	Diff line number	Diff line change
`@@ -86,9 +86,9 @@ private static AirbyteStateMessage convertStateMessage(final io.airbyte.protocol`
`86`	`86`
`87`	`87`	`record StreamState(`
`88`	`88`
`89`		`- AirbyteStateMessage.AirbyteStateType airbyteStateType,`
	`89`	`+ AirbyteStateMessage.AirbyteStateType airbyteStateType,`
`90`	`90`
`91`		`- List<AirbyteStateMessage> airbyteStateMessages) {`
	`91`	`+ List<AirbyteStateMessage> airbyteStateMessages) {`
`92`	`92`
`93`	`93`	`}`
`94`	`94`
Original file line number	Diff line number	Diff line change
`@@ -61,6 +61,13 @@`
`61`	`61`	`"description": "The corresponding secret to the access key id.",`
`62`	`62`	`"airbyte_secret": true,`
`63`	`63`	`"examples": ["a012345678910ABCDEFGH/AbCdEfGhEXAMPLEKEY"]`
	`64`	`+ },`
	`65`	`+ "reserved_attribute_names": {`
	`66`	`+ "title": "Reserved attribute names",`
	`67`	`+ "type": "string",`
	`68`	`+ "description": "Comma separated reserved attribute names present in your tables",`
	`69`	`+ "airbyte_secret": true,`
	`70`	`+ "examples": ["name, field_name, field-name"]`
`64`	`71`	`}`
`65`	`72`	`}`
`66`	`73`	`}`
Original file line number	Diff line number	Diff line change
`@@ -80,6 +80,7 @@ public static JsonNode createJsonConfig(DynamodbContainer dynamodbContainer) {`
`80`	`80`	`.put("region", dynamodbContainer.getRegion())`
`81`	`81`	`.put("access_key_id", dynamodbContainer.getAccessKey())`
`82`	`82`	`.put("secret_access_key", dynamodbContainer.getSecretKey())`
	`83`	`+ .put("reserved_attribute_names", "name, field.name, field-name")`
`83`	`84`	`.build());`
`84`	`85`	`}`
`85`	`86`