This is an automated email from the ASF dual-hosted git repository.
ferenc-csaky pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-aws.git
The following commit(s) were added to refs/heads/main by this push:
new 2fc634d5 [FLINK-35500] Build DELETE key from primary key for CDC
deletes
2fc634d5 is described below
commit 2fc634d5df4a0b09b45d51b51f488072b97d4936
Author: riyarawat-amazon <[email protected]>
AuthorDate: Mon Aug 31 21:40:19 2026 +0530
[FLINK-35500] Build DELETE key from primary key for CDC deletes
---
docs/content/docs/connectors/table/dynamodb.md | 47 ++++++++
.../dynamodb/table/DynamoDbDynamicSink.java | 21 +++-
.../dynamodb/table/DynamoDbDynamicSinkFactory.java | 61 +++++++++-
.../dynamodb/table/RowDataElementConverter.java | 39 +++++--
.../table/RowDataToAttributeValueConverter.java | 48 +++++++-
.../table/DynamoDbDynamicSinkFactoryTest.java | 127 ++++++++++++++++++++
.../table/RowDataElementConverterTest.java | 60 +++++++++-
.../RowDataToAttributeValueConverterTest.java | 129 +++++++++++++++++++++
8 files changed, 510 insertions(+), 22 deletions(-)
diff --git a/docs/content/docs/connectors/table/dynamodb.md
b/docs/content/docs/connectors/table/dynamodb.md
index 66fd1078..130fba79 100644
--- a/docs/content/docs/connectors/table/dynamodb.md
+++ b/docs/content/docs/connectors/table/dynamodb.md
@@ -310,6 +310,53 @@ WITH (
);
```
+## Primary Key and CDC (Changelog) Streams
+
+When the DynamoDB sink consumes a changelog (CDC) stream that contains `DELETE`
+records (for example, from a Debezium source), you must declare a `PRIMARY KEY`
+on the table. DynamoDB requires a delete operation to be issued with only the
+table's key (its partition key and, if present, its sort key). The declared
+`PRIMARY KEY` tells the connector which columns form that key, so it can build
a
+valid delete request instead of sending the whole row. The columns are used in
+the order they are declared: the first column is the partition key and the
+optional second column is the sort key.
+
+```sql
+CREATE TABLE DynamoDbTable (
+ `user_id` BIGINT,
+ `item_id` BIGINT,
+ `category_id` BIGINT,
+ `behavior` STRING,
+ PRIMARY KEY (user_id) NOT ENFORCED
+) PARTITIONED BY ( user_id )
+WITH (
+ 'connector' = 'dynamodb',
+ 'table-name' = 'user_behavior',
+ 'aws.region' = 'us-east-2'
+);
+```
+
+If a `DELETE` record is received and no `PRIMARY KEY` has been declared, the
sink
+fails with a clear error asking you to declare one. `INSERT` and `UPDATE_AFTER`
+records are unaffected by the `PRIMARY KEY` declaration and always write the
full
+item.
+
+The `PRIMARY KEY` must have at most two columns, matching a DynamoDB key
schema:
+the first column is the partition key and the optional second column is the
sort
+key. Declaring more than two columns fails table creation. The `NOT ENFORCED`
+qualifier is required because Flink does not own the data and therefore cannot
+enforce key uniqueness; it only uses the declaration as metadata.
+
+Note that `PRIMARY KEY` is distinct from the `PARTITIONED BY` clause described
in
+[Sink Partitioning](#sink-partitioning): `PARTITIONED BY` controls client-side
+deduplication of records within a batch, while `PRIMARY KEY` identifies the
+DynamoDB table key used to build delete requests. When `PRIMARY KEY` is
declared
+but `PARTITIONED BY` is not, the sink automatically uses the primary key for
+client-side deduplication. This is required for changelog streams: a single
batch
+may contain both an upsert and a delete for the same key, and without
+deduplication DynamoDB rejects the batch as containing duplicate keys. If you
+specify both clauses, they should normally list the same columns.
+
## Notice
The current implementation of the DynamoDB SQL connector is write-only and
doesn't provide an implementation for source queries.
diff --git
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSink.java
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSink.java
index 8091306f..5089c293 100644
---
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSink.java
+++
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSink.java
@@ -34,6 +34,7 @@ import org.apache.flink.table.types.DataType;
import javax.annotation.Nullable;
import java.util.ArrayList;
+import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
@@ -52,6 +53,7 @@ public class DynamoDbDynamicSink extends
AsyncDynamicTableSink<DynamoDbWriteRequ
private final Properties dynamoDbClientProperties;
private final DataType physicalDataType;
private final Set<String> overwriteByPartitionKeys;
+ private final List<String> primaryKey;
protected DynamoDbDynamicSink(
@Nullable Integer maxBatchSize,
@@ -64,7 +66,8 @@ public class DynamoDbDynamicSink extends
AsyncDynamicTableSink<DynamoDbWriteRequ
boolean ignoreNulls,
Properties dynamoDbClientProperties,
DataType physicalDataType,
- Set<String> overwriteByPartitionKeys) {
+ Set<String> overwriteByPartitionKeys,
+ List<String> primaryKey) {
super(
maxBatchSize,
maxInFlightRequests,
@@ -77,6 +80,7 @@ public class DynamoDbDynamicSink extends
AsyncDynamicTableSink<DynamoDbWriteRequ
this.dynamoDbClientProperties = dynamoDbClientProperties;
this.physicalDataType = physicalDataType;
this.overwriteByPartitionKeys = overwriteByPartitionKeys;
+ this.primaryKey = primaryKey;
}
@Override
@@ -93,7 +97,8 @@ public class DynamoDbDynamicSink extends
AsyncDynamicTableSink<DynamoDbWriteRequ
.setOverwriteByPartitionKeys(new
ArrayList<>(overwriteByPartitionKeys))
.setDynamoDbProperties(dynamoDbClientProperties)
.setElementConverter(
- new RowDataElementConverter(physicalDataType,
ignoreNulls));
+ new RowDataElementConverter(
+ physicalDataType, primaryKey,
ignoreNulls));
addAsyncOptionsToSinkBuilder(builder);
@@ -113,7 +118,8 @@ public class DynamoDbDynamicSink extends
AsyncDynamicTableSink<DynamoDbWriteRequ
ignoreNulls,
dynamoDbClientProperties,
physicalDataType,
- overwriteByPartitionKeys);
+ overwriteByPartitionKeys,
+ primaryKey);
}
@Override
@@ -142,6 +148,7 @@ public class DynamoDbDynamicSink extends
AsyncDynamicTableSink<DynamoDbWriteRequ
private Properties dynamoDbClientProperties;
private DataType physicalDataType;
private Set<String> overwriteByPartitionKeys;
+ private List<String> primaryKey;
public DynamoDbDynamicTableSinkBuilder setTableName(String tableName) {
this.tableName = tableName;
@@ -175,6 +182,11 @@ public class DynamoDbDynamicSink extends
AsyncDynamicTableSink<DynamoDbWriteRequ
return this;
}
+ public DynamoDbDynamicTableSinkBuilder setPrimaryKey(List<String>
primaryKey) {
+ this.primaryKey = primaryKey;
+ return this;
+ }
+
@Override
public AsyncDynamicTableSink<DynamoDbWriteRequest> build() {
return new DynamoDbDynamicSink(
@@ -188,7 +200,8 @@ public class DynamoDbDynamicSink extends
AsyncDynamicTableSink<DynamoDbWriteRequ
ignoreNulls,
dynamoDbClientProperties,
physicalDataType,
- overwriteByPartitionKeys);
+ overwriteByPartitionKeys,
+ primaryKey);
}
}
}
diff --git
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java
index 8a1033da..1ad0f3ff 100644
---
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java
+++
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java
@@ -21,11 +21,14 @@ package org.apache.flink.connector.dynamodb.table;
import org.apache.flink.annotation.Internal;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.connector.base.table.AsyncDynamicTableSinkFactory;
+import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.UniqueConstraint;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.factories.FactoryUtil;
import java.util.HashSet;
+import java.util.List;
import java.util.Set;
import static
org.apache.flink.connector.dynamodb.table.DynamoDbConnectorOptions.AWS_REGION;
@@ -48,6 +51,10 @@ public class DynamoDbDynamicSinkFactory extends
AsyncDynamicTableSinkFactory {
DynamoDbConfiguration dynamoDbConfiguration =
new DynamoDbConfiguration(catalogTable.getOptions(),
factoryHelper.getOptions());
+ List<String> primaryKey = getValidatedPrimaryKey(catalogTable);
+ Set<String> overwriteByPartitionKeys =
+ getValidatedOverwriteByPartitionKeys(catalogTable, primaryKey);
+
DynamoDbDynamicSink.DynamoDbDynamicTableSinkBuilder builder =
DynamoDbDynamicSink.builder()
.setTableName(dynamoDbConfiguration.getTableName())
@@ -55,7 +62,8 @@ public class DynamoDbDynamicSinkFactory extends
AsyncDynamicTableSinkFactory {
.setIgnoreNulls(dynamoDbConfiguration.getIgnoreNulls())
.setPhysicalDataType(
catalogTable.getResolvedSchema().toPhysicalRowDataType())
- .setOverwriteByPartitionKeys(new
HashSet<>(catalogTable.getPartitionKeys()))
+ .setOverwriteByPartitionKeys(overwriteByPartitionKeys)
+ .setPrimaryKey(primaryKey)
.setDynamoDbClientProperties(
dynamoDbConfiguration.getSinkClientProperties());
@@ -64,6 +72,57 @@ public class DynamoDbDynamicSinkFactory extends
AsyncDynamicTableSinkFactory {
return builder.build();
}
+ /**
+ * Returns the declared PRIMARY KEY columns in declaration order
(partition key first, optional
+ * sort key second), validating that at most two columns are declared.
+ */
+ private static List<String> getValidatedPrimaryKey(ResolvedCatalogTable
catalogTable) {
+ List<String> primaryKey =
+ catalogTable
+ .getResolvedSchema()
+ .getPrimaryKey()
+ .map(UniqueConstraint::getColumns)
+ .orElse(List.of());
+
+ if (primaryKey.size() > 2) {
+ throw new ValidationException(
+ String.format(
+ "The DynamoDB sink supports a PRIMARY KEY of at
most two columns (a "
+ + "partition key and an optional sort
key), but %d columns were "
+ + "declared: %s. Please declare a PRIMARY
KEY that matches the "
+ + "DynamoDB table's key schema.",
+ primaryKey.size(), primaryKey));
+ }
+ return primaryKey;
+ }
+
+ /**
+ * Returns the partition keys used for client-side deduplication,
defaulting to the primary key
+ * when PARTITIONED BY is not specified. When both are declared they must
match; otherwise a CDC
+ * batch could keep an upsert and a delete that map to the same DynamoDB
key, which DynamoDB
+ * rejects as duplicates.
+ */
+ private static Set<String> getValidatedOverwriteByPartitionKeys(
+ ResolvedCatalogTable catalogTable, List<String> primaryKey) {
+ List<String> declaredPartitionKeys = catalogTable.getPartitionKeys();
+
+ if (!declaredPartitionKeys.isEmpty()
+ && !primaryKey.isEmpty()
+ && !new HashSet<>(declaredPartitionKeys).equals(new
HashSet<>(primaryKey))) {
+ throw new ValidationException(
+ String.format(
+ "When both PARTITIONED BY and PRIMARY KEY are
specified for a DynamoDB "
+ + "table they must reference the same
columns, but PARTITIONED "
+ + "BY was %s and PRIMARY KEY was %s.
Either align them or "
+ + "specify only the PRIMARY KEY.",
+ declaredPartitionKeys, primaryKey));
+ }
+
+ return declaredPartitionKeys.isEmpty()
+ ? new HashSet<>(primaryKey)
+ : new HashSet<>(declaredPartitionKeys);
+ }
+
@Override
public String factoryIdentifier() {
return FACTORY_IDENTIFIER;
diff --git
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataElementConverter.java
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataElementConverter.java
index d156dbcc..6e545fa0 100644
---
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataElementConverter.java
+++
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataElementConverter.java
@@ -27,6 +27,8 @@ import org.apache.flink.table.api.TableException;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;
+import java.util.List;
+
/**
* Implementation of an {@link ElementConverter} for the DynamoDb Table sink.
The element converter
* maps the Flink internal type of {@link RowData} to a {@link
DynamoDbWriteRequest} to be used by
@@ -37,39 +39,54 @@ public class RowDataElementConverter implements
ElementConverter<RowData, Dynamo
private boolean ignoreNulls = false;
private final DataType physicalDataType;
+ private final List<String> primaryKey;
private transient RowDataToAttributeValueConverter
rowDataToAttributeValueConverter;
public RowDataElementConverter(DataType physicalDataType) {
- this.physicalDataType = physicalDataType;
- this.rowDataToAttributeValueConverter =
- new RowDataToAttributeValueConverter(physicalDataType);
+ this(physicalDataType, List.of(), false);
}
public RowDataElementConverter(DataType physicalDataType, boolean
ignoreNulls) {
- this.ignoreNulls = ignoreNulls;
+ this(physicalDataType, List.of(), ignoreNulls);
+ }
+
+ public RowDataElementConverter(DataType physicalDataType, List<String>
primaryKey) {
+ this(physicalDataType, primaryKey, false);
+ }
+
+ public RowDataElementConverter(
+ DataType physicalDataType, List<String> primaryKey, boolean
ignoreNulls) {
this.physicalDataType = physicalDataType;
+ this.primaryKey = primaryKey;
+ this.ignoreNulls = ignoreNulls;
this.rowDataToAttributeValueConverter =
- new RowDataToAttributeValueConverter(physicalDataType,
ignoreNulls);
+ new RowDataToAttributeValueConverter(physicalDataType,
primaryKey, ignoreNulls);
}
@Override
public DynamoDbWriteRequest apply(RowData element, SinkWriter.Context
context) {
if (rowDataToAttributeValueConverter == null) {
rowDataToAttributeValueConverter =
- new RowDataToAttributeValueConverter(physicalDataType,
ignoreNulls);
+ new RowDataToAttributeValueConverter(physicalDataType,
primaryKey, ignoreNulls);
}
- DynamoDbWriteRequest.Builder builder =
- DynamoDbWriteRequest.builder()
-
.setItem(rowDataToAttributeValueConverter.convertRowData(element));
+ DynamoDbWriteRequest.Builder builder = DynamoDbWriteRequest.builder();
switch (element.getRowKind()) {
case INSERT:
case UPDATE_AFTER:
- builder.setType(DynamoDbWriteRequestType.PUT);
+ builder.setType(DynamoDbWriteRequestType.PUT)
+
.setItem(rowDataToAttributeValueConverter.convertRowData(element));
break;
case DELETE:
- builder.setType(DynamoDbWriteRequestType.DELETE);
+ if (primaryKey.isEmpty()) {
+ throw new TableException(
+ "Cannot process a DELETE record because no PRIMARY
KEY is defined on "
+ + "the DynamoDB table. Please declare a
PRIMARY KEY on the "
+ + "table to support deletes from a
changelog (CDC) stream.");
+ }
+ builder.setType(DynamoDbWriteRequestType.DELETE)
+
.setItem(rowDataToAttributeValueConverter.convertRowDataToKey(element));
break;
case UPDATE_BEFORE:
default:
diff --git
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverter.java
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverter.java
index 08d62348..c85e3740 100644
---
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverter.java
+++
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverter.java
@@ -38,6 +38,7 @@ import software.amazon.awssdk.enhanced.dynamodb.TableSchema;
import software.amazon.awssdk.enhanced.dynamodb.mapper.StaticTableSchema;
import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -51,15 +52,32 @@ public class RowDataToAttributeValueConverter {
private final DataType physicalDataType;
private final TableSchema<RowData> tableSchema;
+
+ /**
+ * Ordered primary key attribute names. Following DynamoDB's primary key
definition, the first
+ * element is the partition key and the optional second element is the
sort key. Used to build
+ * the key of a {@code DeleteRequest}, which must contain only the primary
key attributes.
+ */
+ private final List<String> primaryKey;
+
private boolean ignoreNulls = false;
public RowDataToAttributeValueConverter(DataType physicalDataType) {
- this.physicalDataType = physicalDataType;
- this.tableSchema = createTableSchema();
+ this(physicalDataType, List.of(), false);
}
public RowDataToAttributeValueConverter(DataType physicalDataType, boolean
ignoreNulls) {
+ this(physicalDataType, List.of(), ignoreNulls);
+ }
+
+ public RowDataToAttributeValueConverter(DataType physicalDataType,
List<String> primaryKey) {
+ this(physicalDataType, primaryKey, false);
+ }
+
+ public RowDataToAttributeValueConverter(
+ DataType physicalDataType, List<String> primaryKey, boolean
ignoreNulls) {
this.physicalDataType = physicalDataType;
+ this.primaryKey = primaryKey;
this.tableSchema = createTableSchema();
this.ignoreNulls = ignoreNulls;
}
@@ -68,6 +86,32 @@ public class RowDataToAttributeValueConverter {
return tableSchema.itemToMap(row, ignoreNulls);
}
+ /**
+ * Builds a map containing only the primary key attributes of the given
row. This is used for
+ * {@code DELETE} requests, where DynamoDB requires the request to contain
only the primary key
+ * (partition key and, if present, sort key) rather than the whole item.
+ *
+ * @param row the row to extract the primary key from
+ * @return a map of the primary key attribute names to their {@link
AttributeValue}s
+ */
+ public Map<String, AttributeValue> convertRowDataToKey(RowData row) {
+ Map<String, AttributeValue> item = tableSchema.itemToMap(row,
ignoreNulls);
+ Map<String, AttributeValue> key = new LinkedHashMap<>();
+ for (String keyAttributeName : primaryKey) {
+ AttributeValue value = item.get(keyAttributeName);
+ if (value == null) {
+ throw new IllegalArgumentException(
+ String.format(
+ "The row to delete is missing a value for the
primary key "
+ + "attribute '%s'. A DELETE request
must contain all "
+ + "primary key attributes.",
+ keyAttributeName));
+ }
+ key.put(keyAttributeName, value);
+ }
+ return key;
+ }
+
private StaticTableSchema<RowData> createTableSchema() {
List<DataTypes.Field> fields = DataType.getFields(physicalDataType);
StaticTableSchema.Builder<RowData> builder =
TableSchema.builder(RowData.class);
diff --git
a/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactoryTest.java
b/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactoryTest.java
index f76205ea..4676214b 100644
---
a/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactoryTest.java
+++
b/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactoryTest.java
@@ -24,6 +24,7 @@ import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.UniqueConstraint;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.sink.SinkV2Provider;
import org.apache.flink.table.data.RowData;
@@ -77,6 +78,7 @@ public class DynamoDbDynamicSinkFactoryTest {
DynamoDbDynamicSink.builder()
.setTableName(DYNAMO_DB_TABLE_NAME)
.setOverwriteByPartitionKeys(new
HashSet<>(partitionKeys))
+ .setPrimaryKey(List.of())
.setDynamoDbClientProperties(dynamoDbClientProperties)
.setPhysicalDataType(sinkSchema.toPhysicalRowDataType())
.build();
@@ -109,6 +111,7 @@ public class DynamoDbDynamicSinkFactoryTest {
DynamoDbDynamicSink.builder()
.setTableName(DYNAMO_DB_TABLE_NAME)
.setOverwriteByPartitionKeys(new HashSet<>())
+ .setPrimaryKey(List.of())
.setDynamoDbClientProperties(defaultSinkProperties())
.setPhysicalDataType(sinkSchema.toPhysicalRowDataType())
.build();
@@ -127,6 +130,7 @@ public class DynamoDbDynamicSinkFactoryTest {
DynamoDbDynamicSink.builder()
.setTableName(DYNAMO_DB_TABLE_NAME)
.setOverwriteByPartitionKeys(new HashSet<>())
+ .setPrimaryKey(List.of())
.setDynamoDbClientProperties(defaultSinkProperties())
.setPhysicalDataType(sinkSchema.toPhysicalRowDataType())
.build();
@@ -154,6 +158,7 @@ public class DynamoDbDynamicSinkFactoryTest {
DynamoDbDynamicSink.builder()
.setTableName(DYNAMO_DB_TABLE_NAME)
.setOverwriteByPartitionKeys(new
HashSet<>(partitionKeys))
+ .setPrimaryKey(List.of())
.setDynamoDbClientProperties(defaultSinkProperties())
.setPhysicalDataType(sinkSchema.toPhysicalRowDataType())
.setFailOnError(true)
@@ -187,6 +192,7 @@ public class DynamoDbDynamicSinkFactoryTest {
DynamoDbDynamicSink.builder()
.setTableName(DYNAMO_DB_TABLE_NAME)
.setOverwriteByPartitionKeys(new
HashSet<>(partitionKeys))
+ .setPrimaryKey(List.of())
.setDynamoDbClientProperties(expectedSinkProperties)
.setPhysicalDataType(sinkSchema.toPhysicalRowDataType())
.setFailOnError(true)
@@ -219,6 +225,7 @@ public class DynamoDbDynamicSinkFactoryTest {
DynamoDbDynamicSink.builder()
.setTableName(DYNAMO_DB_TABLE_NAME)
.setOverwriteByPartitionKeys(new
HashSet<>(partitionKeys))
+ .setPrimaryKey(List.of())
.setDynamoDbClientProperties(expectedSinkProperties)
.setPhysicalDataType(sinkSchema.toPhysicalRowDataType())
.setFailOnError(true)
@@ -255,6 +262,7 @@ public class DynamoDbDynamicSinkFactoryTest {
.setMaxTimeInBufferMS(1000)
.setTableName(DYNAMO_DB_TABLE_NAME)
.setOverwriteByPartitionKeys(new
HashSet<>(partitionKeys))
+ .setPrimaryKey(List.of())
.setDynamoDbClientProperties(defaultSinkProperties())
.setPhysicalDataType(sinkSchema.toPhysicalRowDataType())
.build();
@@ -316,6 +324,125 @@ public class DynamoDbDynamicSinkFactoryTest {
.withMessageContaining(AWS_REGION.key());
}
+ @Test
+ void testPrimaryKeyIsReadFromSchema() {
+ ResolvedSchema sinkSchema =
+ new ResolvedSchema(
+ defaultSinkColumns(),
+ List.of(),
+ UniqueConstraint.primaryKey("pk",
List.of("partition_key")));
+ Map<String, String> sinkOptions = defaultSinkOptions().build();
+ List<String> partitionKeys = List.of("partition_key");
+
+ DynamoDbDynamicSink actualSink =
+ (DynamoDbDynamicSink) createTableSink(sinkSchema,
partitionKeys, sinkOptions);
+
+ DynamoDbDynamicSink expectedSink =
+ (DynamoDbDynamicSink)
+ DynamoDbDynamicSink.builder()
+ .setTableName(DYNAMO_DB_TABLE_NAME)
+ .setOverwriteByPartitionKeys(new
HashSet<>(partitionKeys))
+ .setPrimaryKey(List.of("partition_key"))
+
.setDynamoDbClientProperties(defaultSinkProperties())
+
.setPhysicalDataType(sinkSchema.toPhysicalRowDataType())
+ .build();
+
+
assertThat(actualSink).usingRecursiveComparison().isEqualTo(expectedSink);
+ }
+
+ @Test
+ void testCompositePrimaryKeyPreservesOrder() {
+ ResolvedSchema sinkSchema =
+ new ResolvedSchema(
+ defaultSinkColumns(),
+ List.of(),
+ UniqueConstraint.primaryKey("pk",
List.of("partition_key", "sort_key")));
+ Map<String, String> sinkOptions = defaultSinkOptions().build();
+ List<String> partitionKeys = List.of("partition_key", "sort_key");
+
+ DynamoDbDynamicSink actualSink =
+ (DynamoDbDynamicSink) createTableSink(sinkSchema,
partitionKeys, sinkOptions);
+
+ DynamoDbDynamicSink expectedSink =
+ (DynamoDbDynamicSink)
+ DynamoDbDynamicSink.builder()
+ .setTableName(DYNAMO_DB_TABLE_NAME)
+ .setOverwriteByPartitionKeys(new
HashSet<>(partitionKeys))
+ .setPrimaryKey(List.of("partition_key",
"sort_key"))
+
.setDynamoDbClientProperties(defaultSinkProperties())
+
.setPhysicalDataType(sinkSchema.toPhysicalRowDataType())
+ .build();
+
+
assertThat(actualSink).usingRecursiveComparison().isEqualTo(expectedSink);
+ }
+
+ @Test
+ void testPrimaryKeyDefaultsPartitionKeysWhenNotPartitioned() {
+ // No PARTITIONED BY clause: the primary key must be used for
client-side deduplication.
+ ResolvedSchema sinkSchema =
+ new ResolvedSchema(
+ defaultSinkColumns(),
+ List.of(),
+ UniqueConstraint.primaryKey("pk",
List.of("partition_key")));
+ Map<String, String> sinkOptions = defaultSinkOptions().build();
+
+ DynamoDbDynamicSink actualSink =
+ (DynamoDbDynamicSink) createTableSink(sinkSchema, sinkOptions);
+
+ DynamoDbDynamicSink expectedSink =
+ (DynamoDbDynamicSink)
+ DynamoDbDynamicSink.builder()
+ .setTableName(DYNAMO_DB_TABLE_NAME)
+ .setOverwriteByPartitionKeys(
+ new
HashSet<>(List.of("partition_key")))
+ .setPrimaryKey(List.of("partition_key"))
+
.setDynamoDbClientProperties(defaultSinkProperties())
+
.setPhysicalDataType(sinkSchema.toPhysicalRowDataType())
+ .build();
+
+
assertThat(actualSink).usingRecursiveComparison().isEqualTo(expectedSink);
+ }
+
+ @Test
+ void testBadTableSinkWithPrimaryKeyOfMoreThanTwoColumns() {
+ ResolvedSchema sinkSchema =
+ new ResolvedSchema(
+ defaultSinkColumns(),
+ List.of(),
+ UniqueConstraint.primaryKey(
+ "pk", List.of("partition_key", "sort_key",
"payload")));
+ Map<String, String> sinkOptions = defaultSinkOptions().build();
+
+ assertThatExceptionOfType(ValidationException.class)
+ .isThrownBy(() -> createTableSink(sinkSchema, sinkOptions))
+ .havingCause()
+ .withMessageContaining("at most two columns");
+ }
+
+ @Test
+ void testBadTableSinkWithPartitionKeysDifferentFromPrimaryKey() {
+ ResolvedSchema sinkSchema =
+ new ResolvedSchema(
+ defaultSinkColumns(),
+ List.of(),
+ UniqueConstraint.primaryKey("pk",
List.of("partition_key")));
+ Map<String, String> sinkOptions = defaultSinkOptions().build();
+ // PARTITIONED BY (sort_key) differs from PRIMARY KEY (partition_key).
+ List<String> partitionKeys = List.of("sort_key");
+
+ assertThatExceptionOfType(ValidationException.class)
+ .isThrownBy(() -> createTableSink(sinkSchema, partitionKeys,
sinkOptions))
+ .havingCause()
+ .withMessageContaining("must reference the same columns");
+ }
+
+ private List<Column> defaultSinkColumns() {
+ return List.of(
+ Column.physical("partition_key", DataTypes.STRING()),
+ Column.physical("sort_key", DataTypes.BIGINT()),
+ Column.physical("payload", DataTypes.STRING()));
+ }
+
private ResolvedSchema createResolvedSchemaUsingAllDataTypes() {
return ResolvedSchema.of(
Column.physical("partition_key", DataTypes.STRING()),
diff --git
a/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataElementConverterTest.java
b/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataElementConverterTest.java
index 1f92bf18..93029bf8 100644
---
a/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataElementConverterTest.java
+++
b/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataElementConverterTest.java
@@ -31,8 +31,11 @@ import org.apache.flink.types.RowKind;
import org.apache.flink.util.InstantiationUtil;
import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
import java.io.IOException;
+import java.util.List;
+import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
@@ -46,6 +49,8 @@ public class RowDataElementConverterTest {
DataTypes.FIELD("payload", DataTypes.STRING()));
private static final RowDataElementConverter elementConverter =
new RowDataElementConverter(DATA_TYPE);
+ private static final RowDataElementConverter
elementConverterWithPrimaryKey =
+ new RowDataElementConverter(DATA_TYPE, List.of("partition_key"));
private static final SinkWriter.Context context = new
UnusedSinkWriterContext();
private static final RowDataToAttributeValueConverter
rowDataToAttributeValueConverter =
new RowDataToAttributeValueConverter(DATA_TYPE);
@@ -89,16 +94,63 @@ public class RowDataElementConverterTest {
}
@Test
- void testDelete() {
+ void testDeleteWithPrimaryKeyIncludesOnlyPrimaryKey() {
RowData rowData = createElement(RowKind.DELETE);
- DynamoDbWriteRequest actualWriteRequest =
elementConverter.apply(rowData, context);
- DynamoDbWriteRequest expectedWriterequest =
+ DynamoDbWriteRequest actualWriteRequest =
+ elementConverterWithPrimaryKey.apply(rowData, context);
+
+ Map<String, AttributeValue> expectedKey =
+ Map.of("partition_key",
AttributeValue.builder().s("some_partition_key").build());
+ DynamoDbWriteRequest expectedWriteRequest =
DynamoDbWriteRequest.builder()
.setType(DynamoDbWriteRequestType.DELETE)
+ .setItem(expectedKey)
+ .build();
+
+
assertThat(actualWriteRequest).usingRecursiveComparison().isEqualTo(expectedWriteRequest);
+ // The non-key "payload" attribute must not be part of a DELETE
request.
+
assertThat(actualWriteRequest.getItem()).containsOnlyKeys("partition_key");
+ }
+
+ @Test
+ void testDeleteWithoutPrimaryKeyThrows() {
+ RowData rowData = createElement(RowKind.DELETE);
+
+ assertThatExceptionOfType(TableException.class)
+ .isThrownBy(() -> elementConverter.apply(rowData, context))
+ .withMessageContaining("no PRIMARY KEY is defined");
+ }
+
+ @Test
+ void testPKIgnoredForInsert() {
+ RowData rowData = createElement(RowKind.INSERT);
+ DynamoDbWriteRequest actualWriteRequest =
+ elementConverterWithPrimaryKey.apply(rowData, context);
+ DynamoDbWriteRequest expectedWriteRequest =
+ DynamoDbWriteRequest.builder()
+ .setType(DynamoDbWriteRequestType.PUT)
.setItem(rowDataToAttributeValueConverter.convertRowData(rowData))
.build();
-
assertThat(actualWriteRequest).usingRecursiveComparison().isEqualTo(expectedWriterequest);
+
assertThat(actualWriteRequest).usingRecursiveComparison().isEqualTo(expectedWriteRequest);
+ // Even with a primary key configured, an INSERT still carries the
full item.
+
assertThat(actualWriteRequest.getItem()).containsOnlyKeys("partition_key",
"payload");
+ }
+
+ @Test
+ void testPKIgnoredForUpdateAfter() {
+ RowData rowData = createElement(RowKind.UPDATE_AFTER);
+ DynamoDbWriteRequest actualWriteRequest =
+ elementConverterWithPrimaryKey.apply(rowData, context);
+ DynamoDbWriteRequest expectedWriteRequest =
+ DynamoDbWriteRequest.builder()
+ .setType(DynamoDbWriteRequestType.PUT)
+
.setItem(rowDataToAttributeValueConverter.convertRowData(rowData))
+ .build();
+
+
assertThat(actualWriteRequest).usingRecursiveComparison().isEqualTo(expectedWriteRequest);
+ // Even with a primary key configured, an UPDATE_AFTER still carries
the full item.
+
assertThat(actualWriteRequest.getItem()).containsOnlyKeys("partition_key",
"payload");
}
@Test
diff --git
a/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java
b/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java
index 974a19fe..e1f2c9f2 100644
---
a/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java
+++
b/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java
@@ -34,6 +34,8 @@ import java.math.BigDecimal;
import java.time.Instant;
import java.time.LocalDateTime;
import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -41,6 +43,7 @@ import java.util.stream.IntStream;
import static java.util.Collections.singletonMap;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
/** Test for {@link RowDataToAttributeValueConverter}. */
public class RowDataToAttributeValueConverterTest {
@@ -692,6 +695,132 @@ public class RowDataToAttributeValueConverterTest {
assertThat(actualResult).containsAllEntriesOf(expectedResult);
}
+ @Test
+ void testDeleteOnlyPrimaryKey() {
+ String key = "key";
+ String value = "some_value";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+
+ DataType dataType =
+ DataTypes.ROW(
+ DataTypes.FIELD(key, DataTypes.STRING()),
+ DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, List.of(key));
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowDataToKey(
+ createElement(
+ StringData.fromString(value),
StringData.fromString(otherValue)));
+ Map<String, AttributeValue> expectedResult =
+ singletonMap(key, AttributeValue.builder().s(value).build());
+
+
assertThat(actualResult).containsExactlyInAnyOrderEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testDeleteOnlyPrimaryKeys() {
+ String key = "key";
+ String value = "some_value";
+ String additionalKey = "additional_key";
+ String additionalValue = "additional_value";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+
+ DataType dataType =
+ DataTypes.ROW(
+ DataTypes.FIELD(key, DataTypes.STRING()),
+ DataTypes.FIELD(additionalKey, DataTypes.STRING()),
+ DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, List.of(key,
additionalKey));
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowDataToKey(
+ createElement(
+ StringData.fromString(value),
+ StringData.fromString(additionalValue),
+ StringData.fromString(otherValue)));
+ Map<String, AttributeValue> expectedResult = new HashMap<>();
+ expectedResult.put(key, AttributeValue.builder().s(value).build());
+ expectedResult.put(additionalKey,
AttributeValue.builder().s(additionalValue).build());
+
+
assertThat(actualResult).containsExactlyInAnyOrderEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testPKIgnoredForInsert() {
+ String key = "key";
+ String value = "some_value";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+
+ DataType dataType =
+ DataTypes.ROW(
+ DataTypes.FIELD(key, DataTypes.STRING()),
+ DataTypes.FIELD(otherField, DataTypes.STRING()));
+ // A primary key is configured, but convertRowData (used for
INSERT/UPDATE_AFTER) must still
+ // return the full item.
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, List.of(key));
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(
+ createElement(
+ StringData.fromString(value),
StringData.fromString(otherValue)));
+ Map<String, AttributeValue> expectedResult = new HashMap<>();
+ expectedResult.put(key, AttributeValue.builder().s(value).build());
+ expectedResult.put(otherField,
AttributeValue.builder().s(otherValue).build());
+
+
assertThat(actualResult).containsExactlyInAnyOrderEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testPKIgnoredForUpdateAfter() {
+ // convertRowData is used for both INSERT and UPDATE_AFTER; verifying
it returns the full
+ // item confirms the primary key is ignored for UPDATE_AFTER as well.
+ String key = "key";
+ String value = "some_value";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+
+ DataType dataType =
+ DataTypes.ROW(
+ DataTypes.FIELD(key, DataTypes.STRING()),
+ DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, List.of(key,
otherField));
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(
+ createElement(
+ StringData.fromString(value),
StringData.fromString(otherValue)));
+ Map<String, AttributeValue> expectedResult = new HashMap<>();
+ expectedResult.put(key, AttributeValue.builder().s(value).build());
+ expectedResult.put(otherField,
AttributeValue.builder().s(otherValue).build());
+
+
assertThat(actualResult).containsExactlyInAnyOrderEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testDeleteThrowsWhenPrimaryKeyValueIsMissing() {
+ String key = "key";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+
+ DataType dataType =
+ DataTypes.ROW(
+ DataTypes.FIELD(key, DataTypes.STRING()),
+ DataTypes.FIELD(otherField, DataTypes.STRING()));
+ // ignoreNulls drops null attributes, so a null primary key value is
absent from the item.
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, List.of(key),
true);
+
+ assertThatExceptionOfType(IllegalArgumentException.class)
+ .isThrownBy(
+ () ->
+
rowDataToAttributeValueConverter.convertRowDataToKey(
+ createElement(null,
StringData.fromString(otherValue))))
+ .withMessageContaining("missing a value for the primary key
attribute");
+ }
+
private RowData createElement(Object... values) {
final int valuesLength = values.length;
GenericRowData element = new GenericRowData(valuesLength);
