Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found #152

Open
wants to merge 6 commits into
base: main
Choose a base branch
from
Open

[FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to delete elements due to key not found #152

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
07e5746
[FLINK-35500][Connectors/DynamoDB] DynamoDb Table API Sink fails to d…
robg-eb Jul 24, 2024
68f1253
Add documenation updates
robg-eb Jul 24, 2024
aa030f3
Apply spotless formatting updates
robg-eb Aug 1, 2024
5372665
Use overriden constructor, add test for exception
robg-eb Aug 1, 2024
cdfcd05
formatting
robg-eb Aug 1, 2024
621df92
Updates as per PR feedback
robg-eb Aug 13, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 40 additions & 0 deletions docs/content.zh/docs/connectors/table/dynamodb.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -302,6 +302,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.

## Notice

The current implementation of the DynamoDB SQL connector is write-only and doesn't provide an implementation for source queries.
Expand Down
40 changes: 40 additions & 0 deletions docs/content/docs/connectors/table/dynamodb.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why do we use NOT ENFORCED qualifier? In case of DynamoDB Primary Key property (or properties) are always present on the record.

)
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.
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also, PARTITIONED BY and PRIMARY KEY should always be the same.
(see other comment by @dzikosc below)


## Notice

The current implementation of the DynamoDB SQL connector is write-only and doesn't provide an implementation for source queries.
Expand Down
20 changes: 16 additions & 4 deletions ...dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSink.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ public class DynamoDbDynamicSink extends AsyncDynamicTableSink<DynamoDbWriteRequ
private final Properties dynamoDbClientProperties;
private final DataType physicalDataType;
private final Set<String> overwriteByPartitionKeys;
private final Set<String> primaryKeys;

