riyarawat-amazon commented on code in PR #253:
URL:
https://github.com/apache/flink-connector-aws/pull/253#discussion_r3892683763
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java:
##########
@@ -48,14 +52,53 @@ public DynamicTableSink createDynamicTableSink(Context
context) {
DynamoDbConfiguration dynamoDbConfiguration =
new DynamoDbConfiguration(catalogTable.getOptions(),
factoryHelper.getOptions());
+ List<String> primaryKey =
+ catalogTable
+ .getResolvedSchema()
+ .getPrimaryKey()
+ .map(UniqueConstraint::getColumns)
+ .orElse(Collections.emptyList());
+
+ 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));
+ }
+
+ List<String> declaredPartitionKeys = catalogTable.getPartitionKeys();
+
+ // 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.
+ 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));
+ }
+
+ Set<String> overwriteByPartitionKeys = new
HashSet<>(declaredPartitionKeys);
+ if (overwriteByPartitionKeys.isEmpty()) {
+ overwriteByPartitionKeys = new HashSet<>(primaryKey);
+ }
Review Comment:
Ack, will update
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java:
##########
@@ -48,14 +52,53 @@ public DynamicTableSink createDynamicTableSink(Context
context) {
DynamoDbConfiguration dynamoDbConfiguration =
new DynamoDbConfiguration(catalogTable.getOptions(),
factoryHelper.getOptions());
+ List<String> primaryKey =
+ catalogTable
+ .getResolvedSchema()
+ .getPrimaryKey()
+ .map(UniqueConstraint::getColumns)
+ .orElse(Collections.emptyList());
Review Comment:
Ack
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataElementConverter.java:
##########
@@ -37,39 +40,55 @@ 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, Collections.emptyList(), false);
}
public RowDataElementConverter(DataType physicalDataType, boolean
ignoreNulls) {
- this.ignoreNulls = ignoreNulls;
+ this(physicalDataType, Collections.emptyList(), 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 == null ? Collections.emptyList() :
primaryKey;
Review Comment:
Ack
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
