You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "robg-eb (via GitHub)" <gi...@apache.org> on 2024/07/24 20:16:53 UTC
[PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
robg-eb opened a new pull request, #152:
URL: https://github.com/apache/flink-connector-aws/pull/152
## Purpose of the change
When `DynamoDbSink` is used with CDC sources, it fails to process DELETE records and throws
```
org.apache.flink.connector.dynamodb.shaded.software.amazon.awssdk.services.dynamodb.model.DynamoDbException: The provided key element does not match the schema
```
This is due to `DynamoDbSinkWriter` passing the whole DynamoDb Item as key instead of the constructed primary key alone.
## Verifying this change
This change added tests and can be verified as follows:
- Added tests to `RowDataToAttributeValueConverterTest.java`:
- `testDeleteOnlyPrimaryKey` - Ensures that for a `DELETE` request, only the (single) PK field is included
- `testDeleteOnlyPrimaryKeys` - Ensures that for a `DELETE` request with a composite PK, both PK fields are included.
- `testPKIgnoredForInsert` - Ensures that PK is ignored when an `INSERT` request is done, and all fields continue to be included as they have been in the past.
- `testPKIgnoredForUpdateAfter` - Ensures that PK is ignored when an `UPDATE_AFTER` request is done, and all fields continue to be included as they have been in the past.
- Ran manual tests following the steps noted in https://issues.apache.org/jira/browse/FLINK-35500 under "Steps To Reproduce". Running the SQL statement as described in Step 6 now properly runs a DELETE in DynamoDB.
## Significant changes
Previously, the `PRIMARY KEY` field had no significance for a DynamoDB Sink via Table API. Now, the `PRIMARY KEY` is required when processing a CDC stream that contains `DELETES`. This is not a 'breaking change' because the previous behavior for processing a CDC stream containing `DELETES` was already a failure (`The provided key element does not match the schema`). This change now provides a clear exception informing users to specify a Primary Key to avoid that failure.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "dzikosc (via GitHub)" <gi...@apache.org>.
dzikosc commented on code in PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#discussion_r1757386924
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java:
##########
@@ -58,6 +58,17 @@ public DynamicTableSink createDynamicTableSink(Context context) {
.setDynamoDbClientProperties(
dynamoDbConfiguration.getSinkClientProperties());
+ if (catalogTable.getResolvedSchema().getPrimaryKey().isPresent()) {
Review Comment:
IN case of DynamoDB contents of PrimaryKey and Partition Key should be the same to ensure semantic correctness of various operations.
To simplify the usage, we should aim to use just PrimaryKey property, as it's a fairly intuitive and well understood concept for DDB users. In the meantime, we should probably aim to start deprecation process of the `partition by` mechanism.
The first step could be
- if PK is set, but `partition by` is not, use PK as `partition by`
- document that `partition by` is on deprecation path
- log warnings when `partition by` is being used
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "dzikosc (via GitHub)" <gi...@apache.org>.
dzikosc commented on code in PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#discussion_r1757386924
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java:
##########
@@ -58,6 +58,17 @@ public DynamicTableSink createDynamicTableSink(Context context) {
.setDynamoDbClientProperties(
dynamoDbConfiguration.getSinkClientProperties());
+ if (catalogTable.getResolvedSchema().getPrimaryKey().isPresent()) {
Review Comment:
IN case of DynamoDB contents of PrimaryKey and PARTITION BY should be the same to ensure semantic correctness of various operations.
To avoid repetititiveness and simplify the usage, we should aim to use just PrimaryKey property, as it's a fairly intuitive and well understood concept for DDB users. In the meantime, we should probably aim to start deprecation process of the `partition by` mechanism.
The first step could be
- if PK is set, but `partition by` is not, use PK as `partition by`
- document that `partition by` is on deprecation path
- log warnings when `partition by` is being used
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "z3d1k (via GitHub)" <gi...@apache.org>.
z3d1k commented on code in PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#discussion_r1711207726
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverter.java:
##########
@@ -43,14 +47,35 @@ public class RowDataToAttributeValueConverter {
private final DataType physicalDataType;
private final TableSchema<RowData> tableSchema;
+ private final Set<String> primaryKeys;
- public RowDataToAttributeValueConverter(DataType physicalDataType) {
+ public RowDataToAttributeValueConverter(DataType physicalDataType, Set<String> primaryKeys) {
Review Comment:
Instead of duplicating logic, it will be better to call constructor with all parameters from one with defaults. This way there only 1 place to change if required:
```
public RowDataToAttributeValueConverter(DataType physicalDataType) {
this(physicalDataType, Collections.emptySet());
}
```
##########
flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataElementConverterTest.java:
##########
@@ -45,10 +48,17 @@ public class RowDataElementConverterTest {
DataTypes.FIELD("partition_key", DataTypes.STRING()),
DataTypes.FIELD("payload", DataTypes.STRING()));
private static final RowDataElementConverter elementConverter =
- new RowDataElementConverter(DATA_TYPE);
+ new RowDataElementConverter(DATA_TYPE, null);
private static final SinkWriter.Context context = new UnusedSinkWriterContext();
private static final RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
- new RowDataToAttributeValueConverter(DATA_TYPE);
+ new RowDataToAttributeValueConverter(DATA_TYPE, null);
+
+ private static final Set<String> primaryKeys =
+ new HashSet<>(Collections.singletonList("partition_key"));
Review Comment:
nit: You can use `Collections.singleton(...)` to create Set (immutable) with a single element.
##########
flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java:
##########
@@ -51,7 +58,7 @@ void testChar() {
DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.CHAR(9)));
RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
- new RowDataToAttributeValueConverter(dataType);
+ new RowDataToAttributeValueConverter(dataType, null);
Review Comment:
Here and below, you can remove `null` argument since you've added constructor overload
##########
flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java:
##########
@@ -547,12 +552,132 @@ void testInstantArray() {
assertThat(actualResult).containsAllEntriesOf(expectedResult);
}
+ @Test
+ void testDeleteOnlyPrimaryKey() {
+ String key = "key";
+ String value = "some_string";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>(Collections.singletonList(key));
+
+ // Create a Row with two fields - "key" and "otherField". "key" is the single primary key.
+ // For a Delete request, only "key" should be included in the expectedResult, and not "otherField".
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.STRING()), DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData = createElementWithMultipleFields(StringData.fromString(value), StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.DELETE);
+
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ Map<String, AttributeValue> expectedResult =
+ singletonMap(key, AttributeValue.builder().s(value).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ assertThat(expectedResult).containsAllEntriesOf(actualResult);
+ }
+
+ @Test
+ void testDeleteOnlyPrimaryKeys() {
+ String key = "key";
+ String value = "some_string";
+ String additionalKey = "additional_key";
+ String additionalValue = "additional_value";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>();
+ primaryKeys.add(key);
+ primaryKeys.add(additionalKey);
+
+ // Create a Row with three fields - "key", "additional_key", and "otherField".
+ // "key" and "additional_key" make up the composite primary key.
+ // For a Delete request, only "key" and "additional_key" should be included in the expectedResult, and not "otherField".
+ DataType dataType = DataTypes.ROW(
+ DataTypes.FIELD(key, DataTypes.STRING()),
+ DataTypes.FIELD(additionalKey, DataTypes.STRING()),
+ DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData = createElementWithMultipleFields(
+ StringData.fromString(value), StringData.fromString(additionalValue), StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.DELETE);
+
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ Map<String, AttributeValue> expectedResult = new HashMap<>();
+ expectedResult.put(key, AttributeValue.builder().s(value).build());
+ expectedResult.put(additionalKey, AttributeValue.builder().s(additionalValue).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ assertThat(expectedResult).containsAllEntriesOf(actualResult);
+ }
+
+ @Test
+ void testPKIgnoredForInsert() {
+ String key = "key";
+ String value = "some_string";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>(Collections.singletonList(key));
+
+ // Create a Row with two fields - "key" and "otherField". "key" is the primary key.
+ // For an Insert request, all fields should be included regardless of the Primary Key.
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.STRING()), DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData = createElementWithMultipleFields(StringData.fromString(value), StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.INSERT);
+
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ Map<String, AttributeValue> expectedResult = new HashMap<>();
+ expectedResult.put(key, AttributeValue.builder().s(value).build());
+ expectedResult.put(otherField, AttributeValue.builder().s(otherValue).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
Review Comment:
You can use `assertThat(...).containsExactlyInAnyOrderEntriesOf(...)` here to check that maps contain same elements without worrying about order.
##########
flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java:
##########
@@ -547,12 +554,180 @@ void testInstantArray() {
assertThat(actualResult).containsAllEntriesOf(expectedResult);
}
+ @Test
+ void testDeleteExceptionWhenNoPrimaryKey() {
+ String key = "key";
+ String value = "some_string";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = Collections.emptySet();
+
+ // Create a Row with two fields - "key" and "otherField". "key" is the single primary key.
+ DataType dataType =
+ DataTypes.ROW(
+ DataTypes.FIELD(key, DataTypes.STRING()),
+ DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData =
+ createElementWithMultipleFields(
+ StringData.fromString(value), StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.DELETE);
+
+ assertThrows(
+ TableException.class,
+ () -> {
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ });
+ }
+
+ void testDeleteOnlyPrimaryKey() {
+ String key = "key";
+ String value = "some_string";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>(Collections.singletonList(key));
+
+ // Create a Row with two fields - "key" and "otherField". "key" is the single primary key.
+ // For a Delete request, only "key" should be included in the expectedResult, and not
+ // "otherField".
+ DataType dataType =
+ DataTypes.ROW(
+ DataTypes.FIELD(key, DataTypes.STRING()),
+ DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData =
+ createElementWithMultipleFields(
+ StringData.fromString(value), StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.DELETE);
+
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ Map<String, AttributeValue> expectedResult =
+ singletonMap(key, AttributeValue.builder().s(value).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ assertThat(expectedResult).containsAllEntriesOf(actualResult);
+ }
+
+ @Test
+ void testDeleteOnlyPrimaryKeys() {
+ String key = "key";
+ String value = "some_string";
+ String additionalKey = "additional_key";
+ String additionalValue = "additional_value";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>();
+ primaryKeys.add(key);
+ primaryKeys.add(additionalKey);
+
+ // Create a Row with three fields - "key", "additional_key", and "otherField".
+ // "key" and "additional_key" make up the composite primary key.
+ // For a Delete request, only "key" and "additional_key" should be included in the
+ // expectedResult, and not "otherField".
+ DataType dataType =
+ DataTypes.ROW(
+ DataTypes.FIELD(key, DataTypes.STRING()),
+ DataTypes.FIELD(additionalKey, DataTypes.STRING()),
+ DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData =
+ createElementWithMultipleFields(
+ StringData.fromString(value),
+ StringData.fromString(additionalValue),
+ StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.DELETE);
+
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ Map<String, AttributeValue> expectedResult = new HashMap<>();
+ expectedResult.put(key, AttributeValue.builder().s(value).build());
+ expectedResult.put(additionalKey, AttributeValue.builder().s(additionalValue).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ assertThat(expectedResult).containsAllEntriesOf(actualResult);
+ }
+
+ @Test
+ void testPKIgnoredForInsert() {
+ String key = "key";
+ String value = "some_string";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>(Collections.singletonList(key));
+
+ // Create a Row with two fields - "key" and "otherField". "key" is the primary key.
+ // For an Insert request, all fields should be included regardless of the Primary Key.
+ DataType dataType =
+ DataTypes.ROW(
+ DataTypes.FIELD(key, DataTypes.STRING()),
+ DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData =
+ createElementWithMultipleFields(
+ StringData.fromString(value), StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.INSERT);
+
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ Map<String, AttributeValue> expectedResult = new HashMap<>();
+ expectedResult.put(key, AttributeValue.builder().s(value).build());
+ expectedResult.put(otherField, AttributeValue.builder().s(otherValue).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ assertThat(expectedResult).containsAllEntriesOf(actualResult);
+ }
+
+ @Test
+ void testPKIgnoredForUpdateAfter() {
+ String key = "key";
+ String value = "some_string";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>(Collections.singletonList(key));
Review Comment:
nit: `Collections.singleton(key)`
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "robg-eb (via GitHub)" <gi...@apache.org>.
robg-eb commented on code in PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#discussion_r1715947230
##########
flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataElementConverterTest.java:
##########
@@ -45,10 +48,17 @@ public class RowDataElementConverterTest {
DataTypes.FIELD("partition_key", DataTypes.STRING()),
DataTypes.FIELD("payload", DataTypes.STRING()));
private static final RowDataElementConverter elementConverter =
- new RowDataElementConverter(DATA_TYPE);
+ new RowDataElementConverter(DATA_TYPE, null);
private static final SinkWriter.Context context = new UnusedSinkWriterContext();
private static final RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
- new RowDataToAttributeValueConverter(DATA_TYPE);
+ new RowDataToAttributeValueConverter(DATA_TYPE, null);
+
+ private static final Set<String> primaryKeys =
+ new HashSet<>(Collections.singletonList("partition_key"));
Review Comment:
Implemented these changes now.
##########
flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java:
##########
@@ -51,7 +58,7 @@ void testChar() {
DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.CHAR(9)));
RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
- new RowDataToAttributeValueConverter(dataType);
+ new RowDataToAttributeValueConverter(dataType, null);
Review Comment:
Thanks! Implemented that change.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "vahmed-hamdy (via GitHub)" <gi...@apache.org>.
vahmed-hamdy commented on PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#issuecomment-2254549295
There are some style violations, Could you run `mvn spotless:apply` ahead of the PR please?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "robg-eb (via GitHub)" <gi...@apache.org>.
robg-eb commented on PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#issuecomment-2287168263
I addressed all the open comments from @z3d1k now. Thank you all for the suggestions! Sounds like @hlteoh37 is the next who should take a look?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "dzikosc (via GitHub)" <gi...@apache.org>.
dzikosc commented on code in PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#discussion_r1757368121
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java:
##########
@@ -58,6 +58,17 @@ public DynamicTableSink createDynamicTableSink(Context context) {
.setDynamoDbClientProperties(
dynamoDbConfiguration.getSinkClientProperties());
+ if (catalogTable.getResolvedSchema().getPrimaryKey().isPresent()) {
+ builder =
+ builder.setPrimaryKeys(
+ new HashSet<>(
Review Comment:
DynamoDB primary key is an ordered set of properties - partitionKey and then sortKey. Let's model it like that.
I would suggest using those as dedicated properties instead in the Sink model. We could add an extra validation here to ensure consistency with DDB schema.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "dzikosc (via GitHub)" <gi...@apache.org>.
dzikosc commented on code in PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#discussion_r1757365286
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSink.java:
##########
@@ -62,7 +63,8 @@ protected DynamoDbDynamicSink(
boolean failOnError,
Properties dynamoDbClientProperties,
DataType physicalDataType,
- Set<String> overwriteByPartitionKeys) {
+ Set<String> overwriteByPartitionKeys,
+ Set<String> primaryKeys) {
Review Comment:
In DynamoDB nomenclature, this set of properties is a primaryKey (not primaryKeys). Let's use a proper name
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "dzikosc (via GitHub)" <gi...@apache.org>.
dzikosc commented on code in PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#discussion_r1757368121
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java:
##########
@@ -58,6 +58,17 @@ public DynamicTableSink createDynamicTableSink(Context context) {
.setDynamoDbClientProperties(
dynamoDbConfiguration.getSinkClientProperties());
+ if (catalogTable.getResolvedSchema().getPrimaryKey().isPresent()) {
+ builder =
+ builder.setPrimaryKeys(
+ new HashSet<>(
Review Comment:
DynamoDB primary key is an ordered set of properties - partitionKey and then sortKey. Let's model it like that.
I would suggest using those as dedicated properties instead in the Sink model. We could add an extra validation here to ensure consistency with DDB schema.
- if one column provided that's Partition Key
- if two columns provided, first is Partition Key, the 2nd is a sortKey
- if three or more columns are specified, table registration should fail
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "riyarawat-amazon (via GitHub)" <gi...@apache.org>.
riyarawat-amazon commented on PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#issuecomment-5050442293
@RobGoretsky I have opened https://github.com/apache/flink-connector-aws/pull/253 as a rebased continuation of this PR. Thanks!!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "riyarawat-amazon (via GitHub)" <gi...@apache.org>.
riyarawat-amazon commented on PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#issuecomment-5036710096
Opened #253 (https://github.com/apache/flink-connector-aws/pull/253) as a rebased continuation of this PR, addressing the outstanding review comments
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "robg-eb (via GitHub)" <gi...@apache.org>.
robg-eb commented on PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#issuecomment-2263251802
@vahmed-hamdy -
> There are some style violations, Could you run `mvn spotless:apply` ahead of the PR please?
Done, I've applied updated formatting now!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 commented on PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#issuecomment-2454252403
@robg-eb Thank you for working on this. Can I check if you plan to address the remaining comments?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "robg-eb (via GitHub)" <gi...@apache.org>.
robg-eb commented on code in PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#discussion_r1765702206
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java:
##########
@@ -58,6 +58,17 @@ public DynamicTableSink createDynamicTableSink(Context context) {
.setDynamoDbClientProperties(
dynamoDbConfiguration.getSinkClientProperties());
+ if (catalogTable.getResolvedSchema().getPrimaryKey().isPresent()) {
+ builder =
+ builder.setPrimaryKeys(
+ new HashSet<>(
Review Comment:
@nicusX - While having a dedicated `PrimaryKey` object with two fields _partitionKey_ and _sortKey_ might work for the DataStream API if I were to add it to the Sink model, I am not clear on how that would then translate to the Table API. I was hoping to just use the fact that the Table API / SQL API already supports the concept of passing in a `PRIMARY KEY` for that.
I don't think there's any reason here why we would need to separate the partition key and sort key for the purpose here of identifying the Primary Key - and in fact, there is also some naming collision then as well as the current Table API / SQL connector supports passing in a `PARTITIONED BY` clause, adding to potential confusion.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "bradodarb (via GitHub)" <gi...@apache.org>.
bradodarb commented on PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#issuecomment-3336181496
It's too bad this bugfix is withering on the vine, and really makes me wonder how many folks are actually using Dynamodb as a sink via SQL/Table API 😬.
My team is a bit blocked by this bug, it'd be nice to pick it back up.
If core AWS maintainers aren't triaging this as a priority issue (which it should be in my unsolicited opinion 😄 )- and @robg-eb no longer has time/energy to get this over the finish line, I'd be ok with giving it a shot.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "robg-eb (via GitHub)" <gi...@apache.org>.
robg-eb commented on code in PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#discussion_r1700332902
##########
flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java:
##########
@@ -547,12 +552,132 @@ void testInstantArray() {
assertThat(actualResult).containsAllEntriesOf(expectedResult);
}
+ @Test
+ void testDeleteOnlyPrimaryKey() {
+ String key = "key";
+ String value = "some_string";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>(Collections.singletonList(key));
+
+ // Create a Row with two fields - "key" and "otherField". "key" is the single primary key.
+ // For a Delete request, only "key" should be included in the expectedResult, and not "otherField".
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.STRING()), DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData = createElementWithMultipleFields(StringData.fromString(value), StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.DELETE);
+
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ Map<String, AttributeValue> expectedResult =
+ singletonMap(key, AttributeValue.builder().s(value).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ assertThat(expectedResult).containsAllEntriesOf(actualResult);
+ }
+
+ @Test
+ void testDeleteOnlyPrimaryKeys() {
+ String key = "key";
+ String value = "some_string";
+ String additionalKey = "additional_key";
+ String additionalValue = "additional_value";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>();
+ primaryKeys.add(key);
+ primaryKeys.add(additionalKey);
+
+ // Create a Row with three fields - "key", "additional_key", and "otherField".
+ // "key" and "additional_key" make up the composite primary key.
+ // For a Delete request, only "key" and "additional_key" should be included in the expectedResult, and not "otherField".
+ DataType dataType = DataTypes.ROW(
+ DataTypes.FIELD(key, DataTypes.STRING()),
+ DataTypes.FIELD(additionalKey, DataTypes.STRING()),
+ DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData = createElementWithMultipleFields(
+ StringData.fromString(value), StringData.fromString(additionalValue), StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.DELETE);
+
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ Map<String, AttributeValue> expectedResult = new HashMap<>();
+ expectedResult.put(key, AttributeValue.builder().s(value).build());
+ expectedResult.put(additionalKey, AttributeValue.builder().s(additionalValue).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ assertThat(expectedResult).containsAllEntriesOf(actualResult);
+ }
+
+ @Test
+ void testPKIgnoredForInsert() {
+ String key = "key";
+ String value = "some_string";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>(Collections.singletonList(key));
+
+ // Create a Row with two fields - "key" and "otherField". "key" is the primary key.
+ // For an Insert request, all fields should be included regardless of the Primary Key.
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.STRING()), DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData = createElementWithMultipleFields(StringData.fromString(value), StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.INSERT);
+
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ Map<String, AttributeValue> expectedResult = new HashMap<>();
+ expectedResult.put(key, AttributeValue.builder().s(value).build());
+ expectedResult.put(otherField, AttributeValue.builder().s(otherValue).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
Review Comment:
@vahmed-hamdy - I took a look at doing this, but `containsExactly` only works for (ordered) Lists, whereas the things we're comparing here are actually `HashMap` . We'd have to convert these HashMap and impose an order on them as lists to compare them, which would seem to be more complicated than what I have here in just running two separate assertions - would you agree ? Let me know if you think there's a better way - I'm admittedly not a Java expert..
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "robg-eb (via GitHub)" <gi...@apache.org>.
robg-eb commented on code in PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#discussion_r1700377928
##########
flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java:
##########
@@ -547,12 +552,132 @@ void testInstantArray() {
assertThat(actualResult).containsAllEntriesOf(expectedResult);
}
+ @Test
Review Comment:
Yes, great idea, I've done this now in latest commits!
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverter.java:
##########
@@ -43,14 +47,35 @@ public class RowDataToAttributeValueConverter {
private final DataType physicalDataType;
private final TableSchema<RowData> tableSchema;
+ private final Set<String> primaryKeys;
- public RowDataToAttributeValueConverter(DataType physicalDataType) {
+ public RowDataToAttributeValueConverter(DataType physicalDataType, Set<String> primaryKeys) {
Review Comment:
Yes, I've done this now in latest commits, please review.. Thank you!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "vahmed-hamdy (via GitHub)" <gi...@apache.org>.
vahmed-hamdy commented on PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#issuecomment-2284706925
Thanks @robg-eb ,
@hlteoh37 could you please take a look
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "dzikosc (via GitHub)" <gi...@apache.org>.
dzikosc commented on code in PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#discussion_r1757390907
##########
docs/content/docs/connectors/table/dynamodb.md:
##########
@@ -303,6 +303,46 @@ WITH (
);
```
+## Specifying Primary Key For Deletes
+
+The DynamoDB sink supports Delete requests to DynamoDB, but AWS requires that a Dynamo Delete request contain **only** the key field(s), or else the Delete request will fail with `DynamoDbException: The provided key element does not match the schema`.
+Thus, if a Changelog stream being is being written to DynamoDB that contains DELETEs, you must specify the `PRIMARY KEY` on the table.
+This `PRIMARY KEY` specified for the Flink SQL Table must match the actual Primary Key of the DynamoDB table - so it must be either just the Partition Key, or in case of a composite primary key, it must be the Partition Key and Sort Key.
+
+Example For Partition Key as only Primary Key:
+```sql
+CREATE TABLE DynamoDbTable (
+ `user_id` BIGINT,
+ `item_id` BIGINT,
+ `category_id` BIGINT,
+ `behavior` STRING,
+ PRIMARY KEY (user_id) NOT ENFORCED
Review Comment:
Why do we use NOT ENFORCED qualifier? In case of DynamoDB Primary Key property (or properties) are always present on the record.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "dzikosc (via GitHub)" <gi...@apache.org>.
dzikosc commented on code in PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#discussion_r1757386924
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java:
##########
@@ -58,6 +58,17 @@ public DynamicTableSink createDynamicTableSink(Context context) {
.setDynamoDbClientProperties(
dynamoDbConfiguration.getSinkClientProperties());
+ if (catalogTable.getResolvedSchema().getPrimaryKey().isPresent()) {
Review Comment:
IN case of DynamoDB contents of PrimaryKey and Partition Key should be the same to ensure semantic correctness of various operations.
To avoid repetititiveness and simplify the usage, we should aim to use just PrimaryKey property, as it's a fairly intuitive and well understood concept for DDB users. In the meantime, we should probably aim to start deprecation process of the `partition by` mechanism.
The first step could be
- if PK is set, but `partition by` is not, use PK as `partition by`
- document that `partition by` is on deprecation path
- log warnings when `partition by` is being used
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "nicusX (via GitHub)" <gi...@apache.org>.
nicusX commented on code in PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#discussion_r1765718042
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java:
##########
@@ -58,6 +58,17 @@ public DynamicTableSink createDynamicTableSink(Context context) {
.setDynamoDbClientProperties(
dynamoDbConfiguration.getSinkClientProperties());
+ if (catalogTable.getResolvedSchema().getPrimaryKey().isPresent()) {
+ builder =
+ builder.setPrimaryKeys(
+ new HashSet<>(
Review Comment:
Ack for not having an object with separate fields.
However, in this case the `primaryKey` (singular) should be an ordered set (e.g. a List) and not a Set. The order of the two fields is relevant: the first always being the *partitionKey*, in DDB parlance, and the second, if present, the *sortKey*
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "nicusX (via GitHub)" <gi...@apache.org>.
nicusX commented on code in PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#discussion_r1759682631
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java:
##########
@@ -58,6 +58,17 @@ public DynamicTableSink createDynamicTableSink(Context context) {
.setDynamoDbClientProperties(
dynamoDbConfiguration.getSinkClientProperties());
+ if (catalogTable.getResolvedSchema().getPrimaryKey().isPresent()) {
+ builder =
+ builder.setPrimaryKeys(
+ new HashSet<>(
Review Comment:
+1
It must be an ordered set/list of 1 or 2 elements.
Suggestion: if not too complicated, for a user perspective it would be clearer a no-nonsense passing a *PrimaryKey* object with two fields, *partitionKey* and *sortKey*, where *partitionKey*
##########
docs/content/docs/connectors/table/dynamodb.md:
##########
@@ -303,6 +303,46 @@ WITH (
);
```
+## Specifying Primary Key For Deletes
+
+The DynamoDB sink supports Delete requests to DynamoDB, but AWS requires that a Dynamo Delete request contain **only** the key field(s), or else the Delete request will fail with `DynamoDbException: The provided key element does not match the schema`.
+Thus, if a Changelog stream being is being written to DynamoDB that contains DELETEs, you must specify the `PRIMARY KEY` on the table.
+This `PRIMARY KEY` specified for the Flink SQL Table must match the actual Primary Key of the DynamoDB table - so it must be either just the Partition Key, or in case of a composite primary key, it must be the Partition Key and Sort Key.
+
+Example For Partition Key as only Primary Key:
+```sql
+CREATE TABLE DynamoDbTable (
+ `user_id` BIGINT,
+ `item_id` BIGINT,
+ `category_id` BIGINT,
+ `behavior` STRING,
+ PRIMARY KEY (user_id) NOT ENFORCED
+)
+WITH (
+ 'connector' = 'dynamodb',
+ 'table-name' = 'user_behavior',
+ 'aws.region' = 'us-east-2'
+);
+```
+
+Example For Partition Key and Sort Key as Composite Primary Key:
+```sql
+CREATE TABLE DynamoDbTable (
+ `user_id` BIGINT,
+ `item_id` BIGINT,
+ `category_id` BIGINT,
+ `behavior` STRING,
+ PRIMARY KEY (user_id, item_id) NOT ENFORCED
+)
+WITH (
+ 'connector' = 'dynamodb',
+ 'table-name' = 'user_behavior',
+ 'aws.region' = 'us-east-2'
+);
+```
+
+Note that this Primary Key functionality, specified by `PRIMARY KEY`, can be used alongside the Sink Partitioning mentioned above via `PARTITIONED BY` to dedeuplicate data and support DELETEs.
Review Comment:
I would clarify that `PARTITIONED BY` is used by Flink for deduplication within the same batch (and to support delete), but it's different from DynamoDB's *partitionKey*. If the user is familiar with DynamoDB but not much with Flink this may generate lot of confusion.
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java:
##########
@@ -58,6 +58,17 @@ public DynamicTableSink createDynamicTableSink(Context context) {
.setDynamoDbClientProperties(
dynamoDbConfiguration.getSinkClientProperties());
+ if (catalogTable.getResolvedSchema().getPrimaryKey().isPresent()) {
Review Comment:
+1
Also, I'd add a check that, if `PARTITION BY` and `PRIMARY KEY` are both specified, they must be identical
##########
docs/content/docs/connectors/table/dynamodb.md:
##########
@@ -303,6 +303,46 @@ WITH (
);
```
+## Specifying Primary Key For Deletes
+
+The DynamoDB sink supports Delete requests to DynamoDB, but AWS requires that a Dynamo Delete request contain **only** the key field(s), or else the Delete request will fail with `DynamoDbException: The provided key element does not match the schema`.
+Thus, if a Changelog stream being is being written to DynamoDB that contains DELETEs, you must specify the `PRIMARY KEY` on the table.
+This `PRIMARY KEY` specified for the Flink SQL Table must match the actual Primary Key of the DynamoDB table - so it must be either just the Partition Key, or in case of a composite primary key, it must be the Partition Key and Sort Key.
+
+Example For Partition Key as only Primary Key:
+```sql
+CREATE TABLE DynamoDbTable (
+ `user_id` BIGINT,
+ `item_id` BIGINT,
+ `category_id` BIGINT,
+ `behavior` STRING,
+ PRIMARY KEY (user_id) NOT ENFORCED
+)
+WITH (
+ 'connector' = 'dynamodb',
+ 'table-name' = 'user_behavior',
+ 'aws.region' = 'us-east-2'
+);
+```
+
+Example For Partition Key and Sort Key as Composite Primary Key:
+```sql
+CREATE TABLE DynamoDbTable (
+ `user_id` BIGINT,
+ `item_id` BIGINT,
+ `category_id` BIGINT,
+ `behavior` STRING,
+ PRIMARY KEY (user_id, item_id) NOT ENFORCED
+)
+WITH (
+ 'connector' = 'dynamodb',
+ 'table-name' = 'user_behavior',
+ 'aws.region' = 'us-east-2'
+);
+```
+
+Note that this Primary Key functionality, specified by `PRIMARY KEY`, can be used alongside the Sink Partitioning mentioned above via `PARTITIONED BY` to dedeuplicate data and support DELETEs.
Review Comment:
Also, `PARTITIONED BY` and `PRIMARY KEY` should always be the same.
(see other comment by @dzikosc below)
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSink.java:
##########
@@ -62,7 +63,8 @@ protected DynamoDbDynamicSink(
boolean failOnError,
Properties dynamoDbClientProperties,
DataType physicalDataType,
- Set<String> overwriteByPartitionKeys) {
+ Set<String> overwriteByPartitionKeys,
+ Set<String> primaryKeys) {
Review Comment:
+1
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "robg-eb (via GitHub)" <gi...@apache.org>.
robg-eb commented on code in PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#discussion_r1765712462
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java:
##########
@@ -58,6 +58,17 @@ public DynamicTableSink createDynamicTableSink(Context context) {
.setDynamoDbClientProperties(
dynamoDbConfiguration.getSinkClientProperties());
+ if (catalogTable.getResolvedSchema().getPrimaryKey().isPresent()) {
Review Comment:
@dzikosc - Isn't there a potential use case wherein a user of the existing `PARTITION BY` functionality of the sink wants to use it to deduplicate data by something other than the actual Primary Key of the DynamoDB table? For example, the DynamoDB table has:
```
user_id Partition Key
order_id Sort Key
line_number
price
```
Today, the user of the connector could specify a `PARTITION BY user_id, order_id, line_number` to deduplicate incoming data - Maybe this is a contrived case, but just pointing out that we may now be blocking a use case that was previously supported. Thoughts?
Either way, could we keep the deprecation of `PARTITION BY` to a separate PR to reduce the scope of this one?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "vahmed-hamdy (via GitHub)" <gi...@apache.org>.
vahmed-hamdy commented on code in PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#discussion_r1694261617
##########
flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java:
##########
@@ -547,12 +552,132 @@ void testInstantArray() {
assertThat(actualResult).containsAllEntriesOf(expectedResult);
}
+ @Test
+ void testDeleteOnlyPrimaryKey() {
+ String key = "key";
+ String value = "some_string";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>(Collections.singletonList(key));
+
+ // Create a Row with two fields - "key" and "otherField". "key" is the single primary key.
+ // For a Delete request, only "key" should be included in the expectedResult, and not "otherField".
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.STRING()), DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData = createElementWithMultipleFields(StringData.fromString(value), StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.DELETE);
+
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ Map<String, AttributeValue> expectedResult =
+ singletonMap(key, AttributeValue.builder().s(value).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ assertThat(expectedResult).containsAllEntriesOf(actualResult);
+ }
+
+ @Test
+ void testDeleteOnlyPrimaryKeys() {
+ String key = "key";
+ String value = "some_string";
+ String additionalKey = "additional_key";
+ String additionalValue = "additional_value";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>();
+ primaryKeys.add(key);
+ primaryKeys.add(additionalKey);
+
+ // Create a Row with three fields - "key", "additional_key", and "otherField".
+ // "key" and "additional_key" make up the composite primary key.
+ // For a Delete request, only "key" and "additional_key" should be included in the expectedResult, and not "otherField".
+ DataType dataType = DataTypes.ROW(
+ DataTypes.FIELD(key, DataTypes.STRING()),
+ DataTypes.FIELD(additionalKey, DataTypes.STRING()),
+ DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData = createElementWithMultipleFields(
+ StringData.fromString(value), StringData.fromString(additionalValue), StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.DELETE);
+
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ Map<String, AttributeValue> expectedResult = new HashMap<>();
+ expectedResult.put(key, AttributeValue.builder().s(value).build());
+ expectedResult.put(additionalKey, AttributeValue.builder().s(additionalValue).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
Review Comment:
nit: could be done on one assert using containsExactly
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverter.java:
##########
@@ -43,14 +47,35 @@ public class RowDataToAttributeValueConverter {
private final DataType physicalDataType;
private final TableSchema<RowData> tableSchema;
+ private final Set<String> primaryKeys;
- public RowDataToAttributeValueConverter(DataType physicalDataType) {
+ public RowDataToAttributeValueConverter(DataType physicalDataType, Set<String> primaryKeys) {
Review Comment:
nit: could we overload the constructor instead and pass empty set on default?
##########
flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java:
##########
@@ -547,12 +552,132 @@ void testInstantArray() {
assertThat(actualResult).containsAllEntriesOf(expectedResult);
}
+ @Test
+ void testDeleteOnlyPrimaryKey() {
+ String key = "key";
+ String value = "some_string";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>(Collections.singletonList(key));
+
+ // Create a Row with two fields - "key" and "otherField". "key" is the single primary key.
+ // For a Delete request, only "key" should be included in the expectedResult, and not "otherField".
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.STRING()), DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData = createElementWithMultipleFields(StringData.fromString(value), StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.DELETE);
+
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ Map<String, AttributeValue> expectedResult =
+ singletonMap(key, AttributeValue.builder().s(value).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
Review Comment:
nit: could be done on one assert using `containsExactly`
##########
flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java:
##########
@@ -547,12 +552,132 @@ void testInstantArray() {
assertThat(actualResult).containsAllEntriesOf(expectedResult);
}
+ @Test
Review Comment:
Could we add tests for delete with no primary key and verify exception
##########
flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java:
##########
@@ -547,12 +552,132 @@ void testInstantArray() {
assertThat(actualResult).containsAllEntriesOf(expectedResult);
}
+ @Test
+ void testDeleteOnlyPrimaryKey() {
+ String key = "key";
+ String value = "some_string";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>(Collections.singletonList(key));
+
+ // Create a Row with two fields - "key" and "otherField". "key" is the single primary key.
+ // For a Delete request, only "key" should be included in the expectedResult, and not "otherField".
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.STRING()), DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData = createElementWithMultipleFields(StringData.fromString(value), StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.DELETE);
+
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ Map<String, AttributeValue> expectedResult =
+ singletonMap(key, AttributeValue.builder().s(value).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ assertThat(expectedResult).containsAllEntriesOf(actualResult);
+ }
+
+ @Test
+ void testDeleteOnlyPrimaryKeys() {
+ String key = "key";
+ String value = "some_string";
+ String additionalKey = "additional_key";
+ String additionalValue = "additional_value";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>();
+ primaryKeys.add(key);
+ primaryKeys.add(additionalKey);
+
+ // Create a Row with three fields - "key", "additional_key", and "otherField".
+ // "key" and "additional_key" make up the composite primary key.
+ // For a Delete request, only "key" and "additional_key" should be included in the expectedResult, and not "otherField".
+ DataType dataType = DataTypes.ROW(
+ DataTypes.FIELD(key, DataTypes.STRING()),
+ DataTypes.FIELD(additionalKey, DataTypes.STRING()),
+ DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData = createElementWithMultipleFields(
+ StringData.fromString(value), StringData.fromString(additionalValue), StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.DELETE);
+
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ Map<String, AttributeValue> expectedResult = new HashMap<>();
+ expectedResult.put(key, AttributeValue.builder().s(value).build());
+ expectedResult.put(additionalKey, AttributeValue.builder().s(additionalValue).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ assertThat(expectedResult).containsAllEntriesOf(actualResult);
+ }
+
+ @Test
+ void testPKIgnoredForInsert() {
+ String key = "key";
+ String value = "some_string";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>(Collections.singletonList(key));
+
+ // Create a Row with two fields - "key" and "otherField". "key" is the primary key.
+ // For an Insert request, all fields should be included regardless of the Primary Key.
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.STRING()), DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData = createElementWithMultipleFields(StringData.fromString(value), StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.INSERT);
+
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ Map<String, AttributeValue> expectedResult = new HashMap<>();
+ expectedResult.put(key, AttributeValue.builder().s(value).build());
+ expectedResult.put(otherField, AttributeValue.builder().s(otherValue).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
Review Comment:
nit: could be done on one assert using containsExactly
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "RobGoretsky (via GitHub)" <gi...@apache.org>.
RobGoretsky commented on PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#issuecomment-5050299094
Apologies for my absence here - I left the role at the company that was using Flink and DynamoDB, so I no longer have access to an environment to test this - If someone else can carry this forward, it'd be great!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "robg-eb (via GitHub)" <gi...@apache.org>.
robg-eb commented on code in PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#discussion_r1765691788
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSink.java:
##########
@@ -62,7 +63,8 @@ protected DynamoDbDynamicSink(
boolean failOnError,
Properties dynamoDbClientProperties,
DataType physicalDataType,
- Set<String> overwriteByPartitionKeys) {
+ Set<String> overwriteByPartitionKeys,
+ Set<String> primaryKeys) {
Review Comment:
To be clear, you're just suggesting changing the variable name from `primaryKeys` to `primaryKey`, right? And leaving its data type intact of `Set<String>`, since a Primary Key can consist of two fields - the Partition Key and the Sort Key?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#issuecomment-2248823319
Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "robg-eb (via GitHub)" <gi...@apache.org>.
robg-eb commented on code in PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#discussion_r1715947510
##########
flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java:
##########
@@ -547,12 +554,180 @@ void testInstantArray() {
assertThat(actualResult).containsAllEntriesOf(expectedResult);
}
+ @Test
+ void testDeleteExceptionWhenNoPrimaryKey() {
+ String key = "key";
+ String value = "some_string";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = Collections.emptySet();
+
+ // Create a Row with two fields - "key" and "otherField". "key" is the single primary key.
+ DataType dataType =
+ DataTypes.ROW(
+ DataTypes.FIELD(key, DataTypes.STRING()),
+ DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData =
+ createElementWithMultipleFields(
+ StringData.fromString(value), StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.DELETE);
+
+ assertThrows(
+ TableException.class,
+ () -> {
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ });
+ }
+
+ void testDeleteOnlyPrimaryKey() {
+ String key = "key";
+ String value = "some_string";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>(Collections.singletonList(key));
+
+ // Create a Row with two fields - "key" and "otherField". "key" is the single primary key.
+ // For a Delete request, only "key" should be included in the expectedResult, and not
+ // "otherField".
+ DataType dataType =
+ DataTypes.ROW(
+ DataTypes.FIELD(key, DataTypes.STRING()),
+ DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData =
+ createElementWithMultipleFields(
+ StringData.fromString(value), StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.DELETE);
+
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ Map<String, AttributeValue> expectedResult =
+ singletonMap(key, AttributeValue.builder().s(value).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ assertThat(expectedResult).containsAllEntriesOf(actualResult);
+ }
+
+ @Test
+ void testDeleteOnlyPrimaryKeys() {
+ String key = "key";
+ String value = "some_string";
+ String additionalKey = "additional_key";
+ String additionalValue = "additional_value";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>();
+ primaryKeys.add(key);
+ primaryKeys.add(additionalKey);
+
+ // Create a Row with three fields - "key", "additional_key", and "otherField".
+ // "key" and "additional_key" make up the composite primary key.
+ // For a Delete request, only "key" and "additional_key" should be included in the
+ // expectedResult, and not "otherField".
+ DataType dataType =
+ DataTypes.ROW(
+ DataTypes.FIELD(key, DataTypes.STRING()),
+ DataTypes.FIELD(additionalKey, DataTypes.STRING()),
+ DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData =
+ createElementWithMultipleFields(
+ StringData.fromString(value),
+ StringData.fromString(additionalValue),
+ StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.DELETE);
+
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ Map<String, AttributeValue> expectedResult = new HashMap<>();
+ expectedResult.put(key, AttributeValue.builder().s(value).build());
+ expectedResult.put(additionalKey, AttributeValue.builder().s(additionalValue).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ assertThat(expectedResult).containsAllEntriesOf(actualResult);
+ }
+
+ @Test
+ void testPKIgnoredForInsert() {
+ String key = "key";
+ String value = "some_string";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>(Collections.singletonList(key));
+
+ // Create a Row with two fields - "key" and "otherField". "key" is the primary key.
+ // For an Insert request, all fields should be included regardless of the Primary Key.
+ DataType dataType =
+ DataTypes.ROW(
+ DataTypes.FIELD(key, DataTypes.STRING()),
+ DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData =
+ createElementWithMultipleFields(
+ StringData.fromString(value), StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.INSERT);
+
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ Map<String, AttributeValue> expectedResult = new HashMap<>();
+ expectedResult.put(key, AttributeValue.builder().s(value).build());
+ expectedResult.put(otherField, AttributeValue.builder().s(otherValue).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ assertThat(expectedResult).containsAllEntriesOf(actualResult);
+ }
+
+ @Test
+ void testPKIgnoredForUpdateAfter() {
+ String key = "key";
+ String value = "some_string";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>(Collections.singletonList(key));
Review Comment:
Thanks, addressed this now.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found [flink-connector-aws]
Posted by "robg-eb (via GitHub)" <gi...@apache.org>.
robg-eb commented on code in PR #152:
URL: https://github.com/apache/flink-connector-aws/pull/152#discussion_r1700377499
##########
flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java:
##########
@@ -547,12 +552,132 @@ void testInstantArray() {
assertThat(actualResult).containsAllEntriesOf(expectedResult);
}
+ @Test
+ void testDeleteOnlyPrimaryKey() {
+ String key = "key";
+ String value = "some_string";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>(Collections.singletonList(key));
+
+ // Create a Row with two fields - "key" and "otherField". "key" is the single primary key.
+ // For a Delete request, only "key" should be included in the expectedResult, and not "otherField".
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.STRING()), DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData = createElementWithMultipleFields(StringData.fromString(value), StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.DELETE);
+
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ Map<String, AttributeValue> expectedResult =
+ singletonMap(key, AttributeValue.builder().s(value).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
Review Comment:
See response below
##########
flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java:
##########
@@ -547,12 +552,132 @@ void testInstantArray() {
assertThat(actualResult).containsAllEntriesOf(expectedResult);
}
+ @Test
+ void testDeleteOnlyPrimaryKey() {
+ String key = "key";
+ String value = "some_string";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>(Collections.singletonList(key));
+
+ // Create a Row with two fields - "key" and "otherField". "key" is the single primary key.
+ // For a Delete request, only "key" should be included in the expectedResult, and not "otherField".
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.STRING()), DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData = createElementWithMultipleFields(StringData.fromString(value), StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.DELETE);
+
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ Map<String, AttributeValue> expectedResult =
+ singletonMap(key, AttributeValue.builder().s(value).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ assertThat(expectedResult).containsAllEntriesOf(actualResult);
+ }
+
+ @Test
+ void testDeleteOnlyPrimaryKeys() {
+ String key = "key";
+ String value = "some_string";
+ String additionalKey = "additional_key";
+ String additionalValue = "additional_value";
+ String otherField = "other_field";
+ String otherValue = "other_value";
+ Set<String> primaryKeys = new HashSet<>();
+ primaryKeys.add(key);
+ primaryKeys.add(additionalKey);
+
+ // Create a Row with three fields - "key", "additional_key", and "otherField".
+ // "key" and "additional_key" make up the composite primary key.
+ // For a Delete request, only "key" and "additional_key" should be included in the expectedResult, and not "otherField".
+ DataType dataType = DataTypes.ROW(
+ DataTypes.FIELD(key, DataTypes.STRING()),
+ DataTypes.FIELD(additionalKey, DataTypes.STRING()),
+ DataTypes.FIELD(otherField, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType, primaryKeys);
+ RowData rowData = createElementWithMultipleFields(
+ StringData.fromString(value), StringData.fromString(additionalValue), StringData.fromString(otherValue));
+ rowData.setRowKind(RowKind.DELETE);
+
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(rowData);
+ Map<String, AttributeValue> expectedResult = new HashMap<>();
+ expectedResult.put(key, AttributeValue.builder().s(value).build());
+ expectedResult.put(additionalKey, AttributeValue.builder().s(additionalValue).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
Review Comment:
See response below
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org