You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by da...@apache.org on 2022/11/22 09:44:45 UTC
[flink-connector-aws] branch main updated: [FLINK-29900][Connectors/DynamoDB] Implement Table API for DynamoDB sink
This is an automated email from the ASF dual-hosted git repository.
dannycranmer pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-aws.git
The following commit(s) were added to refs/heads/main by this push:
new b2606b6 [FLINK-29900][Connectors/DynamoDB] Implement Table API for DynamoDB sink
b2606b6 is described below
commit b2606b634df44c1bc412e72c9494842703052998
Author: Hong Liang Teoh <li...@amazon.com>
AuthorDate: Sun Nov 6 21:24:55 2022 +0000
[FLINK-29900][Connectors/DynamoDB] Implement Table API for DynamoDB sink
---
flink-connector-dynamodb/pom.xml | 44 +-
.../dynamodb/table/DynamoDbConfiguration.java | 44 ++
.../dynamodb/table/DynamoDbConnectorOptions.java | 52 ++
.../dynamodb/table/DynamoDbDynamicSink.java | 182 +++++++
.../dynamodb/table/DynamoDbDynamicSinkFactory.java | 84 +++
.../dynamodb/table/RowDataElementConverter.java | 73 +++
.../table/RowDataToAttributeValueConverter.java | 104 ++++
.../table/converter/ArrayAttributeConverter.java | 69 +++
.../converter/ArrayAttributeConverterProvider.java | 82 +++
.../org.apache.flink.table.factories.Factory | 16 +
.../dynamodb/sink/examples/example_dynamodb.sql | 67 +++
.../example_dynamodb_static_partitions.sql | 67 +++
.../table/DynamoDbDynamicSinkFactoryTest.java | 295 +++++++++++
.../table/RowDataElementConverterTest.java | 148 ++++++
.../RowDataToAttributeValueConverterTest.java | 561 +++++++++++++++++++++
.../converter/ArrayAttributeConverterTest.java | 90 ++++
flink-sql-connector-dynamodb/pom.xml | 1 +
.../src/main/resources/META-INF/NOTICE | 25 +-
.../services/dynamodb/execution.interceptors | 1 +
19 files changed, 1984 insertions(+), 21 deletions(-)
diff --git a/flink-connector-dynamodb/pom.xml b/flink-connector-dynamodb/pom.xml
index 25f7756..932f069 100644
--- a/flink-connector-dynamodb/pom.xml
+++ b/flink-connector-dynamodb/pom.xml
@@ -18,19 +18,19 @@ specific language governing permissions and limitations
under the License.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
- <modelVersion>4.0.0</modelVersion>
+ <modelVersion>4.0.0</modelVersion>
- <parent>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-aws-parent</artifactId>
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-aws-parent</artifactId>
<version>1.0.0-SNAPSHOT</version>
- </parent>
+ </parent>
- <artifactId>flink-connector-dynamodb</artifactId>
- <name>Flink : Connectors : Amazon DynamoDB</name>
+ <artifactId>flink-connector-dynamodb</artifactId>
+ <name>Flink : Connectors : Amazon DynamoDB</name>
<packaging>jar</packaging>
@@ -76,6 +76,22 @@ under the License.
<artifactId>dynamodb-enhanced</artifactId>
</dependency>
+ <!-- Table ecosystem -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-api-java-bridge</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ <optional>true</optional>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-runtime</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ <optional>true</optional>
+ </dependency>
+
<!-- Test dependencies -->
<dependency>
<groupId>org.apache.flink</groupId>
@@ -106,5 +122,15 @@ under the License.
<version>${flink.version}</version>
<scope>test</scope>
</dependency>
+
+ <!-- Table API test dependencies -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-common</artifactId>
+ <version>${flink.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
</dependencies>
</project>
diff --git a/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbConfiguration.java b/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbConfiguration.java
new file mode 100644
index 0000000..ad4ca3c
--- /dev/null
+++ b/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbConfiguration.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.dynamodb.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ReadableConfig;
+
+import static org.apache.flink.connector.dynamodb.table.DynamoDbConnectorOptions.FAIL_ON_ERROR;
+import static org.apache.flink.connector.dynamodb.table.DynamoDbConnectorOptions.TABLE_NAME;
+
+/** DynamoDb specific configuration. */
+@Internal
+public class DynamoDbConfiguration {
+
+ private final ReadableConfig config;
+
+ public DynamoDbConfiguration(ReadableConfig config) {
+ this.config = config;
+ }
+
+ public String getTableName() {
+ return config.get(TABLE_NAME);
+ }
+
+ public boolean getFailOnError() {
+ return config.get(FAIL_ON_ERROR);
+ }
+}
diff --git a/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbConnectorOptions.java b/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbConnectorOptions.java
new file mode 100644
index 0000000..899b690
--- /dev/null
+++ b/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbConnectorOptions.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.dynamodb.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.table.api.TableDescriptor;
+
+/** DynamoDb connector options. Made public for {@link TableDescriptor} to access it. */
+@PublicEvolving
+public class DynamoDbConnectorOptions {
+
+ public static final ConfigOption<String> TABLE_NAME =
+ ConfigOptions.key("table-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Name of destination DynamoDB table.");
+
+ public static final ConfigOption<String> AWS_REGION =
+ ConfigOptions.key("aws.region")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("AWS region for the destination DynamoDB table.");
+
+ public static final ConfigOption<Boolean> FAIL_ON_ERROR =
+ ConfigOptions.key("sink.fail-on-error")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Determines whether an exception should fail the job, otherwise failed requests are retried.");
+
+ private DynamoDbConnectorOptions() {
+ // private constructor to prevent initialization of static class
+ }
+}
diff --git a/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSink.java b/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSink.java
new file mode 100644
index 0000000..d79cd78
--- /dev/null
+++ b/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSink.java
@@ -0,0 +1,182 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.dynamodb.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.base.table.sink.AsyncDynamicTableSink;
+import org.apache.flink.connector.base.table.sink.AsyncDynamicTableSinkBuilder;
+import org.apache.flink.connector.dynamodb.sink.DynamoDbSink;
+import org.apache.flink.connector.dynamodb.sink.DynamoDbSinkBuilder;
+import org.apache.flink.connector.dynamodb.sink.DynamoDbWriteRequest;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.SinkV2Provider;
+import org.apache.flink.table.connector.sink.abilities.SupportsPartitioning;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+/**
+ * A {@link DynamicTableSink} that describes how to create a {@link DynamoDbSink} from a logical
+ * description.
+ */
+@Internal
+public class DynamoDbDynamicSink extends AsyncDynamicTableSink<DynamoDbWriteRequest>
+ implements SupportsPartitioning {
+
+ private final String tableName;
+ private final boolean failOnError;
+ private final Properties dynamoDbClientProperties;
+ private final DataType physicalDataType;
+ private final Set<String> overwriteByPartitionKeys;
+
+ protected DynamoDbDynamicSink(
+ @Nullable Integer maxBatchSize,
+ @Nullable Integer maxInFlightRequests,
+ @Nullable Integer maxBufferedRequests,
+ @Nullable Long maxBufferSizeInBytes,
+ @Nullable Long maxTimeInBufferMS,
+ String tableName,
+ boolean failOnError,
+ Properties dynamoDbClientProperties,
+ DataType physicalDataType,
+ Set<String> overwriteByPartitionKeys) {
+ super(
+ maxBatchSize,
+ maxInFlightRequests,
+ maxBufferedRequests,
+ maxBufferSizeInBytes,
+ maxTimeInBufferMS);
+ this.tableName = tableName;
+ this.failOnError = failOnError;
+ this.dynamoDbClientProperties = dynamoDbClientProperties;
+ this.physicalDataType = physicalDataType;
+ this.overwriteByPartitionKeys = overwriteByPartitionKeys;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode(ChangelogMode changelogMode) {
+ return ChangelogMode.upsert();
+ }
+
+ @Override
+ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
+ DynamoDbSinkBuilder<RowData> builder =
+ DynamoDbSink.<RowData>builder()
+ .setTableName(tableName)
+ .setFailOnError(failOnError)
+ .setOverwriteByPartitionKeys(new ArrayList<>(overwriteByPartitionKeys))
+ .setDynamoDbProperties(dynamoDbClientProperties)
+ .setElementConverter(new RowDataElementConverter(physicalDataType));
+
+ addAsyncOptionsToSinkBuilder(builder);
+
+ return SinkV2Provider.of(builder.build());
+ }
+
+ @Override
+ public DynamicTableSink copy() {
+ return new DynamoDbDynamicSink(
+ maxBatchSize,
+ maxInFlightRequests,
+ maxBufferedRequests,
+ maxBufferSizeInBytes,
+ maxTimeInBufferMS,
+ tableName,
+ failOnError,
+ dynamoDbClientProperties,
+ physicalDataType,
+ overwriteByPartitionKeys);
+ }
+
+ @Override
+ public String asSummaryString() {
+ return "DynamoDB";
+ }
+
+ @Override
+ public void applyStaticPartition(Map<String, String> partitions) {
+ // We don't need to do anything here because the DynamoDB sink handles a static partition
+ // just like a normal partition.
+ }
+
+ public static DynamoDbDynamicTableSinkBuilder builder() {
+ return new DynamoDbDynamicTableSinkBuilder();
+ }
+
+ /** Builder class for {@link DynamoDbDynamicSink}. */
+ @Internal
+ public static class DynamoDbDynamicTableSinkBuilder
+ extends AsyncDynamicTableSinkBuilder<
+ DynamoDbWriteRequest, DynamoDbDynamicTableSinkBuilder> {
+ private String tableName;
+ private boolean failOnError;
+ private Properties dynamoDbClientProperties;
+ private DataType physicalDataType;
+ private Set<String> overwriteByPartitionKeys;
+
+ public DynamoDbDynamicTableSinkBuilder setTableName(String tableName) {
+ this.tableName = tableName;
+ return this;
+ }
+
+ public DynamoDbDynamicTableSinkBuilder setFailOnError(boolean failOnError) {
+ this.failOnError = failOnError;
+ return this;
+ }
+
+ public DynamoDbDynamicTableSinkBuilder setDynamoDbClientProperties(
+ Properties dynamoDbClientProperties) {
+ this.dynamoDbClientProperties = dynamoDbClientProperties;
+ return this;
+ }
+
+ public DynamoDbDynamicTableSinkBuilder setPhysicalDataType(DataType physicalDataType) {
+ this.physicalDataType = physicalDataType;
+ return this;
+ }
+
+ public DynamoDbDynamicTableSinkBuilder setOverwriteByPartitionKeys(
+ Set<String> overwriteByPartitionKeys) {
+ this.overwriteByPartitionKeys = overwriteByPartitionKeys;
+ return this;
+ }
+
+ @Override
+ public AsyncDynamicTableSink<DynamoDbWriteRequest> build() {
+ return new DynamoDbDynamicSink(
+ getMaxBatchSize(),
+ getMaxInFlightRequests(),
+ getMaxBufferedRequests(),
+ getMaxBufferSizeInBytes(),
+ getMaxTimeInBufferMS(),
+ tableName,
+ failOnError,
+ dynamoDbClientProperties,
+ physicalDataType,
+ overwriteByPartitionKeys);
+ }
+ }
+}
diff --git a/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java b/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java
new file mode 100644
index 0000000..8072a39
--- /dev/null
+++ b/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.dynamodb.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.connector.base.table.AsyncDynamicTableSinkFactory;
+import org.apache.flink.connector.base.table.sink.options.AsyncSinkConfigurationValidator;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.factories.FactoryUtil;
+
+import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableSet;
+
+import java.util.HashSet;
+import java.util.Properties;
+import java.util.Set;
+
+import static org.apache.flink.connector.dynamodb.table.DynamoDbConnectorOptions.AWS_REGION;
+import static org.apache.flink.connector.dynamodb.table.DynamoDbConnectorOptions.TABLE_NAME;
+
+/** Factory for creating {@link DynamoDbDynamicSink}. */
+@Internal
+public class DynamoDbDynamicSinkFactory extends AsyncDynamicTableSinkFactory {
+
+ public static final String FACTORY_IDENTIFIER = "dynamodb";
+
+ @Override
+ public DynamicTableSink createDynamicTableSink(Context context) {
+ FactoryUtil.TableFactoryHelper factoryHelper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+ ResolvedCatalogTable catalogTable = context.getCatalogTable();
+
+ FactoryUtil.validateFactoryOptions(this, factoryHelper.getOptions());
+
+ DynamoDbConfiguration dynamoDbConfiguration =
+ new DynamoDbConfiguration(factoryHelper.getOptions());
+
+ Properties tableProperties = new Properties();
+ tableProperties.putAll(catalogTable.getOptions());
+
+ DynamoDbDynamicSink.DynamoDbDynamicTableSinkBuilder builder =
+ DynamoDbDynamicSink.builder()
+ .setTableName(dynamoDbConfiguration.getTableName())
+ .setFailOnError(dynamoDbConfiguration.getFailOnError())
+ .setPhysicalDataType(
+ catalogTable.getResolvedSchema().toPhysicalRowDataType())
+ .setOverwriteByPartitionKeys(new HashSet<>(catalogTable.getPartitionKeys()))
+ .setDynamoDbClientProperties(tableProperties);
+
+ AsyncSinkConfigurationValidator asyncSinkConfigurationValidator =
+ new AsyncSinkConfigurationValidator(factoryHelper.getOptions());
+ addAsyncOptionsToBuilder(
+ asyncSinkConfigurationValidator.getValidatedConfigurations(), builder);
+
+ return builder.build();
+ }
+
+ @Override
+ public String factoryIdentifier() {
+ return FACTORY_IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return ImmutableSet.of(TABLE_NAME, AWS_REGION);
+ }
+}
diff --git a/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataElementConverter.java b/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataElementConverter.java
new file mode 100644
index 0000000..d0c4c90
--- /dev/null
+++ b/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataElementConverter.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.dynamodb.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.dynamodb.sink.DynamoDbWriteRequest;
+import org.apache.flink.connector.dynamodb.sink.DynamoDbWriteRequestType;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+
+/**
+ * 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
+ * the DynamoDb sink.
+ */
+@Internal
+public class RowDataElementConverter implements ElementConverter<RowData, DynamoDbWriteRequest> {
+
+ private final DataType physicalDataType;
+ private transient RowDataToAttributeValueConverter rowDataToAttributeValueConverter;
+
+ public RowDataElementConverter(DataType physicalDataType) {
+ this.physicalDataType = physicalDataType;
+ this.rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(physicalDataType);
+ }
+
+ @Override
+ public DynamoDbWriteRequest apply(RowData element, SinkWriter.Context context) {
+ if (rowDataToAttributeValueConverter == null) {
+ rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(physicalDataType);
+ }
+
+ DynamoDbWriteRequest.Builder builder =
+ DynamoDbWriteRequest.builder()
+ .setItem(rowDataToAttributeValueConverter.convertRowData(element));
+
+ switch (element.getRowKind()) {
+ case INSERT:
+ case UPDATE_AFTER:
+ builder.setType(DynamoDbWriteRequestType.PUT);
+ break;
+ case DELETE:
+ builder.setType(DynamoDbWriteRequestType.DELETE);
+ break;
+ case UPDATE_BEFORE:
+ default:
+ throw new TableException("Unsupported message kind: " + element.getRowKind());
+ }
+
+ return builder.build();
+ }
+}
diff --git a/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverter.java b/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverter.java
new file mode 100644
index 0000000..1b7a05e
--- /dev/null
+++ b/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverter.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.dynamodb.table;
+
+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.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 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.List;
+import java.util.Map;
+
+import static org.apache.flink.table.data.RowData.createFieldGetter;
+
+/** Converts from Flink Table API internal type of {@link RowData} to {@link AttributeValue}. */
+@Internal
+public class RowDataToAttributeValueConverter {
+
+ private final DataType physicalDataType;
+ private final TableSchema<RowData> tableSchema;
+
+ public RowDataToAttributeValueConverter(DataType physicalDataType) {
+ this.physicalDataType = physicalDataType;
+ this.tableSchema = createTableSchema();
+ }
+
+ public Map<String, AttributeValue> convertRowData(RowData row) {
+ return tableSchema.itemToMap(row, false);
+ }
+
+ private StaticTableSchema<RowData> createTableSchema() {
+ List<DataTypes.Field> fields = DataType.getFields(physicalDataType);
+ StaticTableSchema.Builder<RowData> builder = TableSchema.builder(RowData.class);
+
+ AttributeConverterProvider newAttributeConverterProvider =
+ new ArrayAttributeConverterProvider();
+ builder.attributeConverterProviders(
+ newAttributeConverterProvider, AttributeConverterProvider.defaultProvider());
+
+ for (int i = 0; i < fields.size(); i++) {
+ DataTypes.Field field = fields.get(i);
+ RowData.FieldGetter fieldGetter =
+ createFieldGetter(field.getDataType().getLogicalType(), i);
+
+ builder = addAttribute(builder, field, fieldGetter);
+ }
+ return builder.build();
+ }
+
+ private StaticTableSchema.Builder<RowData> addAttribute(
+ StaticTableSchema.Builder<RowData> builder,
+ DataTypes.Field field,
+ RowData.FieldGetter fieldGetter) {
+
+ return builder.addAttribute(
+ getEnhancedType(field.getDataType()),
+ a ->
+ a.name(field.getName())
+ .getter(
+ rowData ->
+ DataStructureConverters.getConverter(
+ field.getDataType())
+ .toExternal(
+ fieldGetter.getFieldOrNull(
+ rowData)))
+ .setter(((rowData, t) -> {})));
+ }
+
+ private <T> EnhancedType<T> getEnhancedType(DataType dataType) {
+ if (dataType instanceof KeyValueDataType) {
+ return (EnhancedType<T>)
+ EnhancedType.mapOf(
+ getEnhancedType(((KeyValueDataType) dataType).getKeyDataType()),
+ getEnhancedType(((KeyValueDataType) dataType).getValueDataType()));
+ } else {
+ return (EnhancedType<T>) EnhancedType.of(dataType.getConversionClass());
+ }
+ }
+}
diff --git a/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/converter/ArrayAttributeConverter.java b/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/converter/ArrayAttributeConverter.java
new file mode 100644
index 0000000..004e396
--- /dev/null
+++ b/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/converter/ArrayAttributeConverter.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.dynamodb.table.converter;
+
+import org.apache.flink.annotation.Internal;
+
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverter;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeValueType;
+import software.amazon.awssdk.enhanced.dynamodb.EnhancedType;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.util.Arrays;
+import java.util.stream.Collectors;
+
+/** A converter between T[] and {@link AttributeValue}. */
+@Internal
+public class ArrayAttributeConverter<T> implements AttributeConverter<T[]> {
+
+ private final AttributeConverter<T> tAttributeConverter;
+ private final EnhancedType<T[]> enhancedType;
+
+ public ArrayAttributeConverter(
+ AttributeConverter<T> tAttributeConverter, EnhancedType<T[]> enhancedType) {
+ this.tAttributeConverter = tAttributeConverter;
+ this.enhancedType = enhancedType;
+ }
+
+ @Override
+ public AttributeValue transformFrom(T[] input) {
+ return AttributeValue.builder()
+ .l(
+ Arrays.stream(input)
+ .map(tAttributeConverter::transformFrom)
+ .collect(Collectors.toList()))
+ .build();
+ }
+
+ @Override
+ public T[] transformTo(AttributeValue input) {
+ throw new UnsupportedOperationException(
+ "transformTo() is unsupported because the DynamoDB sink does not need it.");
+ }
+
+ @Override
+ public EnhancedType<T[]> type() {
+ return enhancedType;
+ }
+
+ @Override
+ public AttributeValueType attributeValueType() {
+ return AttributeValueType.L;
+ }
+}
diff --git a/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/converter/ArrayAttributeConverterProvider.java b/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/converter/ArrayAttributeConverterProvider.java
new file mode 100644
index 0000000..e219357
--- /dev/null
+++ b/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/converter/ArrayAttributeConverterProvider.java
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.dynamodb.table.converter;
+
+import org.apache.flink.annotation.Internal;
+
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverter;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverterProvider;
+import software.amazon.awssdk.enhanced.dynamodb.EnhancedType;
+
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+
+/** Attribute converter provider for String Array. */
+@Internal
+public class ArrayAttributeConverterProvider implements AttributeConverterProvider {
+
+ private static final AttributeConverterProvider defaultAttributeConverterProvider =
+ AttributeConverterProvider.defaultProvider();
+
+ @Override
+ public <T> AttributeConverter<T> converterFor(EnhancedType<T> enhancedType) {
+ if (enhancedType.equals(EnhancedType.of(String[].class))) {
+ return (AttributeConverter<T>) getArrayAttributeConverter(String.class, enhancedType);
+ } else if (enhancedType.equals(EnhancedType.of(Boolean[].class))) {
+ return (AttributeConverter<T>) getArrayAttributeConverter(Boolean.class, enhancedType);
+ } else if (enhancedType.equals(EnhancedType.of(BigDecimal[].class))) {
+ return (AttributeConverter<T>)
+ getArrayAttributeConverter(BigDecimal.class, enhancedType);
+ } else if (enhancedType.equals(EnhancedType.of(Byte[].class))) {
+ return (AttributeConverter<T>) getArrayAttributeConverter(Byte.class, enhancedType);
+ } else if (enhancedType.equals(EnhancedType.of(Double[].class))) {
+ return (AttributeConverter<T>) getArrayAttributeConverter(Double.class, enhancedType);
+ } else if (enhancedType.equals(EnhancedType.of(Short[].class))) {
+ return (AttributeConverter<T>) getArrayAttributeConverter(Short.class, enhancedType);
+ } else if (enhancedType.equals(EnhancedType.of(Integer[].class))) {
+ return (AttributeConverter<T>) getArrayAttributeConverter(Integer.class, enhancedType);
+ } else if (enhancedType.equals(EnhancedType.of(Long[].class))) {
+ return (AttributeConverter<T>) getArrayAttributeConverter(Long.class, enhancedType);
+ } else if (enhancedType.equals(EnhancedType.of(Float[].class))) {
+ return (AttributeConverter<T>) getArrayAttributeConverter(Float.class, enhancedType);
+ } else if (enhancedType.equals(EnhancedType.of(LocalDate[].class))) {
+ return (AttributeConverter<T>)
+ getArrayAttributeConverter(LocalDate.class, enhancedType);
+ } else if (enhancedType.equals(EnhancedType.of(LocalTime[].class))) {
+ return (AttributeConverter<T>)
+ getArrayAttributeConverter(LocalTime.class, enhancedType);
+ } else if (enhancedType.equals(EnhancedType.of(LocalDateTime[].class))) {
+ return (AttributeConverter<T>)
+ getArrayAttributeConverter(LocalDateTime.class, enhancedType);
+ } else if (enhancedType.equals(EnhancedType.of(Instant[].class))) {
+ return (AttributeConverter<T>) getArrayAttributeConverter(Instant.class, enhancedType);
+ }
+ return null;
+ }
+
+ private <R, T> ArrayAttributeConverter<R> getArrayAttributeConverter(
+ Class<R> clazz, EnhancedType<T> enhancedType) {
+ return new ArrayAttributeConverter<>(
+ defaultAttributeConverterProvider.converterFor(EnhancedType.of(clazz)),
+ (EnhancedType<R[]>) enhancedType);
+ }
+}
diff --git a/flink-connector-dynamodb/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory b/flink-connector-dynamodb/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
new file mode 100644
index 0000000..b9d00c8
--- /dev/null
+++ b/flink-connector-dynamodb/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+org.apache.flink.connector.dynamodb.table.DynamoDbDynamicSinkFactory
diff --git a/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/examples/example_dynamodb.sql b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/examples/example_dynamodb.sql
new file mode 100644
index 0000000..ab011af
--- /dev/null
+++ b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/examples/example_dynamodb.sql
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+CREATE TABLE dynamo_db_table (
+ `partition_key` STRING,
+ `sort_key` STRING,
+ `some_char` CHAR,
+ `some_varchar` VARCHAR,
+ `some_string` STRING,
+ `some_boolean` BOOLEAN,
+ `some_decimal` DECIMAL,
+ `some_tinyint` TINYINT,
+ `some_smallint` SMALLINT,
+ `some_int` INT,
+ `some_bigint` BIGINT,
+ `some_float` FLOAT,
+ `some_date` DATE,
+ `some_time` TIME,
+ `some_timestamp` TIMESTAMP(3),
+ `some_timestamp_ltz` TIMESTAMP_LTZ(5),
+ `some_char_array` ARRAY<CHAR>,
+ `some_varchar_array` ARRAY<VARCHAR>,
+ `some_string_array` ARRAY<STRING>,
+ `some_boolean_array` ARRAY<BOOLEAN>,
+ `some_decimal_array` ARRAY<DECIMAL>,
+ `some_tinyint_array` ARRAY<TINYINT>,
+ `some_smallint_array` ARRAY<SMALLINT>,
+ `some_int_array` ARRAY<INT>,
+ `some_bigint_array` ARRAY<BIGINT>,
+ `some_float_array` ARRAY<FLOAT>,
+ `some_date_array` ARRAY<DATE>,
+ `some_time_array` ARRAY<TIME>,
+ `some_timestamp_array` ARRAY<TIMESTAMP(3)>,
+ `some_timestamp_ltz_array` ARRAY<TIMESTAMP_LTZ(5)>,
+ `some_string_map` MAP<STRING,STRING>,
+ `some_boolean_map` MAP<STRING,BOOLEAN>
+) PARTITIONED BY ( partition_key )
+ WITH (
+ 'connector' = 'dynamodb',
+ 'table-name' = 'DynamoDBSinkWithKeys',
+ 'aws.region' = 'us-east-1'
+ );
+
+-- create a bounded mock table
+CREATE TEMPORARY TABLE datagen
+WITH (
+ 'connector' = 'datagen',
+ 'number-of-rows' = '50'
+)
+LIKE dynamo_db_table (EXCLUDING ALL);
+
+INSERT INTO dynamo_db_table SELECT * from datagen;
\ No newline at end of file
diff --git a/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/examples/example_dynamodb_static_partitions.sql b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/examples/example_dynamodb_static_partitions.sql
new file mode 100644
index 0000000..964dbdb
--- /dev/null
+++ b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/examples/example_dynamodb_static_partitions.sql
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+CREATE TABLE dynamo_db_table (
+ `partition_key` STRING,
+ `sort_key` STRING,
+ `some_char` CHAR,
+ `some_varchar` VARCHAR,
+ `some_string` STRING,
+ `some_boolean` BOOLEAN,
+ `some_decimal` DECIMAL,
+ `some_tinyint` TINYINT,
+ `some_smallint` SMALLINT,
+ `some_int` INT,
+ `some_bigint` BIGINT,
+ `some_float` FLOAT,
+ `some_date` DATE,
+ `some_time` TIME,
+ `some_timestamp` TIMESTAMP(3),
+ `some_timestamp_ltz` TIMESTAMP_LTZ(5),
+ `some_char_array` ARRAY<CHAR>,
+ `some_varchar_array` ARRAY<VARCHAR>,
+ `some_string_array` ARRAY<STRING>,
+ `some_boolean_array` ARRAY<BOOLEAN>,
+ `some_decimal_array` ARRAY<DECIMAL>,
+ `some_tinyint_array` ARRAY<TINYINT>,
+ `some_smallint_array` ARRAY<SMALLINT>,
+ `some_int_array` ARRAY<INT>,
+ `some_bigint_array` ARRAY<BIGINT>,
+ `some_float_array` ARRAY<FLOAT>,
+ `some_date_array` ARRAY<DATE>,
+ `some_time_array` ARRAY<TIME>,
+ `some_timestamp_array` ARRAY<TIMESTAMP(3)>,
+ `some_timestamp_ltz_array` ARRAY<TIMESTAMP_LTZ(5)>,
+ `some_string_map` MAP<STRING,STRING>,
+ `some_boolean_map` MAP<STRING,BOOLEAN>
+) PARTITIONED BY ( partition_key, sort_key )
+ WITH (
+ 'connector' = 'dynamodb',
+ 'table-name' = 'DynamoDBSinkWithKeys',
+ 'aws.region' = 'us-east-1'
+ );
+
+-- create a bounded mock table
+CREATE TEMPORARY TABLE datagen
+WITH (
+ 'connector' = 'datagen',
+ 'number-of-rows' = '50'
+)
+LIKE dynamo_db_table (EXCLUDING ALL);
+
+INSERT INTO dynamo_db_table PARTITION (partition_key='123',sort_key='345') SELECT `some_char`,`some_varchar`,`some_string`,`some_boolean`,`some_decimal`,`some_tinyint`,`some_smallint`,`some_int`,`some_bigint`,`some_float`,`some_date`,`some_time`,`some_timestamp`,`some_timestamp_ltz`,`some_char_array`,`some_varchar_array`,`some_string_array`,`some_boolean_array`,`some_decimal_array`,`some_tinyint_array`,`some_smallint_array`,`some_int_array`,`some_bigint_array`,`some_float_array`,`some_da [...]
\ No newline at end of file
diff --git a/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactoryTest.java b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactoryTest.java
new file mode 100644
index 0000000..df146ac
--- /dev/null
+++ b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactoryTest.java
@@ -0,0 +1,295 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.dynamodb.table;
+
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.connector.dynamodb.sink.DynamoDbSink;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.SinkV2Provider;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.TableOptionsBuilder;
+import org.apache.flink.table.factories.TestFormatFactory;
+import org.apache.flink.table.runtime.connector.sink.SinkRuntimeProviderContext;
+
+import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableMap;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.apache.flink.connector.base.table.AsyncSinkConnectorOptions.FLUSH_BUFFER_SIZE;
+import static org.apache.flink.connector.base.table.AsyncSinkConnectorOptions.FLUSH_BUFFER_TIMEOUT;
+import static org.apache.flink.connector.base.table.AsyncSinkConnectorOptions.MAX_BATCH_SIZE;
+import static org.apache.flink.connector.base.table.AsyncSinkConnectorOptions.MAX_BUFFERED_REQUESTS;
+import static org.apache.flink.connector.base.table.AsyncSinkConnectorOptions.MAX_IN_FLIGHT_REQUESTS;
+import static org.apache.flink.connector.dynamodb.table.DynamoDbConnectorOptions.AWS_REGION;
+import static org.apache.flink.connector.dynamodb.table.DynamoDbConnectorOptions.FAIL_ON_ERROR;
+import static org.apache.flink.connector.dynamodb.table.DynamoDbConnectorOptions.TABLE_NAME;
+import static org.apache.flink.table.factories.utils.FactoryMocks.createTableSink;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+
+/** Test for {@link DynamoDbDynamicSink} created by {@link DynamoDbDynamicSinkFactory}. */
+public class DynamoDbDynamicSinkFactoryTest {
+
+ private static final String DYNAMO_DB_TABLE_NAME = "TestDynamoDBTable";
+
+ @Test
+ void testGoodPartitionedTableSink() {
+ ResolvedSchema sinkSchema = createResolvedSchemaUsingAllDataTypes();
+
+ Map<String, String> sinkOptions = defaultSinkOptions().build();
+ List<String> partitionKeys = Collections.singletonList("partition_key");
+
+ // Construct actual sink
+ DynamoDbDynamicSink actualSink =
+ (DynamoDbDynamicSink) createTableSink(sinkSchema, partitionKeys, sinkOptions);
+
+ // Construct expected sink
+ Properties dynamoDbClientProperties = new Properties();
+ dynamoDbClientProperties.putAll(sinkOptions);
+ DynamoDbDynamicSink expectedSink =
+ (DynamoDbDynamicSink)
+ DynamoDbDynamicSink.builder()
+ .setTableName(DYNAMO_DB_TABLE_NAME)
+ .setOverwriteByPartitionKeys(new HashSet<>(partitionKeys))
+ .setDynamoDbClientProperties(dynamoDbClientProperties)
+ .setPhysicalDataType(sinkSchema.toPhysicalRowDataType())
+ .build();
+
+ assertThat(actualSink).usingRecursiveComparison().isEqualTo(expectedSink);
+ assertThat(actualSink.getChangelogMode(ChangelogMode.insertOnly()))
+ .isEqualTo(ChangelogMode.upsert());
+ assertThat(actualSink.asSummaryString()).isEqualTo("DynamoDB");
+
+ Sink<RowData> createdSink =
+ ((SinkV2Provider)
+ actualSink.getSinkRuntimeProvider(
+ new SinkRuntimeProviderContext(false)))
+ .createSink();
+ assertThat(createdSink).isInstanceOf(DynamoDbSink.class);
+ }
+
+ @Test
+ void testGoodNonPartitionedTableSink() {
+ ResolvedSchema sinkSchema = defaultSinkSchema();
+ Map<String, String> sinkOptions = defaultSinkOptions().build();
+
+ // Construct actual sink
+ DynamoDbDynamicSink actualSink =
+ (DynamoDbDynamicSink) createTableSink(sinkSchema, sinkOptions);
+
+ // Construct expected sink
+ Properties dynamoDbClientProperties = new Properties();
+ dynamoDbClientProperties.putAll(sinkOptions);
+ DynamoDbDynamicSink expectedSink =
+ (DynamoDbDynamicSink)
+ DynamoDbDynamicSink.builder()
+ .setTableName(DYNAMO_DB_TABLE_NAME)
+ .setOverwriteByPartitionKeys(new HashSet<>())
+ .setDynamoDbClientProperties(dynamoDbClientProperties)
+ .setPhysicalDataType(sinkSchema.toPhysicalRowDataType())
+ .build();
+
+ assertThat(actualSink).usingRecursiveComparison().isEqualTo(expectedSink);
+ }
+
+ @Test
+ void testCopyTableSink() {
+ ResolvedSchema sinkSchema = defaultSinkSchema();
+ Map<String, String> sinkOptions = defaultSinkOptions().build();
+
+ // Construct expected sink
+ Properties dynamoDbClientProperties = new Properties();
+ dynamoDbClientProperties.putAll(sinkOptions);
+ DynamoDbDynamicSink originalSink =
+ (DynamoDbDynamicSink)
+ DynamoDbDynamicSink.builder()
+ .setTableName(DYNAMO_DB_TABLE_NAME)
+ .setOverwriteByPartitionKeys(new HashSet<>())
+ .setDynamoDbClientProperties(dynamoDbClientProperties)
+ .setPhysicalDataType(sinkSchema.toPhysicalRowDataType())
+ .build();
+
+ assertThat(originalSink).usingRecursiveComparison().isEqualTo(originalSink.copy());
+ }
+
+ @Test
+ void testGoodTableSinkWithOptionalOptions() {
+ ResolvedSchema sinkSchema = defaultSinkSchema();
+ Map<String, String> sinkOptions =
+ defaultSinkOptions().withTableOption(FAIL_ON_ERROR, "true").build();
+ List<String> partitionKeys = Collections.singletonList("partition_key");
+
+ // Construct actual sink
+ DynamoDbDynamicSink actualSink =
+ (DynamoDbDynamicSink) createTableSink(sinkSchema, partitionKeys, sinkOptions);
+
+ // Construct expected sink
+ Properties dynamoDbClientProperties = new Properties();
+ dynamoDbClientProperties.putAll(sinkOptions);
+ DynamoDbDynamicSink expectedSink =
+ (DynamoDbDynamicSink)
+ DynamoDbDynamicSink.builder()
+ .setTableName(DYNAMO_DB_TABLE_NAME)
+ .setOverwriteByPartitionKeys(new HashSet<>(partitionKeys))
+ .setDynamoDbClientProperties(dynamoDbClientProperties)
+ .setPhysicalDataType(sinkSchema.toPhysicalRowDataType())
+ .setFailOnError(true)
+ .build();
+
+ assertThat(actualSink).usingRecursiveComparison().isEqualTo(expectedSink);
+ }
+
+ @Test
+ void testGoodTableSinkWithAsyncProperties() {
+ ResolvedSchema sinkSchema = defaultSinkSchema();
+ Map<String, String> sinkOptions =
+ defaultSinkOptions()
+ .withTableOption(MAX_BATCH_SIZE, "100")
+ .withTableOption(MAX_IN_FLIGHT_REQUESTS, "100")
+ .withTableOption(MAX_BUFFERED_REQUESTS, "100")
+ .withTableOption(FLUSH_BUFFER_SIZE, "1024")
+ .withTableOption(FLUSH_BUFFER_TIMEOUT, "1000")
+ .build();
+ List<String> partitionKeys = Collections.singletonList("partition_key");
+
+ // Construct actual sink
+ DynamoDbDynamicSink actualSink =
+ (DynamoDbDynamicSink) createTableSink(sinkSchema, partitionKeys, sinkOptions);
+
+ // Construct expected sink
+ Properties dynamoDbClientProperties = new Properties();
+ dynamoDbClientProperties.putAll(sinkOptions);
+ DynamoDbDynamicSink expectedSink =
+ (DynamoDbDynamicSink)
+ DynamoDbDynamicSink.builder()
+ .setMaxBatchSize(100)
+ .setMaxInFlightRequests(100)
+ .setMaxBufferedRequests(100)
+ .setMaxBufferSizeInBytes(1024)
+ .setMaxTimeInBufferMS(1000)
+ .setTableName(DYNAMO_DB_TABLE_NAME)
+ .setOverwriteByPartitionKeys(new HashSet<>(partitionKeys))
+ .setDynamoDbClientProperties(dynamoDbClientProperties)
+ .setPhysicalDataType(sinkSchema.toPhysicalRowDataType())
+ .build();
+
+ assertThat(actualSink).usingRecursiveComparison().isEqualTo(expectedSink);
+ }
+
+ @Test
+ void testGoodTableSinkWithStaticPartitions() {
+ ResolvedSchema sinkSchema = defaultSinkSchema();
+ Map<String, String> sinkOptions = defaultSinkOptions().build();
+ List<String> partitionKeys = Collections.singletonList("partition_key");
+
+ // Construct actual sink
+ DynamoDbDynamicSink originalSink =
+ (DynamoDbDynamicSink) createTableSink(sinkSchema, partitionKeys, sinkOptions);
+
+ DynamoDbDynamicSink sinkWithStaticPartition =
+ (DynamoDbDynamicSink) createTableSink(sinkSchema, partitionKeys, sinkOptions);
+ sinkWithStaticPartition.applyStaticPartition(ImmutableMap.of("no_op_key", "no_op_value"));
+
+ // Verify no-op for applyStaticPartition
+ assertThat(sinkWithStaticPartition).usingRecursiveComparison().isEqualTo(originalSink);
+ }
+
+ @Test
+ void testBadTableSinkWithoutTableName() {
+ ResolvedSchema sinkSchema = defaultSinkSchema();
+ Map<String, String> sinkOptions =
+ new TableOptionsBuilder(
+ DynamoDbDynamicSinkFactory.FACTORY_IDENTIFIER,
+ TestFormatFactory.IDENTIFIER)
+ .withTableOption(AWS_REGION, "us-east-1")
+ .build();
+ List<String> partitionKeys = Collections.singletonList("partition_key");
+
+ assertThatExceptionOfType(ValidationException.class)
+ .isThrownBy(() -> createTableSink(sinkSchema, partitionKeys, sinkOptions))
+ .havingCause()
+ .withMessageContaining("One or more required options are missing.")
+ .withMessageContaining(TABLE_NAME.key());
+ }
+
+ @Test
+ void testBadTableSinkWithoutAwsRegion() {
+ ResolvedSchema sinkSchema = defaultSinkSchema();
+ Map<String, String> sinkOptions =
+ new TableOptionsBuilder(
+ DynamoDbDynamicSinkFactory.FACTORY_IDENTIFIER,
+ TestFormatFactory.IDENTIFIER)
+ .withTableOption(TABLE_NAME, DYNAMO_DB_TABLE_NAME)
+ .build();
+ List<String> partitionKeys = Collections.singletonList("partition_key");
+
+ assertThatExceptionOfType(ValidationException.class)
+ .isThrownBy(() -> createTableSink(sinkSchema, partitionKeys, sinkOptions))
+ .havingCause()
+ .withMessageContaining("One or more required options are missing.")
+ .withMessageContaining(AWS_REGION.key());
+ }
+
+ private ResolvedSchema createResolvedSchemaUsingAllDataTypes() {
+ return ResolvedSchema.of(
+ Column.physical("partition_key", DataTypes.STRING()),
+ Column.physical("sort_key", DataTypes.BIGINT()),
+ Column.physical("payload", DataTypes.STRING()),
+ Column.physical("some_char_array", DataTypes.ARRAY(DataTypes.CHAR(1))),
+ Column.physical("some_varchar_array", DataTypes.ARRAY(DataTypes.VARCHAR(1))),
+ Column.physical("some_string_array", DataTypes.ARRAY(DataTypes.STRING())),
+ Column.physical("some_boolean_array", DataTypes.ARRAY(DataTypes.BOOLEAN())),
+ Column.physical("some_decimal_array", DataTypes.ARRAY(DataTypes.DECIMAL(1, 1))),
+ Column.physical("some_tinyint_array", DataTypes.ARRAY(DataTypes.TINYINT())),
+ Column.physical("some_smallint_array", DataTypes.ARRAY(DataTypes.SMALLINT())),
+ Column.physical("some_int_array", DataTypes.ARRAY(DataTypes.INT())),
+ Column.physical("some_bigint_array", DataTypes.ARRAY(DataTypes.BIGINT())),
+ Column.physical("some_float_array", DataTypes.ARRAY(DataTypes.FLOAT())),
+ Column.physical("some_date_array", DataTypes.ARRAY(DataTypes.DATE())),
+ Column.physical("some_time_array", DataTypes.ARRAY(DataTypes.TIME())),
+ Column.physical("some_timestamp_array", DataTypes.ARRAY(DataTypes.TIMESTAMP())),
+ Column.physical(
+ "some_timestamp_ltz_array", DataTypes.ARRAY(DataTypes.TIMESTAMP_LTZ())),
+ Column.physical("some_map", DataTypes.MAP(DataTypes.STRING(), DataTypes.BIGINT())));
+ }
+
+ private ResolvedSchema defaultSinkSchema() {
+ return ResolvedSchema.of(
+ Column.physical("partition_key", DataTypes.STRING()),
+ Column.physical("sort_key", DataTypes.BIGINT()),
+ Column.physical("payload", DataTypes.STRING()));
+ }
+
+ private TableOptionsBuilder defaultSinkOptions() {
+ return new TableOptionsBuilder(
+ DynamoDbDynamicSinkFactory.FACTORY_IDENTIFIER, TestFormatFactory.IDENTIFIER)
+ .withTableOption(TABLE_NAME, DYNAMO_DB_TABLE_NAME)
+ .withTableOption("aws.region", "us-east-1");
+ }
+}
diff --git a/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataElementConverterTest.java b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataElementConverterTest.java
new file mode 100644
index 0000000..8ec38f2
--- /dev/null
+++ b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataElementConverterTest.java
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.dynamodb.table;
+
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.connector.dynamodb.sink.DynamoDbWriteRequest;
+import org.apache.flink.connector.dynamodb.sink.DynamoDbWriteRequestType;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.types.RowKind;
+import org.apache.flink.util.InstantiationUtil;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+
+/** Test for {@link RowDataElementConverter}. */
+public class RowDataElementConverterTest {
+
+ private static final DataType DATA_TYPE =
+ DataTypes.ROW(
+ DataTypes.FIELD("partition_key", DataTypes.STRING()),
+ DataTypes.FIELD("payload", DataTypes.STRING()));
+ private static final RowDataElementConverter elementConverter =
+ new RowDataElementConverter(DATA_TYPE);
+ private static final SinkWriter.Context context = new UnusedSinkWriterContext();
+ private static final RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(DATA_TYPE);
+
+ @Test
+ void testInsert() {
+ RowData rowData = createElement(RowKind.INSERT);
+ DynamoDbWriteRequest actualWriteRequest = elementConverter.apply(rowData, context);
+ DynamoDbWriteRequest expectedWriterequest =
+ DynamoDbWriteRequest.builder()
+ .setType(DynamoDbWriteRequestType.PUT)
+ .setItem(rowDataToAttributeValueConverter.convertRowData(rowData))
+ .build();
+
+ assertThat(actualWriteRequest).isEqualTo(expectedWriterequest);
+ }
+
+ @Test
+ void testUpdateAfter() {
+ RowData rowData = createElement(RowKind.UPDATE_AFTER);
+ DynamoDbWriteRequest actualWriteRequest = elementConverter.apply(rowData, context);
+ DynamoDbWriteRequest expectedWriterequest =
+ DynamoDbWriteRequest.builder()
+ .setType(DynamoDbWriteRequestType.PUT)
+ .setItem(rowDataToAttributeValueConverter.convertRowData(rowData))
+ .build();
+
+ assertThat(actualWriteRequest).isEqualTo(expectedWriterequest);
+ }
+
+ @Test
+ void testUpdateBeforeIsUnsupported() {
+ // UPDATE_BEFORE only makes sense in tables that do not have a uniquely identifiable index
+ // for each row.
+ // DynamoDB requires a partition key to be specified, so we do not support UPDATE_BEFORE
+ RowData rowData = createElement(RowKind.UPDATE_BEFORE);
+
+ assertThatExceptionOfType(TableException.class)
+ .isThrownBy(() -> elementConverter.apply(rowData, context))
+ .withMessageContaining("Unsupported message kind: UPDATE_BEFORE");
+ }
+
+ @Test
+ void testDelete() {
+ RowData rowData = createElement(RowKind.DELETE);
+ DynamoDbWriteRequest actualWriteRequest = elementConverter.apply(rowData, context);
+ DynamoDbWriteRequest expectedWriterequest =
+ DynamoDbWriteRequest.builder()
+ .setType(DynamoDbWriteRequestType.DELETE)
+ .setItem(rowDataToAttributeValueConverter.convertRowData(rowData))
+ .build();
+
+ assertThat(actualWriteRequest).isEqualTo(expectedWriterequest);
+ }
+
+ @Test
+ void testAttributeConverterReinitializedAfterSerialization()
+ throws IOException, ClassNotFoundException {
+ RowData rowData = createElement(RowKind.INSERT);
+
+ RowDataElementConverter originalConverter = new RowDataElementConverter(DATA_TYPE);
+ RowDataElementConverter transformedConverter =
+ InstantiationUtil.deserializeObject(
+ InstantiationUtil.serializeObject(originalConverter),
+ this.getClass().getClassLoader());
+
+ assertThat(transformedConverter).extracting("rowDataToAttributeValueConverter").isNull();
+
+ DynamoDbWriteRequest actualWriteRequest = transformedConverter.apply(rowData, context);
+ DynamoDbWriteRequest expectedWriterequest =
+ DynamoDbWriteRequest.builder()
+ .setType(DynamoDbWriteRequestType.PUT)
+ .setItem(rowDataToAttributeValueConverter.convertRowData(rowData))
+ .build();
+
+ assertThat(transformedConverter).extracting("rowDataToAttributeValueConverter").isNotNull();
+
+ assertThat(actualWriteRequest).isEqualTo(expectedWriterequest);
+ }
+
+ private RowData createElement(RowKind kind) {
+ GenericRowData element = new GenericRowData(kind, 2);
+ element.setField(0, StringData.fromString("some_partition_key"));
+ element.setField(1, StringData.fromString("some_payload"));
+ return element;
+ }
+
+ private static class UnusedSinkWriterContext implements SinkWriter.Context {
+
+ @Override
+ public long currentWatermark() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Long timestamp() {
+ throw new UnsupportedOperationException();
+ }
+ }
+}
diff --git a/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java
new file mode 100644
index 0000000..542addc
--- /dev/null
+++ b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java
@@ -0,0 +1,561 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.dynamodb.table;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
+
+import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableMap;
+
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link RowDataToAttributeValueConverter}. */
+public class RowDataToAttributeValueConverterTest {
+
+ @Test
+ void testChar() {
+ String key = "key";
+ String value = "some_char";
+
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.CHAR(9)));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(
+ createElement(StringData.fromString(value)));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(key, AttributeValue.builder().s(value).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testVarChar() {
+ String key = "key";
+ String value = "some_var_char";
+
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.VARCHAR(13)));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(
+ createElement(StringData.fromString(value)));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(key, AttributeValue.builder().s(value).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testString() {
+ String key = "key";
+ String value = "some_string";
+
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.STRING()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(
+ createElement(StringData.fromString(value)));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(key, AttributeValue.builder().s(value).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testBoolean() {
+ String key = "key";
+ boolean value = true;
+
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.BOOLEAN()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(createElement(value));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(key, AttributeValue.builder().bool(value).build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testDecimal() {
+ String key = "key";
+ BigDecimal value = BigDecimal.valueOf(1.001);
+
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.DECIMAL(5, 4)));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(
+ createElement(DecimalData.fromBigDecimal(value, 5, 4)));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(key, AttributeValue.builder().n("1.0010").build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testTinyInt() {
+ String key = "key";
+ byte value = 5;
+
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.TINYINT()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(createElement(value));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(key, AttributeValue.builder().n("5").build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testSmallInt() {
+ String key = "key";
+ short value = 256;
+
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.SMALLINT()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(createElement(value));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(key, AttributeValue.builder().n("256").build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testInt() {
+ String key = "key";
+ int value = 65536;
+
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.INT()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(createElement(value));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(key, AttributeValue.builder().n("65536").build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testBigInt() {
+ String key = "key";
+ long value = 4294967295L;
+
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.BIGINT()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(createElement(value));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(key, AttributeValue.builder().n("4294967295").build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testFloat() {
+ String key = "key";
+ float value = 123456789123456789.000001f;
+
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.FLOAT()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(createElement(value));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(key, AttributeValue.builder().n("1.23456791E17").build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testDouble() {
+ String key = "key";
+ double value = 1.23456789123456789e19;
+
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.DOUBLE()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(createElement(value));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(key, AttributeValue.builder().n("1.234567891234568E19").build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testTimestamp() {
+ String key = "key";
+ LocalDateTime value = LocalDateTime.of(2022, 11, 10, 0, 0);
+
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.TIMESTAMP()));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(
+ createElement(TimestampData.fromLocalDateTime(value)));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(key, AttributeValue.builder().s("2022-11-10T00:00").build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testStringArray() {
+ String key = "key";
+ String[] value = {"Some", "String", "Array"};
+
+ DataType dataType =
+ DataTypes.ROW(DataTypes.FIELD(key, DataTypes.ARRAY(DataTypes.STRING())));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(
+ createArray(value, StringData::fromString));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(
+ key,
+ AttributeValue.builder()
+ .l(
+ Arrays.stream(value)
+ .map(s -> AttributeValue.builder().s(s).build())
+ .collect(Collectors.toList()))
+ .build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testBooleanArray() {
+ String key = "key";
+ Boolean[] value = {true, false, true};
+
+ DataType dataType =
+ DataTypes.ROW(DataTypes.FIELD(key, DataTypes.ARRAY(DataTypes.BOOLEAN())));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(createArray(value, t -> t));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(
+ key,
+ AttributeValue.builder()
+ .l(
+ Arrays.stream(value)
+ .map(
+ bool ->
+ AttributeValue.builder()
+ .bool(bool)
+ .build())
+ .collect(Collectors.toList()))
+ .build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testBigDecimalArray() {
+ String key = "key";
+ BigDecimal[] value = {BigDecimal.ONE, BigDecimal.ZERO};
+
+ DataType dataType =
+ DataTypes.ROW(DataTypes.FIELD(key, DataTypes.ARRAY(DataTypes.DECIMAL(1, 0))));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(
+ createArray(value, d -> DecimalData.fromBigDecimal(d, 1, 0)));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(
+ key,
+ AttributeValue.builder()
+ .l(
+ Arrays.stream(value)
+ .map(
+ s ->
+ AttributeValue.builder()
+ .n(String.valueOf(s))
+ .build())
+ .collect(Collectors.toList()))
+ .build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testByteArray() {
+ String key = "key";
+ Byte[] value = {1, 2, 3, 4};
+
+ DataType dataType =
+ DataTypes.ROW(DataTypes.FIELD(key, DataTypes.ARRAY(DataTypes.TINYINT())));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(createArray(value, t -> t));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(
+ key,
+ AttributeValue.builder()
+ .l(
+ Arrays.stream(value)
+ .map(
+ byt ->
+ AttributeValue.builder()
+ .n(String.valueOf(byt))
+ .build())
+ .collect(Collectors.toList()))
+ .build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testShortArray() {
+ String key = "key";
+ Short[] value = {1, 2, 3, 4};
+
+ DataType dataType =
+ DataTypes.ROW(DataTypes.FIELD(key, DataTypes.ARRAY(DataTypes.SMALLINT())));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(createArray(value, t -> t));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(
+ key,
+ AttributeValue.builder()
+ .l(
+ Arrays.stream(value)
+ .map(
+ sht ->
+ AttributeValue.builder()
+ .n(String.valueOf(sht))
+ .build())
+ .collect(Collectors.toList()))
+ .build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testIntegerArray() {
+ String key = "key";
+ Integer[] value = {1, 2, 3, 4};
+
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.ARRAY(DataTypes.INT())));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(createArray(value, t -> t));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(
+ key,
+ AttributeValue.builder()
+ .l(
+ Arrays.stream(value)
+ .map(
+ integer ->
+ AttributeValue.builder()
+ .n(String.valueOf(integer))
+ .build())
+ .collect(Collectors.toList()))
+ .build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testLongArray() {
+ String key = "key";
+ Long[] value = {1L, 2L, 3L, 4L};
+
+ DataType dataType =
+ DataTypes.ROW(DataTypes.FIELD(key, DataTypes.ARRAY(DataTypes.BIGINT())));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(createArray(value, t -> t));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(
+ key,
+ AttributeValue.builder()
+ .l(
+ Arrays.stream(value)
+ .map(
+ lng ->
+ AttributeValue.builder()
+ .n(String.valueOf(lng))
+ .build())
+ .collect(Collectors.toList()))
+ .build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testFloatArray() {
+ String key = "key";
+ Float[] value = {1f, 2f, 3f, 4f};
+
+ DataType dataType = DataTypes.ROW(DataTypes.FIELD(key, DataTypes.ARRAY(DataTypes.FLOAT())));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(createArray(value, t -> t));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(
+ key,
+ AttributeValue.builder()
+ .l(
+ Arrays.stream(value)
+ .map(
+ flt ->
+ AttributeValue.builder()
+ .n(String.valueOf(flt))
+ .build())
+ .collect(Collectors.toList()))
+ .build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testDoubleArray() {
+ String key = "key";
+ Double[] value = {1E1, 2E2, 3E3, 4E4};
+
+ DataType dataType =
+ DataTypes.ROW(DataTypes.FIELD(key, DataTypes.ARRAY(DataTypes.DOUBLE())));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(createArray(value, t -> t));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(
+ key,
+ AttributeValue.builder()
+ .l(
+ Arrays.stream(value)
+ .map(
+ flt ->
+ AttributeValue.builder()
+ .n(String.valueOf(flt))
+ .build())
+ .collect(Collectors.toList()))
+ .build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testLocalDateTimeArray() {
+ String key = "key";
+ LocalDateTime[] value = {LocalDateTime.now(), LocalDateTime.now(), LocalDateTime.now()};
+
+ DataType dataType =
+ DataTypes.ROW(DataTypes.FIELD(key, DataTypes.ARRAY(DataTypes.TIMESTAMP())));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(
+ createArray(value, TimestampData::fromLocalDateTime));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(
+ key,
+ AttributeValue.builder()
+ .l(
+ Arrays.stream(value)
+ .map(
+ ld ->
+ AttributeValue.builder()
+ .s(String.valueOf(ld))
+ .build())
+ .collect(Collectors.toList()))
+ .build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ @Test
+ void testInstantArray() {
+ String key = "key";
+ Instant[] value = {Instant.now(), Instant.now(), Instant.now()};
+
+ DataType dataType =
+ DataTypes.ROW(DataTypes.FIELD(key, DataTypes.ARRAY(DataTypes.TIMESTAMP_LTZ())));
+ RowDataToAttributeValueConverter rowDataToAttributeValueConverter =
+ new RowDataToAttributeValueConverter(dataType);
+ Map<String, AttributeValue> actualResult =
+ rowDataToAttributeValueConverter.convertRowData(
+ createArray(value, TimestampData::fromInstant));
+ Map<String, AttributeValue> expectedResult =
+ ImmutableMap.of(
+ key,
+ AttributeValue.builder()
+ .l(
+ Arrays.stream(value)
+ .map(
+ ld ->
+ AttributeValue.builder()
+ .s(String.valueOf(ld))
+ .build())
+ .collect(Collectors.toList()))
+ .build());
+
+ assertThat(actualResult).containsAllEntriesOf(expectedResult);
+ }
+
+ private RowData createElement(Object value) {
+ GenericRowData element = new GenericRowData(1);
+ element.setField(0, value);
+ return element;
+ }
+
+ private <T> RowData createArray(T[] value, Function<T, Object> elementConverter) {
+ return createElement(
+ new GenericArrayData(Arrays.stream(value).map(elementConverter).toArray()));
+ }
+}
diff --git a/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/converter/ArrayAttributeConverterTest.java b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/converter/ArrayAttributeConverterTest.java
new file mode 100644
index 0000000..7118b59
--- /dev/null
+++ b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/converter/ArrayAttributeConverterTest.java
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.dynamodb.table.converter;
+
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverterProvider;
+import software.amazon.awssdk.enhanced.dynamodb.EnhancedType;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.util.Arrays;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+
+/** Tests for {@link ArrayAttributeConverter}. */
+public class ArrayAttributeConverterTest {
+
+ @Test
+ void testTransformFrom() {
+ String[] initialStringArray = {"Some", "test", "strings"};
+
+ ArrayAttributeConverter<String> arrayAttributeConverter =
+ new ArrayAttributeConverter<>(
+ AttributeConverterProvider.defaultProvider()
+ .converterFor(EnhancedType.of(String.class)),
+ EnhancedType.of(String[].class));
+
+ AttributeValue actualAttributeValue =
+ arrayAttributeConverter.transformFrom(initialStringArray);
+ AttributeValue expectedAttributeValue =
+ AttributeValue.builder()
+ .l(
+ Arrays.stream(initialStringArray)
+ .map(s -> AttributeValue.builder().s(s).build())
+ .collect(Collectors.toList()))
+ .build();
+
+ assertThat(actualAttributeValue).isEqualTo(expectedAttributeValue);
+ }
+
+ @Test
+ void testTransformTo() {
+ String[] expectedStringArray = {"Some", "test", "strings"};
+ AttributeValue initialAttributeValue =
+ AttributeValue.builder()
+ .l(
+ Arrays.stream(expectedStringArray)
+ .map(s -> AttributeValue.builder().s(s).build())
+ .collect(Collectors.toList()))
+ .build();
+
+ ArrayAttributeConverter<String> arrayAttributeConverter =
+ new ArrayAttributeConverter<>(
+ AttributeConverterProvider.defaultProvider()
+ .converterFor(EnhancedType.of(String.class)),
+ EnhancedType.of(String[].class));
+
+ assertThatExceptionOfType(UnsupportedOperationException.class)
+ .isThrownBy(() -> arrayAttributeConverter.transformTo(initialAttributeValue))
+ .withMessageContaining(
+ "transformTo() is unsupported because the DynamoDB sink does not need it.");
+ }
+
+ @Test
+ void testGetEnhancedType() {
+ ArrayAttributeConverter<String> arrayAttributeConverter =
+ new ArrayAttributeConverter<>(
+ AttributeConverterProvider.defaultProvider()
+ .converterFor(EnhancedType.of(String.class)),
+ EnhancedType.of(String[].class));
+ assertThat(arrayAttributeConverter.type()).isEqualTo(EnhancedType.of(String[].class));
+ }
+}
diff --git a/flink-sql-connector-dynamodb/pom.xml b/flink-sql-connector-dynamodb/pom.xml
index 9867b44..16153ac 100644
--- a/flink-sql-connector-dynamodb/pom.xml
+++ b/flink-sql-connector-dynamodb/pom.xml
@@ -77,6 +77,7 @@ under the License.
<include>org.apache.httpcomponents:*</include>
<include>io.netty:*</include>
<include>commons-logging:commons-logging</include>
+ <include>commons-codec:commons-codec</include>
</includes>
</artifactSet>
<relocations>
diff --git a/flink-sql-connector-dynamodb/src/main/resources/META-INF/NOTICE b/flink-sql-connector-dynamodb/src/main/resources/META-INF/NOTICE
index 89e08bd..96af7b6 100644
--- a/flink-sql-connector-dynamodb/src/main/resources/META-INF/NOTICE
+++ b/flink-sql-connector-dynamodb/src/main/resources/META-INF/NOTICE
@@ -26,19 +26,20 @@ This project bundles the following dependencies under the Apache Software Licens
- software.amazon.awssdk:aws-query-protocol:2.17.247
- software.amazon.awssdk:json-utils:2.17.247
- software.amazon.awssdk:third-party-jackson-core:2.17.247
-- io.netty:netty-codec-http:4.1.70.Final
-- io.netty:netty-codec-http2:4.1.70.Final
-- io.netty:netty-codec:4.1.70.Final
-- io.netty:netty-transport:4.1.70.Final
-- io.netty:netty-resolver:4.1.70.Final
-- io.netty:netty-common:4.1.70.Final
-- io.netty:netty-buffer:4.1.70.Final
-- io.netty:netty-handler:4.1.70.Final
-- io.netty:netty-transport-native-unix-common:4.1.70.Final
-- io.netty:netty-transport-classes-epoll:4.1.70.Final
+- io.netty:netty-codec-http:4.1.77.Final
+- io.netty:netty-codec-http2:4.1.77.Final
+- io.netty:netty-codec:4.1.77.Final
+- io.netty:netty-transport:4.1.77.Final
+- io.netty:netty-resolver:4.1.77.Final
+- io.netty:netty-common:4.1.77.Final
+- io.netty:netty-buffer:4.1.77.Final
+- io.netty:netty-handler:4.1.77.Final
+- io.netty:netty-transport-native-unix-common:4.1.77.Final
+- io.netty:netty-transport-classes-epoll:4.1.77.Final
- org.apache.httpcomponents:httpclient:4.5.13
-- org.apache.httpcomponents:httpcore:4.4.14
-- commons-logging:commons-logging:1.1.3
+- org.apache.httpcomponents:httpcore:4.4.13
+- commons-logging:commons-logging:1.2
+- commons-codec:commons-codec:1.11
This project bundles the following dependencies under the Creative Commons Zero license (https://creativecommons.org/publicdomain/zero/1.0/).
diff --git a/flink-sql-connector-dynamodb/src/main/resources/org/apache/flink/connector/dynamodb/shaded/software/amazon/awssdk/services/dynamodb/execution.interceptors b/flink-sql-connector-dynamodb/src/main/resources/org/apache/flink/connector/dynamodb/shaded/software/amazon/awssdk/services/dynamodb/execution.interceptors
new file mode 100644
index 0000000..4c22909
--- /dev/null
+++ b/flink-sql-connector-dynamodb/src/main/resources/org/apache/flink/connector/dynamodb/shaded/software/amazon/awssdk/services/dynamodb/execution.interceptors
@@ -0,0 +1 @@
+org.apache.flink.connector.dynamodb.shaded.software.amazon.awssdk.enhanced.dynamodb.internal.ApplyUserAgentInterceptor