ferenc-csaky commented on code in PR #253:
URL:
https://github.com/apache/flink-connector-aws/pull/253#discussion_r3855594487
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSink.java:
##########
@@ -142,6 +149,7 @@ public static class DynamoDbDynamicTableSinkBuilder
private Properties dynamoDbClientProperties;
private DataType physicalDataType;
private Set<String> overwriteByPartitionKeys;
+ private List<String> primaryKey = Collections.emptyList();
Review Comment:
Do not see the relevance of the default value. The factory will set this
properly, which is why it's omitted for all the other fields. I suggest to
comply with that.
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSink.java:
##########
@@ -77,6 +81,7 @@ protected DynamoDbDynamicSink(
this.dynamoDbClientProperties = dynamoDbClientProperties;
this.physicalDataType = physicalDataType;
this.overwriteByPartitionKeys = overwriteByPartitionKeys;
+ this.primaryKey = primaryKey == null ? Collections.emptyList() :
primaryKey;
Review Comment:
Do not see the relevance of the null check here. This ctor is only called
from the builder, which is only called from the factory, which prevalidates
these fields, which I guess is why it's omitted for all the other fields. I
suggest to comply with that.
##########
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:
I think it makes sense to extract these into 2 separate private methods that
deals with the validation and only returns the value, so the general property
flow is more clean.
##########
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:
nit: Collections.emptyList() -> List.of()
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverter.java:
##########
@@ -51,15 +53,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, Collections.emptyList(), false);
}
public RowDataToAttributeValueConverter(DataType physicalDataType, boolean
ignoreNulls) {
+ this(physicalDataType, Collections.emptyList(), 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 == null ? Collections.emptyList() :
primaryKey;
Review Comment:
Same as my other comments about the null-check
##########
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:
Since this already gets the safe primaryKeys, technically the null-check can
be omitted, so I suggest to do so, as no extra validation is done for the other
fields too.
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverter.java:
##########
@@ -68,6 +87,26 @@ public Map<String, AttributeValue> convertRowData(RowData
row) {
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) {
Review Comment:
If for some reason `value` is null here, shouldn't we throw an exception
here instead of swallowing it and possibly return an empty or invalid map here?
I guess it will fail on the DynamoDB side anyways, but we can probably
pinpoint the anomaly a bit more accurate if we throw an exception here,
probably `IllegalArgumentException` or something like that.
--
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]