protected DynamoDbDynamicSink(
@Nullable Integer maxBatchSize,
Expand All @@ -62,7 +63,8 @@ protected DynamoDbDynamicSink(
boolean failOnError,
Properties dynamoDbClientProperties,
DataType physicalDataType,
Set<String> overwriteByPartitionKeys) {
Set<String> overwriteByPartitionKeys,
Set<String> primaryKeys) {
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In DynamoDB nomenclature, this set of properties is a primaryKey (not primaryKeys). Let's use a proper name

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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?

super(
maxBatchSize,
maxInFlightRequests,
Expand All @@ -74,6 +76,7 @@ protected DynamoDbDynamicSink(
this.dynamoDbClientProperties = dynamoDbClientProperties;
this.physicalDataType = physicalDataType;
this.overwriteByPartitionKeys = overwriteByPartitionKeys;
this.primaryKeys = primaryKeys;
}

@Override
Expand All @@ -89,7 +92,8 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
.setFailOnError(failOnError)
.setOverwriteByPartitionKeys(new ArrayList<>(overwriteByPartitionKeys))
.setDynamoDbProperties(dynamoDbClientProperties)
.setElementConverter(new RowDataElementConverter(physicalDataType));
.setElementConverter(
new RowDataElementConverter(physicalDataType, primaryKeys));

addAsyncOptionsToSinkBuilder(builder);

Expand All @@ -108,7 +112,8 @@ public DynamicTableSink copy() {
failOnError,
dynamoDbClientProperties,
physicalDataType,
overwriteByPartitionKeys);
overwriteByPartitionKeys,
primaryKeys);
}

@Override
Expand Down Expand Up @@ -136,6 +141,7 @@ public static class DynamoDbDynamicTableSinkBuilder
private Properties dynamoDbClientProperties;
private DataType physicalDataType;
private Set<String> overwriteByPartitionKeys;
private Set<String> primaryKeys;

public DynamoDbDynamicTableSinkBuilder setTableName(String tableName) {
this.tableName = tableName;
Expand Down Expand Up @@ -164,6 +170,11 @@ public DynamoDbDynamicTableSinkBuilder setOverwriteByPartitionKeys(
return this;
}

public DynamoDbDynamicTableSinkBuilder setPrimaryKeys(Set<String> primaryKeys) {
this.primaryKeys = primaryKeys;
return this;
}

@Override
public AsyncDynamicTableSink<DynamoDbWriteRequest> build() {
return new DynamoDbDynamicSink(
Expand All @@ -176,7 +187,8 @@ public AsyncDynamicTableSink<DynamoDbWriteRequest> build() {
failOnError,
dynamoDbClientProperties,
physicalDataType,
overwriteByPartitionKeys);
overwriteByPartitionKeys,
primaryKeys);
}
}
}
11 changes: 11 additions & 0 deletions ...b/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,17 @@ public DynamicTableSink createDynamicTableSink(Context context) {
.setDynamoDbClientProperties(
dynamoDbConfiguration.getSinkClientProperties());

if (catalogTable.getResolvedSchema().getPrimaryKey().isPresent()) {
Copy link

@dzikosc dzikosc Sep 12, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1
Also, I'd add a check that, if PARTITION BY and PRIMARY KEY are both specified, they must be identical

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@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?

builder =
builder.setPrimaryKeys(
new HashSet<>(
Copy link

@dzikosc dzikosc Sep 12, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+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

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@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.

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

catalogTable
.getResolvedSchema()
.getPrimaryKey()
.get()
.getColumns()));
}

addAsyncOptionsToBuilder(dynamoDbConfiguration.getAsyncSinkProperties(), builder);

return builder.build();
Expand Down
10 changes: 7 additions & 3 deletions ...modb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataElementConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,8 @@
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;

import java.util.Set;

/**
* 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
Expand All @@ -36,19 +38,21 @@
public class RowDataElementConverter implements ElementConverter<RowData, DynamoDbWriteRequest> {

private final DataType physicalDataType;
private final Set<String> primaryKeys;
private transient RowDataToAttributeValueConverter rowDataToAttributeValueConverter;

public RowDataElementConverter(DataType physicalDataType) {
public RowDataElementConverter(DataType physicalDataType, Set<String> primaryKeys) {
this.physicalDataType = physicalDataType;
this.primaryKeys = primaryKeys;
this.rowDataToAttributeValueConverter =
new RowDataToAttributeValueConverter(physicalDataType);
new RowDataToAttributeValueConverter(physicalDataType, primaryKeys);
}

@Override
public DynamoDbWriteRequest apply(RowData element, SinkWriter.Context context) {
if (rowDataToAttributeValueConverter == null) {
rowDataToAttributeValueConverter =
new RowDataToAttributeValueConverter(physicalDataType);
new RowDataToAttributeValueConverter(physicalDataType, primaryKeys);
}

DynamoDbWriteRequest.Builder builder =
Expand Down
32 changes: 31 additions & 1 deletion ...main/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,19 +21,24 @@
import org.apache.flink.annotation.Internal;
import org.apache.flink.connector.dynamodb.table.converter.ArrayAttributeConverterProvider;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.conversion.DataStructureConverters;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.KeyValueDataType;
import org.apache.flink.types.RowKind;

import software.amazon.awssdk.enhanced.dynamodb.AttributeConverterProvider;
import software.amazon.awssdk.enhanced.dynamodb.EnhancedType;
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.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import static org.apache.flink.table.data.RowData.createFieldGetter;

Expand All @@ -43,14 +48,39 @@ public class RowDataToAttributeValueConverter {

private final DataType physicalDataType;
private final TableSchema<RowData> tableSchema;
private final Set<String> primaryKeys;

public RowDataToAttributeValueConverter(DataType physicalDataType) {
this(physicalDataType, Collections.emptySet());
}

public RowDataToAttributeValueConverter(DataType physicalDataType, Set<String> primaryKeys) {
robg-eb marked this conversation as resolved.
Show resolved Hide resolved
this.physicalDataType = physicalDataType;
this.primaryKeys = primaryKeys;
this.tableSchema = createTableSchema();
}

public Map<String, AttributeValue> convertRowData(RowData row) {
return tableSchema.itemToMap(row, false);
Map<String, AttributeValue> itemMap = new HashMap<>();
itemMap = tableSchema.itemToMap(row, false);

// In case of DELETE, only the primary key field(s) should be sent in the request
// In order to accomplish this, we need PRIMARY KEY fields to have been set in Table
// definition.
if (row.getRowKind() == RowKind.DELETE) {
if (primaryKeys == null || primaryKeys.isEmpty()) {
throw new TableException(
"PRIMARY KEY on Table must be set for DynamoDB DELETE operation");
}
Map<String, AttributeValue> pkOnlyMap = new HashMap<String, AttributeValue>();
for (String key : primaryKeys) {
AttributeValue value = itemMap.get(key);
pkOnlyMap.put(key, value);
}
return pkOnlyMap;
} else {
return itemMap;
}
}

private StaticTableSchema<RowData> createTableSchema() {
Expand Down
19 changes: 14 additions & 5 deletions .../src/test/java/org/apache/flink/connector/dynamodb/table/RowDataElementConverterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,8 @@
import org.junit.jupiter.api.Test;

import java.io.IOException;
import java.util.Collections;
import java.util.Set;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
Expand All @@ -45,10 +47,16 @@ 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 = Collections.singleton("partition_key");
private static final RowDataElementConverter elementConverterWithPK =
new RowDataElementConverter(DATA_TYPE, primaryKeys);
private static final RowDataToAttributeValueConverter rowDataToAttributeValueConverterWithPK =
new RowDataToAttributeValueConverter(DATA_TYPE, primaryKeys);

@Test
void testInsert() {
Expand Down Expand Up @@ -91,11 +99,12 @@ void testUpdateBeforeIsUnsupported() {
@Test
void testDelete() {
RowData rowData = createElement(RowKind.DELETE);
DynamoDbWriteRequest actualWriteRequest = elementConverter.apply(rowData, context);
// In case of DELETE, a set of Primary Key(s) is required.
DynamoDbWriteRequest actualWriteRequest = elementConverterWithPK.apply(rowData, context);
DynamoDbWriteRequest expectedWriterequest =
DynamoDbWriteRequest.builder()
.setType(DynamoDbWriteRequestType.DELETE)
.setItem(rowDataToAttributeValueConverter.convertRowData(rowData))
.setItem(rowDataToAttributeValueConverterWithPK.convertRowData(rowData))
.build();

assertThat(actualWriteRequest).usingRecursiveComparison().isEqualTo(expectedWriterequest);
Expand All @@ -106,7 +115,7 @@ void testAttributeConverterReinitializedAfterSerialization()
throws IOException, ClassNotFoundException {
RowData rowData = createElement(RowKind.INSERT);

RowDataElementConverter originalConverter = new RowDataElementConverter(DATA_TYPE);
RowDataElementConverter originalConverter = new RowDataElementConverter(DATA_TYPE, null);
RowDataElementConverter transformedConverter =
InstantiationUtil.deserializeObject(
InstantiationUtil.serializeObject(originalConverter),
Expand Down
Loading
Loading