You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "vahmed-hamdy (via GitHub)" <gi...@apache.org> on 2024/04/11 11:26:47 UTC
[PR] [FLINK-35022] Add TypeInformed DDB Element Converter [flink-connector-aws]
vahmed-hamdy opened a new pull request, #136:
URL: https://github.com/apache/flink-connector-aws/pull/136
<!--
*Thank you for contributing to Apache Flink AWS Connectors - we are happy that you want to help us improve our Flink connectors. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- The name of the pull request should correspond to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Commits should be in the form of "[FLINK-XXXX][component] Title of the pull request", where [FLINK-XXXX] should be replaced by the actual issue number.
Generally, [component] should be the connector you are working on.
For example: "[FLINK-XXXX][Connectors/Kinesis] XXXX" if you are working on the Kinesis connector or "[FLINK-XXXX][Connectors/AWS] XXXX" if you are working on components shared among all the connectors.
- Each pull request should only have one JIRA issue.
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
-->
## Purpose of the change
Add `DynamoDbTypeInformedElementConverter` to convert Elements to dynamoDb Sink using its provided type Info.
## Verifying this change
This change added tests and can be verified as follows:
- Added unit tests
## Significant changes
*(Please check any boxes [x] if the answer is "yes". You can first publish the PR and check them afterwards, for convenience.)*
- [ ] Dependencies have been added or upgraded
- [ ] Public API has been changed (Public API is any class annotated with `@Public(Evolving)`)
- [ ] Serializers have been changed
- [ ] New feature has been introduced
- If yes, how is this documented? (not applicable / docs / JavaDocs / not documented)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35022][Connector/DynamoDB] Add TypeInformed DDB Element Converter as default element converter [flink-connector-aws]
Posted by "dannycranmer (via GitHub)" <gi...@apache.org>.
dannycranmer commented on PR #136:
URL: https://github.com/apache/flink-connector-aws/pull/136#issuecomment-2158683163
@hlteoh37 / @z3d1k can you please take a look at this one too?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35022] Add TypeInformed DDB Element Converter [flink-connector-aws]
Posted by "vahmed-hamdy (via GitHub)" <gi...@apache.org>.
vahmed-hamdy commented on code in PR #136:
URL: https://github.com/apache/flink-connector-aws/pull/136#discussion_r1565513965
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbTypeInformedElementConverter.java:
##########
@@ -0,0 +1,245 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.NumericTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * A {@link ElementConverter} that converts an element to a {@link DynamoDbWriteRequest} using
+ * TypeInformation provided.
+ */
+@PublicEvolving
+public class DynamoDbTypeInformedElementConverter<inputT>
+ implements ElementConverter<inputT, DynamoDbWriteRequest> {
+ private final CompositeType<inputT> typeInfo;
+
+ /**
+ * Creates a {@link DynamoDbTypeInformedElementConverter} that converts an element to a {@link
+ * DynamoDbWriteRequest} using the provided {@link CompositeType}. Usage: {@code new
+ * DynamoDbTypeInformedElementConverter<>(TypeInformation.of(MyPojoClass.class))}
+ *
+ * @param typeInfo The {@link CompositeType} that provides the type information for the element.
+ */
+ public DynamoDbTypeInformedElementConverter(CompositeType<inputT> typeInfo) {
+ this.typeInfo = typeInfo;
+ }
+
+ @Override
+ public DynamoDbWriteRequest apply(inputT input, SinkWriter.Context context) {
+ try {
+ return DynamoDbWriteRequest.builder()
+ .setType(DynamoDbWriteRequestType.PUT)
+ .setItem(convertElementUsingTypeInfo(input, typeInfo))
+ .build();
+ } catch (IllegalArgumentException e) {
+ throw new FlinkRuntimeException("Couldn't convert Element to AttributeVal", e);
+ }
+ }
+
+ private <attT> Map<String, AttributeValue> convertElementUsingTypeInfo(
+ attT t, CompositeType<attT> typeInfo) {
+ Map<String, AttributeValue> map = new HashMap<>();
+ for (String fieldKey : typeInfo.getFieldNames()) {
+ TypeInformation<attT> fieldType = typeInfo.getTypeAt(fieldKey);
+ try {
+ Field field = t.getClass().getDeclaredField(fieldKey);
+ field.setAccessible(true);
+ Object fieldVal = field.get(t);
+ checkTypeCompatibility(fieldVal, fieldType);
+ attT fieldValCaster = (attT) fieldVal;
+ map.put(fieldKey, convertValue(fieldValCaster, fieldType));
+ } catch (NoSuchFieldException | IllegalAccessException e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Failed to extract field %s declared in TypeInfo "
+ + "from Object %s",
+ fieldKey, t),
+ e);
+ }
+ }
+
+ return map;
+ }
+
+ private <attrT> AttributeValue convertValue(
+ attrT attribute, TypeInformation<attrT> objectTypeInformation) {
+ if (attribute == null) {
+ return AttributeValue.builder().nul(true).build();
Review Comment:
@dannycranmer , yeah just checked `testConvertOrderToDynamoDbWriteRequestWithIgnoresNull`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35022][Connector/DynamoDB] Add TypeInformed DDB Element Converter as default element converter [flink-connector-aws]
Posted by "vahmed-hamdy (via GitHub)" <gi...@apache.org>.
vahmed-hamdy commented on code in PR #136:
URL: https://github.com/apache/flink-connector-aws/pull/136#discussion_r1638374818
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbTypeInformedElementConverter.java:
##########
@@ -0,0 +1,380 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.NumericTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.PojoTypeInfo;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.dynamodb.table.converter.ArrayAttributeConverterProvider;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverter;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverterProvider;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeValueType;
+import software.amazon.awssdk.enhanced.dynamodb.EnhancedType;
+import software.amazon.awssdk.enhanced.dynamodb.TableSchema;
+import software.amazon.awssdk.enhanced.dynamodb.internal.mapper.BeanAttributeGetter;
+import software.amazon.awssdk.enhanced.dynamodb.mapper.StaticTableSchema;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.beans.BeanInfo;
+import java.beans.IntrospectionException;
+import java.beans.Introspector;
+import java.beans.PropertyDescriptor;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Function;
+
+/**
+ * A {@link ElementConverter} that converts an element to a {@link DynamoDbWriteRequest} using
+ * TypeInformation provided.
+ */
+@PublicEvolving
+public class DynamoDbTypeInformedElementConverter<T>
+ implements ElementConverter<T, DynamoDbWriteRequest> {
+
+ private final CompositeType<T> typeInfo;
+ private final boolean ignoreNulls;
+ private final TableSchema<T> tableSchema;
+
+ /**
+ * Creates a {@link DynamoDbTypeInformedElementConverter} that converts an element to a {@link
+ * DynamoDbWriteRequest} using the provided {@link CompositeType}. Usage: {@code new
+ * DynamoDbTypeInformedElementConverter<>(TypeInformation.of(MyPojoClass.class))}
+ *
+ * @param typeInfo The {@link CompositeType} that provides the type information for the element.
+ */
+ public DynamoDbTypeInformedElementConverter(CompositeType<T> typeInfo) {
+ this(typeInfo, true);
+ }
+
+ public DynamoDbTypeInformedElementConverter(CompositeType<T> typeInfo, boolean ignoreNulls) {
+
+ try {
+ this.typeInfo = typeInfo;
+ this.ignoreNulls = ignoreNulls;
+ this.tableSchema = createTableSchema(typeInfo);
+ } catch (IntrospectionException | IllegalStateException | IllegalArgumentException e) {
+ throw new FlinkRuntimeException("Failed to extract DynamoDb table schema", e);
+ }
+ }
+
+ @Override
+ public DynamoDbWriteRequest apply(T input, SinkWriter.Context context) {
+ Preconditions.checkNotNull(tableSchema, "TableSchema is not initialized");
+ try {
+ return DynamoDbWriteRequest.builder()
+ .setType(DynamoDbWriteRequestType.PUT)
+ .setItem(tableSchema.itemToMap(input, ignoreNulls))
+ .build();
+ } catch (ClassCastException | IllegalArgumentException e) {
+ throw new FlinkRuntimeException(
+ String.format(
+ "Failed to convert %s to DynamoDbWriteRequest using %s",
+ input, typeInfo),
+ e);
+ }
+ }
+
+ private <AttributeT> TableSchema<AttributeT> createTableSchema(
+ CompositeType<AttributeT> typeInfo) throws IntrospectionException {
+ if (typeInfo instanceof RowTypeInfo) {
+ return (TableSchema<AttributeT>) createTableSchemaFromRowType((RowTypeInfo) typeInfo);
+ } else if (typeInfo instanceof PojoTypeInfo<?>) {
+ return createTableSchemaFromPojo((PojoTypeInfo<AttributeT>) typeInfo);
+ } else if (typeInfo instanceof TupleTypeInfo<?>) {
+ return createTableSchemaFromTuple((TupleTypeInfo<?>) typeInfo);
+ } else {
+ throw new IllegalArgumentException(String.format("Unsupported TypeInfo %s", typeInfo));
+ }
+ }
+
+ private TableSchema<Row> createTableSchemaFromRowType(RowTypeInfo typeInfo) {
+ StaticTableSchema.Builder<Row> tableSchemaBuilder =
+ StaticTableSchema.builder(typeInfo.getTypeClass());
+
+ String[] fieldNames = typeInfo.getFieldNames();
+ for (int i = 0; i < typeInfo.getArity(); i++) {
+ TypeInformation<?> fieldType = typeInfo.getTypeAt(i);
+ int finalI = i;
+ addAttribute(
+ tableSchemaBuilder,
+ fieldNames[finalI],
+ (AttributeT) -> AttributeT.getField(finalI),
+ (TypeInformation<? super Object>) fieldType);
+ }
+
+ return tableSchemaBuilder.build();
+ }
+
+ private <AttributeT> TableSchema<AttributeT> createTableSchemaFromTuple(
+ TupleTypeInfo<?> typeInfo) {
+ TypeInformation<?>[] fieldTypes = typeInfo.getFieldTypes();
+ String[] fieldNames = typeInfo.getFieldNames();
+
+ StaticTableSchema.Builder<?> tableSchemaBuilder =
+ StaticTableSchema.builder(typeInfo.getTypeClass());
+ for (int i = 0; i < fieldNames.length; i++) {
+ int finalI = i;
+ addAttribute(
+ tableSchemaBuilder,
+ fieldNames[finalI],
+ (tuple) -> ((Tuple) tuple).getField(finalI),
+ (TypeInformation<?>) fieldTypes[i]);
+ }
+
+ return (TableSchema<AttributeT>) tableSchemaBuilder.build();
+ }
+
+ private <AttributeT> TableSchema<AttributeT> createTableSchemaFromPojo(
+ PojoTypeInfo<AttributeT> typeInfo) throws IntrospectionException {
+ StaticTableSchema.Builder<AttributeT> tableSchemaBuilder =
+ StaticTableSchema.builder(typeInfo.getTypeClass());
+ BeanInfo beanInfo = Introspector.getBeanInfo(typeInfo.getTypeClass());
+ PropertyDescriptor[] propertyDescriptors = beanInfo.getPropertyDescriptors();
+ for (PropertyDescriptor propertyDescriptor : propertyDescriptors) {
+ Set<String> fieldNames = new HashSet<>(Arrays.asList(typeInfo.getFieldNames()));
+ if (!fieldNames.contains(propertyDescriptor.getName())) {
+ // Skip properties that are not part of the PojoTypeInfo
+ continue;
+ }
+
+ TypeInformation<?> fieldInfo = typeInfo.getTypeAt(propertyDescriptor.getName());
+ addAttribute(
+ tableSchemaBuilder,
+ propertyDescriptor.getName(),
+ BeanAttributeGetter.create(
+ typeInfo.getTypeClass(), propertyDescriptor.getReadMethod()),
+ fieldInfo);
+ }
+
+ return tableSchemaBuilder.build();
+ }
+
+ private <T, AttributeT> void addAttribute(
+ StaticTableSchema.Builder<T> builder,
+ String fieldName,
+ Function<T, AttributeT> getter,
+ TypeInformation<AttributeT> typeInfo) {
+ builder.addAttribute(
+ typeInfo.getTypeClass(),
+ a ->
+ a.name(fieldName)
+ .getter(getter)
+ .setter((e, o) -> {})
+ .attributeConverter(getAttributeConverter(typeInfo)));
+ }
+
+ private <AttributeT> AttributeConverter<AttributeT> getAttributeConverter(
+ TypeInformation<AttributeT> typeInfo) {
+ if (typeInfo instanceof BasicTypeInfo) {
+ return AttributeConverterProvider.defaultProvider()
+ .converterFor(EnhancedType.of(typeInfo.getTypeClass()));
+ } else if (typeInfo.equals(PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO)) {
+ return getAttributeConverter(
+ AttributeValueType.B,
+ bytes ->
+ bytes instanceof SdkBytes
+ ? AttributeValue.fromB((SdkBytes) bytes)
+ : AttributeValue.fromB(SdkBytes.fromByteArray((byte[]) bytes)));
+ } else if (typeInfo instanceof BasicArrayTypeInfo) {
+ BasicArrayTypeInfo<AttributeT, ?> basicArrayTypeInfo =
+ (BasicArrayTypeInfo<AttributeT, ?>) typeInfo;
+ if (basicArrayTypeInfo.getComponentInfo().equals(BasicTypeInfo.STRING_TYPE_INFO)) {
+ return getAttributeConverter(
+ AttributeValueType.SS,
+ array -> AttributeValue.fromSs(Arrays.asList((String[]) array)));
+ } else if (basicArrayTypeInfo.getComponentInfo() instanceof NumericTypeInfo) {
+ return getAttributeConverter(
+ AttributeValueType.NS,
+ array ->
+ AttributeValue.fromNs(
+ convertObjectArrayToStringList((Object[]) array)));
+ }
+
+ return new ArrayAttributeConverterProvider()
+ .converterFor(EnhancedType.of(typeInfo.getTypeClass()));
+ } else if (typeInfo instanceof ObjectArrayTypeInfo) {
+ return getObjectArrayTypeConverter((ObjectArrayTypeInfo<AttributeT, ?>) typeInfo);
+ } else if (typeInfo instanceof PrimitiveArrayTypeInfo) {
+ PrimitiveArrayTypeInfo<AttributeT> primitiveArrayTypeInfo =
+ (PrimitiveArrayTypeInfo<AttributeT>) typeInfo;
+ if (primitiveArrayTypeInfo.getComponentType() instanceof NumericTypeInfo) {
+ return getAttributeConverter(
+ AttributeValueType.NS,
+ array -> AttributeValue.fromNs(convertPrimitiveArrayToStringList(array)));
+ } else {
+ throw new IllegalArgumentException(
+ String.format(
+ "Unsupported primitive array typeInfo %s",
+ primitiveArrayTypeInfo.getComponentType()));
+ }
Review Comment:
@hlteoh37 yes, the whole approach actually is very similar, however The way `AttributeConverterProvider` works is by mapping `EnhancedType` provided to an `AttributeConverter`, We here rely on Flink's `TypeInfo` rather than the `EnhancedType` that's why we created an alternative structure. Wdyt?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35022][Connector/DynamoDB] Add TypeInformed DDB Element Converter as default element converter [flink-connector-aws]
Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 commented on code in PR #136:
URL: https://github.com/apache/flink-connector-aws/pull/136#discussion_r1650817669
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbTypeInformedElementConverter.java:
##########
@@ -0,0 +1,380 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.NumericTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.PojoTypeInfo;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.dynamodb.table.converter.ArrayAttributeConverterProvider;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverter;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverterProvider;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeValueType;
+import software.amazon.awssdk.enhanced.dynamodb.EnhancedType;
+import software.amazon.awssdk.enhanced.dynamodb.TableSchema;
+import software.amazon.awssdk.enhanced.dynamodb.internal.mapper.BeanAttributeGetter;
+import software.amazon.awssdk.enhanced.dynamodb.mapper.StaticTableSchema;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.beans.BeanInfo;
+import java.beans.IntrospectionException;
+import java.beans.Introspector;
+import java.beans.PropertyDescriptor;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Function;
+
+/**
+ * A {@link ElementConverter} that converts an element to a {@link DynamoDbWriteRequest} using
+ * TypeInformation provided.
+ */
+@PublicEvolving
+public class DynamoDbTypeInformedElementConverter<T>
+ implements ElementConverter<T, DynamoDbWriteRequest> {
+
+ private final CompositeType<T> typeInfo;
+ private final boolean ignoreNulls;
+ private final TableSchema<T> tableSchema;
+
+ /**
+ * Creates a {@link DynamoDbTypeInformedElementConverter} that converts an element to a {@link
+ * DynamoDbWriteRequest} using the provided {@link CompositeType}. Usage: {@code new
+ * DynamoDbTypeInformedElementConverter<>(TypeInformation.of(MyPojoClass.class))}
+ *
+ * @param typeInfo The {@link CompositeType} that provides the type information for the element.
+ */
+ public DynamoDbTypeInformedElementConverter(CompositeType<T> typeInfo) {
+ this(typeInfo, true);
+ }
+
+ public DynamoDbTypeInformedElementConverter(CompositeType<T> typeInfo, boolean ignoreNulls) {
+
+ try {
+ this.typeInfo = typeInfo;
+ this.ignoreNulls = ignoreNulls;
+ this.tableSchema = createTableSchema(typeInfo);
+ } catch (IntrospectionException | IllegalStateException | IllegalArgumentException e) {
+ throw new FlinkRuntimeException("Failed to extract DynamoDb table schema", e);
+ }
+ }
+
+ @Override
+ public DynamoDbWriteRequest apply(T input, SinkWriter.Context context) {
+ Preconditions.checkNotNull(tableSchema, "TableSchema is not initialized");
+ try {
+ return DynamoDbWriteRequest.builder()
+ .setType(DynamoDbWriteRequestType.PUT)
+ .setItem(tableSchema.itemToMap(input, ignoreNulls))
+ .build();
+ } catch (ClassCastException | IllegalArgumentException e) {
+ throw new FlinkRuntimeException(
+ String.format(
+ "Failed to convert %s to DynamoDbWriteRequest using %s",
+ input, typeInfo),
+ e);
+ }
+ }
+
+ private <AttributeT> TableSchema<AttributeT> createTableSchema(
+ CompositeType<AttributeT> typeInfo) throws IntrospectionException {
+ if (typeInfo instanceof RowTypeInfo) {
+ return (TableSchema<AttributeT>) createTableSchemaFromRowType((RowTypeInfo) typeInfo);
+ } else if (typeInfo instanceof PojoTypeInfo<?>) {
+ return createTableSchemaFromPojo((PojoTypeInfo<AttributeT>) typeInfo);
+ } else if (typeInfo instanceof TupleTypeInfo<?>) {
+ return createTableSchemaFromTuple((TupleTypeInfo<?>) typeInfo);
+ } else {
+ throw new IllegalArgumentException(String.format("Unsupported TypeInfo %s", typeInfo));
+ }
+ }
+
+ private TableSchema<Row> createTableSchemaFromRowType(RowTypeInfo typeInfo) {
+ StaticTableSchema.Builder<Row> tableSchemaBuilder =
+ StaticTableSchema.builder(typeInfo.getTypeClass());
+
+ String[] fieldNames = typeInfo.getFieldNames();
+ for (int i = 0; i < typeInfo.getArity(); i++) {
+ TypeInformation<?> fieldType = typeInfo.getTypeAt(i);
+ int finalI = i;
+ addAttribute(
+ tableSchemaBuilder,
+ fieldNames[finalI],
+ (AttributeT) -> AttributeT.getField(finalI),
+ (TypeInformation<? super Object>) fieldType);
+ }
+
+ return tableSchemaBuilder.build();
+ }
+
+ private <AttributeT> TableSchema<AttributeT> createTableSchemaFromTuple(
+ TupleTypeInfo<?> typeInfo) {
+ TypeInformation<?>[] fieldTypes = typeInfo.getFieldTypes();
+ String[] fieldNames = typeInfo.getFieldNames();
+
+ StaticTableSchema.Builder<?> tableSchemaBuilder =
+ StaticTableSchema.builder(typeInfo.getTypeClass());
+ for (int i = 0; i < fieldNames.length; i++) {
+ int finalI = i;
+ addAttribute(
+ tableSchemaBuilder,
+ fieldNames[finalI],
+ (tuple) -> ((Tuple) tuple).getField(finalI),
+ (TypeInformation<?>) fieldTypes[i]);
+ }
+
+ return (TableSchema<AttributeT>) tableSchemaBuilder.build();
+ }
+
+ private <AttributeT> TableSchema<AttributeT> createTableSchemaFromPojo(
+ PojoTypeInfo<AttributeT> typeInfo) throws IntrospectionException {
+ StaticTableSchema.Builder<AttributeT> tableSchemaBuilder =
+ StaticTableSchema.builder(typeInfo.getTypeClass());
+ BeanInfo beanInfo = Introspector.getBeanInfo(typeInfo.getTypeClass());
+ PropertyDescriptor[] propertyDescriptors = beanInfo.getPropertyDescriptors();
+ for (PropertyDescriptor propertyDescriptor : propertyDescriptors) {
+ Set<String> fieldNames = new HashSet<>(Arrays.asList(typeInfo.getFieldNames()));
+ if (!fieldNames.contains(propertyDescriptor.getName())) {
+ // Skip properties that are not part of the PojoTypeInfo
+ continue;
+ }
+
+ TypeInformation<?> fieldInfo = typeInfo.getTypeAt(propertyDescriptor.getName());
+ addAttribute(
+ tableSchemaBuilder,
+ propertyDescriptor.getName(),
+ BeanAttributeGetter.create(
+ typeInfo.getTypeClass(), propertyDescriptor.getReadMethod()),
+ fieldInfo);
+ }
+
+ return tableSchemaBuilder.build();
+ }
+
+ private <T, AttributeT> void addAttribute(
+ StaticTableSchema.Builder<T> builder,
+ String fieldName,
+ Function<T, AttributeT> getter,
+ TypeInformation<AttributeT> typeInfo) {
+ builder.addAttribute(
+ typeInfo.getTypeClass(),
+ a ->
+ a.name(fieldName)
+ .getter(getter)
+ .setter((e, o) -> {})
+ .attributeConverter(getAttributeConverter(typeInfo)));
+ }
+
+ private <AttributeT> AttributeConverter<AttributeT> getAttributeConverter(
+ TypeInformation<AttributeT> typeInfo) {
+ if (typeInfo instanceof BasicTypeInfo) {
+ return AttributeConverterProvider.defaultProvider()
+ .converterFor(EnhancedType.of(typeInfo.getTypeClass()));
+ } else if (typeInfo.equals(PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO)) {
+ return getAttributeConverter(
+ AttributeValueType.B,
+ bytes ->
+ bytes instanceof SdkBytes
+ ? AttributeValue.fromB((SdkBytes) bytes)
+ : AttributeValue.fromB(SdkBytes.fromByteArray((byte[]) bytes)));
+ } else if (typeInfo instanceof BasicArrayTypeInfo) {
+ BasicArrayTypeInfo<AttributeT, ?> basicArrayTypeInfo =
+ (BasicArrayTypeInfo<AttributeT, ?>) typeInfo;
+ if (basicArrayTypeInfo.getComponentInfo().equals(BasicTypeInfo.STRING_TYPE_INFO)) {
+ return getAttributeConverter(
+ AttributeValueType.SS,
+ array -> AttributeValue.fromSs(Arrays.asList((String[]) array)));
+ } else if (basicArrayTypeInfo.getComponentInfo() instanceof NumericTypeInfo) {
+ return getAttributeConverter(
+ AttributeValueType.NS,
+ array ->
+ AttributeValue.fromNs(
+ convertObjectArrayToStringList((Object[]) array)));
+ }
+
+ return new ArrayAttributeConverterProvider()
+ .converterFor(EnhancedType.of(typeInfo.getTypeClass()));
+ } else if (typeInfo instanceof ObjectArrayTypeInfo) {
+ return getObjectArrayTypeConverter((ObjectArrayTypeInfo<AttributeT, ?>) typeInfo);
+ } else if (typeInfo instanceof PrimitiveArrayTypeInfo) {
+ PrimitiveArrayTypeInfo<AttributeT> primitiveArrayTypeInfo =
+ (PrimitiveArrayTypeInfo<AttributeT>) typeInfo;
+ if (primitiveArrayTypeInfo.getComponentType() instanceof NumericTypeInfo) {
+ return getAttributeConverter(
+ AttributeValueType.NS,
+ array -> AttributeValue.fromNs(convertPrimitiveArrayToStringList(array)));
+ } else {
+ throw new IllegalArgumentException(
+ String.format(
+ "Unsupported primitive array typeInfo %s",
+ primitiveArrayTypeInfo.getComponentType()));
+ }
Review Comment:
Hm, ultimately `EnhancedType` is just a wrapper for the `AttributeConverterProvider` to match the right `AttributeConverter`. We cannot run away from this - in fact you have created a custom implementation of this logic in `getAttributeConverter`, which takes type information and constructs a custom attribute converter.
Wouldn't using the `AttributeConverterProvider` more directly simplify the code?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35022][Connector/DynamoDB] Add TypeInformed DDB Element Converter as default element converter [flink-connector-aws]
Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 commented on code in PR #136:
URL: https://github.com/apache/flink-connector-aws/pull/136#discussion_r1634758095
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbTypeInformedElementConverter.java:
##########
@@ -0,0 +1,380 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.NumericTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.PojoTypeInfo;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.dynamodb.table.converter.ArrayAttributeConverterProvider;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverter;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverterProvider;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeValueType;
+import software.amazon.awssdk.enhanced.dynamodb.EnhancedType;
+import software.amazon.awssdk.enhanced.dynamodb.TableSchema;
+import software.amazon.awssdk.enhanced.dynamodb.internal.mapper.BeanAttributeGetter;
+import software.amazon.awssdk.enhanced.dynamodb.mapper.StaticTableSchema;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.beans.BeanInfo;
+import java.beans.IntrospectionException;
+import java.beans.Introspector;
+import java.beans.PropertyDescriptor;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Function;
+
+/**
+ * A {@link ElementConverter} that converts an element to a {@link DynamoDbWriteRequest} using
+ * TypeInformation provided.
+ */
+@PublicEvolving
+public class DynamoDbTypeInformedElementConverter<T>
+ implements ElementConverter<T, DynamoDbWriteRequest> {
+
+ private final CompositeType<T> typeInfo;
+ private final boolean ignoreNulls;
+ private final TableSchema<T> tableSchema;
+
+ /**
+ * Creates a {@link DynamoDbTypeInformedElementConverter} that converts an element to a {@link
+ * DynamoDbWriteRequest} using the provided {@link CompositeType}. Usage: {@code new
+ * DynamoDbTypeInformedElementConverter<>(TypeInformation.of(MyPojoClass.class))}
+ *
+ * @param typeInfo The {@link CompositeType} that provides the type information for the element.
+ */
+ public DynamoDbTypeInformedElementConverter(CompositeType<T> typeInfo) {
+ this(typeInfo, true);
+ }
+
+ public DynamoDbTypeInformedElementConverter(CompositeType<T> typeInfo, boolean ignoreNulls) {
+
+ try {
+ this.typeInfo = typeInfo;
+ this.ignoreNulls = ignoreNulls;
+ this.tableSchema = createTableSchema(typeInfo);
+ } catch (IntrospectionException | IllegalStateException | IllegalArgumentException e) {
+ throw new FlinkRuntimeException("Failed to extract DynamoDb table schema", e);
+ }
+ }
+
+ @Override
+ public DynamoDbWriteRequest apply(T input, SinkWriter.Context context) {
+ Preconditions.checkNotNull(tableSchema, "TableSchema is not initialized");
+ try {
+ return DynamoDbWriteRequest.builder()
+ .setType(DynamoDbWriteRequestType.PUT)
+ .setItem(tableSchema.itemToMap(input, ignoreNulls))
+ .build();
+ } catch (ClassCastException | IllegalArgumentException e) {
+ throw new FlinkRuntimeException(
+ String.format(
+ "Failed to convert %s to DynamoDbWriteRequest using %s",
+ input, typeInfo),
+ e);
+ }
+ }
+
+ private <AttributeT> TableSchema<AttributeT> createTableSchema(
+ CompositeType<AttributeT> typeInfo) throws IntrospectionException {
+ if (typeInfo instanceof RowTypeInfo) {
+ return (TableSchema<AttributeT>) createTableSchemaFromRowType((RowTypeInfo) typeInfo);
+ } else if (typeInfo instanceof PojoTypeInfo<?>) {
+ return createTableSchemaFromPojo((PojoTypeInfo<AttributeT>) typeInfo);
+ } else if (typeInfo instanceof TupleTypeInfo<?>) {
+ return createTableSchemaFromTuple((TupleTypeInfo<?>) typeInfo);
+ } else {
+ throw new IllegalArgumentException(String.format("Unsupported TypeInfo %s", typeInfo));
+ }
+ }
+
+ private TableSchema<Row> createTableSchemaFromRowType(RowTypeInfo typeInfo) {
+ StaticTableSchema.Builder<Row> tableSchemaBuilder =
+ StaticTableSchema.builder(typeInfo.getTypeClass());
+
+ String[] fieldNames = typeInfo.getFieldNames();
+ for (int i = 0; i < typeInfo.getArity(); i++) {
+ TypeInformation<?> fieldType = typeInfo.getTypeAt(i);
+ int finalI = i;
+ addAttribute(
+ tableSchemaBuilder,
+ fieldNames[finalI],
+ (AttributeT) -> AttributeT.getField(finalI),
+ (TypeInformation<? super Object>) fieldType);
+ }
+
+ return tableSchemaBuilder.build();
+ }
+
+ private <AttributeT> TableSchema<AttributeT> createTableSchemaFromTuple(
+ TupleTypeInfo<?> typeInfo) {
+ TypeInformation<?>[] fieldTypes = typeInfo.getFieldTypes();
+ String[] fieldNames = typeInfo.getFieldNames();
+
+ StaticTableSchema.Builder<?> tableSchemaBuilder =
+ StaticTableSchema.builder(typeInfo.getTypeClass());
+ for (int i = 0; i < fieldNames.length; i++) {
+ int finalI = i;
+ addAttribute(
+ tableSchemaBuilder,
+ fieldNames[finalI],
+ (tuple) -> ((Tuple) tuple).getField(finalI),
+ (TypeInformation<?>) fieldTypes[i]);
+ }
+
+ return (TableSchema<AttributeT>) tableSchemaBuilder.build();
+ }
+
+ private <AttributeT> TableSchema<AttributeT> createTableSchemaFromPojo(
+ PojoTypeInfo<AttributeT> typeInfo) throws IntrospectionException {
+ StaticTableSchema.Builder<AttributeT> tableSchemaBuilder =
+ StaticTableSchema.builder(typeInfo.getTypeClass());
+ BeanInfo beanInfo = Introspector.getBeanInfo(typeInfo.getTypeClass());
+ PropertyDescriptor[] propertyDescriptors = beanInfo.getPropertyDescriptors();
+ for (PropertyDescriptor propertyDescriptor : propertyDescriptors) {
+ Set<String> fieldNames = new HashSet<>(Arrays.asList(typeInfo.getFieldNames()));
+ if (!fieldNames.contains(propertyDescriptor.getName())) {
+ // Skip properties that are not part of the PojoTypeInfo
+ continue;
+ }
+
+ TypeInformation<?> fieldInfo = typeInfo.getTypeAt(propertyDescriptor.getName());
+ addAttribute(
+ tableSchemaBuilder,
+ propertyDescriptor.getName(),
+ BeanAttributeGetter.create(
+ typeInfo.getTypeClass(), propertyDescriptor.getReadMethod()),
+ fieldInfo);
+ }
+
+ return tableSchemaBuilder.build();
+ }
+
+ private <T, AttributeT> void addAttribute(
+ StaticTableSchema.Builder<T> builder,
+ String fieldName,
+ Function<T, AttributeT> getter,
+ TypeInformation<AttributeT> typeInfo) {
+ builder.addAttribute(
+ typeInfo.getTypeClass(),
+ a ->
+ a.name(fieldName)
+ .getter(getter)
+ .setter((e, o) -> {})
+ .attributeConverter(getAttributeConverter(typeInfo)));
+ }
+
+ private <AttributeT> AttributeConverter<AttributeT> getAttributeConverter(
+ TypeInformation<AttributeT> typeInfo) {
+ if (typeInfo instanceof BasicTypeInfo) {
+ return AttributeConverterProvider.defaultProvider()
+ .converterFor(EnhancedType.of(typeInfo.getTypeClass()));
+ } else if (typeInfo.equals(PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO)) {
+ return getAttributeConverter(
+ AttributeValueType.B,
+ bytes ->
+ bytes instanceof SdkBytes
+ ? AttributeValue.fromB((SdkBytes) bytes)
+ : AttributeValue.fromB(SdkBytes.fromByteArray((byte[]) bytes)));
+ } else if (typeInfo instanceof BasicArrayTypeInfo) {
+ BasicArrayTypeInfo<AttributeT, ?> basicArrayTypeInfo =
+ (BasicArrayTypeInfo<AttributeT, ?>) typeInfo;
+ if (basicArrayTypeInfo.getComponentInfo().equals(BasicTypeInfo.STRING_TYPE_INFO)) {
+ return getAttributeConverter(
+ AttributeValueType.SS,
+ array -> AttributeValue.fromSs(Arrays.asList((String[]) array)));
+ } else if (basicArrayTypeInfo.getComponentInfo() instanceof NumericTypeInfo) {
+ return getAttributeConverter(
+ AttributeValueType.NS,
+ array ->
+ AttributeValue.fromNs(
+ convertObjectArrayToStringList((Object[]) array)));
+ }
+
+ return new ArrayAttributeConverterProvider()
+ .converterFor(EnhancedType.of(typeInfo.getTypeClass()));
+ } else if (typeInfo instanceof ObjectArrayTypeInfo) {
+ return getObjectArrayTypeConverter((ObjectArrayTypeInfo<AttributeT, ?>) typeInfo);
+ } else if (typeInfo instanceof PrimitiveArrayTypeInfo) {
+ PrimitiveArrayTypeInfo<AttributeT> primitiveArrayTypeInfo =
+ (PrimitiveArrayTypeInfo<AttributeT>) typeInfo;
+ if (primitiveArrayTypeInfo.getComponentType() instanceof NumericTypeInfo) {
+ return getAttributeConverter(
+ AttributeValueType.NS,
+ array -> AttributeValue.fromNs(convertPrimitiveArrayToStringList(array)));
+ } else {
+ throw new IllegalArgumentException(
+ String.format(
+ "Unsupported primitive array typeInfo %s",
+ primitiveArrayTypeInfo.getComponentType()));
+ }
Review Comment:
Have we considered using the `.attributeConverterProviders()` method in the table schema builder and reusing the constructs we use to DDB Table API?
```
TableSchema.builder(...)
.attributeConverterProviders(
new ArrayAttributeConverterProvider(),
... (add any more here) ...,
AttributeConverterProvider.defaultProvider());
```
We have an example here:
https://github.com/apache/flink-connector-aws/blob/main/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverter.java#L56-L72
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35022] Add TypeInformed DDB Element Converter [flink-connector-aws]
Posted by "dannycranmer (via GitHub)" <gi...@apache.org>.
dannycranmer commented on PR #136:
URL: https://github.com/apache/flink-connector-aws/pull/136#issuecomment-2056383239
I have left a comment regarding the need for this on the Jira: https://issues.apache.org/jira/browse/FLINK-35022
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35022][Connector/DynamoDB] Add TypeInformed DDB Element Converter [flink-connector-aws]
Posted by "vahmed-hamdy (via GitHub)" <gi...@apache.org>.
vahmed-hamdy commented on code in PR #136:
URL: https://github.com/apache/flink-connector-aws/pull/136#discussion_r1581114799
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbTypeInformedElementConverter.java:
##########
@@ -0,0 +1,380 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.NumericTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.PojoTypeInfo;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.dynamodb.table.converter.ArrayAttributeConverterProvider;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverter;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverterProvider;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeValueType;
+import software.amazon.awssdk.enhanced.dynamodb.EnhancedType;
+import software.amazon.awssdk.enhanced.dynamodb.TableSchema;
+import software.amazon.awssdk.enhanced.dynamodb.internal.mapper.BeanAttributeGetter;
+import software.amazon.awssdk.enhanced.dynamodb.mapper.StaticTableSchema;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.beans.BeanInfo;
+import java.beans.IntrospectionException;
+import java.beans.Introspector;
+import java.beans.PropertyDescriptor;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Function;
+
+/**
+ * A {@link ElementConverter} that converts an element to a {@link DynamoDbWriteRequest} using
+ * TypeInformation provided.
+ */
+@PublicEvolving
+public class DynamoDbTypeInformedElementConverter<inputT>
+ implements ElementConverter<inputT, DynamoDbWriteRequest> {
+ private final CompositeType<inputT> typeInfo;
+ private final boolean ignoreNulls;
+ private TableSchema<inputT> tableSchema;
+
+ /**
+ * Creates a {@link DynamoDbTypeInformedElementConverter} that converts an element to a {@link
+ * DynamoDbWriteRequest} using the provided {@link CompositeType}. Usage: {@code new
+ * DynamoDbTypeInformedElementConverter<>(TypeInformation.of(MyPojoClass.class))}
+ *
+ * @param typeInfo The {@link CompositeType} that provides the type information for the element.
+ */
+ public DynamoDbTypeInformedElementConverter(CompositeType<inputT> typeInfo) {
+ this(typeInfo, true);
+ }
+
+ public DynamoDbTypeInformedElementConverter(
+ CompositeType<inputT> typeInfo, boolean ignoreNulls) {
+ this.typeInfo = typeInfo;
+ this.ignoreNulls = ignoreNulls;
+ }
+
+ @Override
+ public void open(Sink.InitContext context) {
+ try {
+ tableSchema = createTableSchema(typeInfo);
+ } catch (IntrospectionException | IllegalStateException | IllegalArgumentException e) {
+ throw new FlinkRuntimeException("Failed to extract DynamoDb table schema", e);
+ }
+ }
+
+ @Override
+ public DynamoDbWriteRequest apply(inputT input, SinkWriter.Context context) {
+ Preconditions.checkNotNull(tableSchema, "TableSchema is not initialized");
+ try {
+ return DynamoDbWriteRequest.builder()
+ .setType(DynamoDbWriteRequestType.PUT)
+ .setItem(tableSchema.itemToMap(input, ignoreNulls))
+ .build();
+ } catch (ClassCastException | IllegalArgumentException e) {
+ throw new FlinkRuntimeException(
+ String.format(
+ "Failed to convert %s to DynamoDbWriteRequest using %s",
+ input, typeInfo),
+ e);
+ }
+ }
+
+ private <attrT> TableSchema<attrT> createTableSchema(CompositeType<attrT> typeInfo)
+ throws IntrospectionException {
+ if (typeInfo instanceof RowTypeInfo) {
+ return (TableSchema<attrT>) createTableSchemaFromRowType((RowTypeInfo) typeInfo);
+ } else if (typeInfo instanceof PojoTypeInfo<?>) {
+ return createTableSchemaFromPojo((PojoTypeInfo<attrT>) typeInfo);
+ } else if (typeInfo instanceof TupleTypeInfo<?>) {
+ return createTableSchemaFromTuple((TupleTypeInfo<?>) typeInfo);
+ } else {
+ throw new IllegalArgumentException(String.format("Unsupported TypeInfo %s", typeInfo));
+ }
+ }
+
+ private TableSchema<Row> createTableSchemaFromRowType(RowTypeInfo typeInfo) {
+ StaticTableSchema.Builder<Row> tableSchemaBuilder =
+ StaticTableSchema.builder(typeInfo.getTypeClass());
+
+ String[] fieldNames = typeInfo.getFieldNames();
+ for (int i = 0; i < typeInfo.getArity(); i++) {
+ TypeInformation<?> fieldType = typeInfo.getTypeAt(i);
+ int finalI = i;
+ addAttribute(
+ tableSchemaBuilder,
+ fieldNames[finalI],
+ (attrT) -> attrT.getField(finalI),
+ (TypeInformation<? super Object>) fieldType);
+ }
+
+ return tableSchemaBuilder.build();
+ }
+
+ private <attT> TableSchema<attT> createTableSchemaFromTuple(TupleTypeInfo<?> typeInfo) {
+ TypeInformation<?>[] fieldTypes = typeInfo.getFieldTypes();
+ String[] fieldNames = typeInfo.getFieldNames();
+
+ StaticTableSchema.Builder<?> tableSchemaBuilder =
+ StaticTableSchema.builder(typeInfo.getTypeClass());
+ for (int i = 0; i < fieldNames.length; i++) {
+ int finalI = i;
+ addAttribute(
+ tableSchemaBuilder,
+ fieldNames[finalI],
+ (attrT) -> ((Tuple) attrT).getField(finalI),
+ (TypeInformation<?>) fieldTypes[i]);
+ }
+
+ return (TableSchema<attT>) tableSchemaBuilder.build();
+ }
+
+ private <attrT> TableSchema<attrT> createTableSchemaFromPojo(PojoTypeInfo<attrT> typeInfo)
+ throws IntrospectionException {
+ StaticTableSchema.Builder<attrT> tableSchemaBuilder =
+ StaticTableSchema.builder(typeInfo.getTypeClass());
+ BeanInfo beanInfo = Introspector.getBeanInfo(typeInfo.getTypeClass());
+ PropertyDescriptor[] propertyDescriptors = beanInfo.getPropertyDescriptors();
Review Comment:
I am not sure I get what we need to cache, these descriptors are only created once when creating the schema, during element conversion only the Enhanced getters are used.
Btw, this is also what the BeanTableSchema does :D
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35022][Connector/DynamoDB] Add TypeInformed DDB Element Converter [flink-connector-aws]
Posted by "vahmed-hamdy (via GitHub)" <gi...@apache.org>.
vahmed-hamdy commented on code in PR #136:
URL: https://github.com/apache/flink-connector-aws/pull/136#discussion_r1581083313
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbTypeInformedElementConverter.java:
##########
@@ -0,0 +1,380 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.NumericTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.PojoTypeInfo;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.dynamodb.table.converter.ArrayAttributeConverterProvider;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverter;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverterProvider;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeValueType;
+import software.amazon.awssdk.enhanced.dynamodb.EnhancedType;
+import software.amazon.awssdk.enhanced.dynamodb.TableSchema;
+import software.amazon.awssdk.enhanced.dynamodb.internal.mapper.BeanAttributeGetter;
+import software.amazon.awssdk.enhanced.dynamodb.mapper.StaticTableSchema;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.beans.BeanInfo;
+import java.beans.IntrospectionException;
+import java.beans.Introspector;
+import java.beans.PropertyDescriptor;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Function;
+
+/**
+ * A {@link ElementConverter} that converts an element to a {@link DynamoDbWriteRequest} using
+ * TypeInformation provided.
+ */
+@PublicEvolving
+public class DynamoDbTypeInformedElementConverter<inputT>
Review Comment:
I agree, I would suggest `DefaultDynamoDbElementConverter` since we are following up with making it default and changing it to optional? wdyt?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35022][Connector/DynamoDB] Add TypeInformed DDB Element Converter as default element converter [flink-connector-aws]
Posted by "vahmed-hamdy (via GitHub)" <gi...@apache.org>.
vahmed-hamdy commented on PR #136:
URL: https://github.com/apache/flink-connector-aws/pull/136#issuecomment-2186211486
@hlteoh37 Do you have other comments or could we merge this PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35022][Connector/DynamoDB] Add TypeInformed DDB Element Converter as default element converter [flink-connector-aws]
Posted by "dannycranmer (via GitHub)" <gi...@apache.org>.
dannycranmer commented on code in PR #136:
URL: https://github.com/apache/flink-connector-aws/pull/136#discussion_r1633458080
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DefaultDynamoDbElementConverter.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+
+/**
+ * Default implementation of {@link ElementConverter} that lazily falls back to {@link
+ * DynamoDbTypeInformedElementConverter}.
+ */
+@PublicEvolving
+public class DefaultDynamoDbElementConverter<T>
+ implements ElementConverter<T, DynamoDbWriteRequest> {
+
+ private ElementConverter<T, DynamoDbWriteRequest> elementConverter;
+
+ public DefaultDynamoDbElementConverter() {}
+
+ @Override
+ public DynamoDbWriteRequest apply(T t, SinkWriter.Context context) {
+ if (elementConverter == null) {
+ TypeInformation<T> typeInfo = (TypeInformation<T>) TypeInformation.of(t.getClass());
+ if (!(typeInfo instanceof CompositeType<?>)) {
+ throw new IllegalArgumentException("The input type must be a CompositeType.");
+ }
+
+ elementConverter =
+ new DynamoDbTypeInformedElementConverter<>((CompositeType<T>) typeInfo);
+ }
+
+ return elementConverter.apply(t, context);
+ }
+
+ @Override
+ public void open(Sink.InitContext context) {}
Review Comment:
nit: Do we need this? I think it has a no-op default implementation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35022][Connector/DynamoDB] Add TypeInformed DDB Element Converter as default element converter [flink-connector-aws]
Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 commented on PR #136:
URL: https://github.com/apache/flink-connector-aws/pull/136#issuecomment-2186279507
Hi @vahmed-hamdy, my preference would be for us to refactor the code in `DynamoDbTypeInformedElementConverter` to use a cleaner structure rather than a long list of `if-else`. Is this something we need to close urgently?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35022][Connector/DynamoDB] Add TypeInformed DDB Element Converter as default element converter [flink-connector-aws]
Posted by "vahmed-hamdy (via GitHub)" <gi...@apache.org>.
vahmed-hamdy commented on PR #136:
URL: https://github.com/apache/flink-connector-aws/pull/136#issuecomment-2165938144
@hlteoh37 Thanks, for the feedback, the reason we took this approach is that We are trying to couple it as much as possible with Flink's TypeInfo Class, Using `AttributeConverterProvider` is closer to DDB's `EnhancedType` rather than Flink's
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35022][Connector/DynamoDB] Add TypeInformed DDB Element Converter [flink-connector-aws]
Posted by "vahmed-hamdy (via GitHub)" <gi...@apache.org>.
vahmed-hamdy commented on PR #136:
URL: https://github.com/apache/flink-connector-aws/pull/136#issuecomment-2067496399
@dannycranmer Thanks for the feedback, I have addressed your comments. Alternatively I decided to follow an approach similar to `RowDataToAttributeValueConverter` to avoid using reflection. PTAL
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35022][Connector/DynamoDB] Add TypeInformed DDB Element Converter [flink-connector-aws]
Posted by "dannycranmer (via GitHub)" <gi...@apache.org>.
dannycranmer commented on code in PR #136:
URL: https://github.com/apache/flink-connector-aws/pull/136#discussion_r1579301456
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbTypeInformedElementConverter.java:
##########
@@ -0,0 +1,380 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.NumericTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.PojoTypeInfo;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.dynamodb.table.converter.ArrayAttributeConverterProvider;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverter;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverterProvider;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeValueType;
+import software.amazon.awssdk.enhanced.dynamodb.EnhancedType;
+import software.amazon.awssdk.enhanced.dynamodb.TableSchema;
+import software.amazon.awssdk.enhanced.dynamodb.internal.mapper.BeanAttributeGetter;
+import software.amazon.awssdk.enhanced.dynamodb.mapper.StaticTableSchema;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.beans.BeanInfo;
+import java.beans.IntrospectionException;
+import java.beans.Introspector;
+import java.beans.PropertyDescriptor;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Function;
+
+/**
+ * A {@link ElementConverter} that converts an element to a {@link DynamoDbWriteRequest} using
+ * TypeInformation provided.
+ */
+@PublicEvolving
+public class DynamoDbTypeInformedElementConverter<inputT>
Review Comment:
nit: I am wondering if there is a better name for this? Also, we could make this the default element converter and promote the field to optional on the sink?
How about `AutoDynamoDbElementConverter` or `SimpleDynamoDbElementConverter`? I am not sold on any name yet!
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbTypeInformedElementConverter.java:
##########
@@ -0,0 +1,380 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.NumericTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.PojoTypeInfo;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.dynamodb.table.converter.ArrayAttributeConverterProvider;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverter;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverterProvider;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeValueType;
+import software.amazon.awssdk.enhanced.dynamodb.EnhancedType;
+import software.amazon.awssdk.enhanced.dynamodb.TableSchema;
+import software.amazon.awssdk.enhanced.dynamodb.internal.mapper.BeanAttributeGetter;
+import software.amazon.awssdk.enhanced.dynamodb.mapper.StaticTableSchema;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.beans.BeanInfo;
+import java.beans.IntrospectionException;
+import java.beans.Introspector;
+import java.beans.PropertyDescriptor;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Function;
+
+/**
+ * A {@link ElementConverter} that converts an element to a {@link DynamoDbWriteRequest} using
+ * TypeInformation provided.
+ */
+@PublicEvolving
+public class DynamoDbTypeInformedElementConverter<inputT>
+ implements ElementConverter<inputT, DynamoDbWriteRequest> {
+ private final CompositeType<inputT> typeInfo;
Review Comment:
Generic types should be `UpperCamelCase` :: `CompositeType<inputT>` -> `CompositeType<InputT>`
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbTypeInformedElementConverter.java:
##########
@@ -0,0 +1,380 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.NumericTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.PojoTypeInfo;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.dynamodb.table.converter.ArrayAttributeConverterProvider;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverter;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverterProvider;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeValueType;
+import software.amazon.awssdk.enhanced.dynamodb.EnhancedType;
+import software.amazon.awssdk.enhanced.dynamodb.TableSchema;
+import software.amazon.awssdk.enhanced.dynamodb.internal.mapper.BeanAttributeGetter;
+import software.amazon.awssdk.enhanced.dynamodb.mapper.StaticTableSchema;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.beans.BeanInfo;
+import java.beans.IntrospectionException;
+import java.beans.Introspector;
+import java.beans.PropertyDescriptor;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Function;
+
+/**
+ * A {@link ElementConverter} that converts an element to a {@link DynamoDbWriteRequest} using
+ * TypeInformation provided.
+ */
+@PublicEvolving
+public class DynamoDbTypeInformedElementConverter<inputT>
+ implements ElementConverter<inputT, DynamoDbWriteRequest> {
+ private final CompositeType<inputT> typeInfo;
+ private final boolean ignoreNulls;
+ private TableSchema<inputT> tableSchema;
+
+ /**
+ * Creates a {@link DynamoDbTypeInformedElementConverter} that converts an element to a {@link
+ * DynamoDbWriteRequest} using the provided {@link CompositeType}. Usage: {@code new
+ * DynamoDbTypeInformedElementConverter<>(TypeInformation.of(MyPojoClass.class))}
+ *
+ * @param typeInfo The {@link CompositeType} that provides the type information for the element.
+ */
+ public DynamoDbTypeInformedElementConverter(CompositeType<inputT> typeInfo) {
+ this(typeInfo, true);
+ }
+
+ public DynamoDbTypeInformedElementConverter(
+ CompositeType<inputT> typeInfo, boolean ignoreNulls) {
+ this.typeInfo = typeInfo;
+ this.ignoreNulls = ignoreNulls;
+ }
+
+ @Override
+ public void open(Sink.InitContext context) {
+ try {
+ tableSchema = createTableSchema(typeInfo);
+ } catch (IntrospectionException | IllegalStateException | IllegalArgumentException e) {
+ throw new FlinkRuntimeException("Failed to extract DynamoDb table schema", e);
+ }
Review Comment:
We should do this in the constructor as a validation and fail fast. Do not need to store, but nice to fail on client and not when running job
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbTypeInformedElementConverter.java:
##########
@@ -0,0 +1,380 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.NumericTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.PojoTypeInfo;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.dynamodb.table.converter.ArrayAttributeConverterProvider;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverter;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverterProvider;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeValueType;
+import software.amazon.awssdk.enhanced.dynamodb.EnhancedType;
+import software.amazon.awssdk.enhanced.dynamodb.TableSchema;
+import software.amazon.awssdk.enhanced.dynamodb.internal.mapper.BeanAttributeGetter;
+import software.amazon.awssdk.enhanced.dynamodb.mapper.StaticTableSchema;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.beans.BeanInfo;
+import java.beans.IntrospectionException;
+import java.beans.Introspector;
+import java.beans.PropertyDescriptor;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Function;
+
+/**
+ * A {@link ElementConverter} that converts an element to a {@link DynamoDbWriteRequest} using
+ * TypeInformation provided.
+ */
+@PublicEvolving
+public class DynamoDbTypeInformedElementConverter<inputT>
+ implements ElementConverter<inputT, DynamoDbWriteRequest> {
+ private final CompositeType<inputT> typeInfo;
+ private final boolean ignoreNulls;
+ private TableSchema<inputT> tableSchema;
+
+ /**
+ * Creates a {@link DynamoDbTypeInformedElementConverter} that converts an element to a {@link
+ * DynamoDbWriteRequest} using the provided {@link CompositeType}. Usage: {@code new
+ * DynamoDbTypeInformedElementConverter<>(TypeInformation.of(MyPojoClass.class))}
+ *
+ * @param typeInfo The {@link CompositeType} that provides the type information for the element.
+ */
+ public DynamoDbTypeInformedElementConverter(CompositeType<inputT> typeInfo) {
+ this(typeInfo, true);
+ }
+
+ public DynamoDbTypeInformedElementConverter(
+ CompositeType<inputT> typeInfo, boolean ignoreNulls) {
+ this.typeInfo = typeInfo;
+ this.ignoreNulls = ignoreNulls;
+ }
+
+ @Override
+ public void open(Sink.InitContext context) {
+ try {
+ tableSchema = createTableSchema(typeInfo);
+ } catch (IntrospectionException | IllegalStateException | IllegalArgumentException e) {
+ throw new FlinkRuntimeException("Failed to extract DynamoDb table schema", e);
+ }
+ }
+
+ @Override
+ public DynamoDbWriteRequest apply(inputT input, SinkWriter.Context context) {
+ Preconditions.checkNotNull(tableSchema, "TableSchema is not initialized");
+ try {
+ return DynamoDbWriteRequest.builder()
+ .setType(DynamoDbWriteRequestType.PUT)
+ .setItem(tableSchema.itemToMap(input, ignoreNulls))
+ .build();
+ } catch (ClassCastException | IllegalArgumentException e) {
+ throw new FlinkRuntimeException(
+ String.format(
+ "Failed to convert %s to DynamoDbWriteRequest using %s",
+ input, typeInfo),
+ e);
+ }
+ }
+
+ private <attrT> TableSchema<attrT> createTableSchema(CompositeType<attrT> typeInfo)
+ throws IntrospectionException {
+ if (typeInfo instanceof RowTypeInfo) {
+ return (TableSchema<attrT>) createTableSchemaFromRowType((RowTypeInfo) typeInfo);
+ } else if (typeInfo instanceof PojoTypeInfo<?>) {
+ return createTableSchemaFromPojo((PojoTypeInfo<attrT>) typeInfo);
+ } else if (typeInfo instanceof TupleTypeInfo<?>) {
+ return createTableSchemaFromTuple((TupleTypeInfo<?>) typeInfo);
+ } else {
+ throw new IllegalArgumentException(String.format("Unsupported TypeInfo %s", typeInfo));
+ }
+ }
+
+ private TableSchema<Row> createTableSchemaFromRowType(RowTypeInfo typeInfo) {
+ StaticTableSchema.Builder<Row> tableSchemaBuilder =
+ StaticTableSchema.builder(typeInfo.getTypeClass());
+
+ String[] fieldNames = typeInfo.getFieldNames();
+ for (int i = 0; i < typeInfo.getArity(); i++) {
+ TypeInformation<?> fieldType = typeInfo.getTypeAt(i);
+ int finalI = i;
+ addAttribute(
+ tableSchemaBuilder,
+ fieldNames[finalI],
+ (attrT) -> attrT.getField(finalI),
+ (TypeInformation<? super Object>) fieldType);
+ }
+
+ return tableSchemaBuilder.build();
+ }
+
+ private <attT> TableSchema<attT> createTableSchemaFromTuple(TupleTypeInfo<?> typeInfo) {
+ TypeInformation<?>[] fieldTypes = typeInfo.getFieldTypes();
+ String[] fieldNames = typeInfo.getFieldNames();
+
+ StaticTableSchema.Builder<?> tableSchemaBuilder =
+ StaticTableSchema.builder(typeInfo.getTypeClass());
+ for (int i = 0; i < fieldNames.length; i++) {
+ int finalI = i;
+ addAttribute(
+ tableSchemaBuilder,
+ fieldNames[finalI],
+ (attrT) -> ((Tuple) attrT).getField(finalI),
+ (TypeInformation<?>) fieldTypes[i]);
+ }
+
+ return (TableSchema<attT>) tableSchemaBuilder.build();
+ }
+
+ private <attrT> TableSchema<attrT> createTableSchemaFromPojo(PojoTypeInfo<attrT> typeInfo)
+ throws IntrospectionException {
+ StaticTableSchema.Builder<attrT> tableSchemaBuilder =
+ StaticTableSchema.builder(typeInfo.getTypeClass());
+ BeanInfo beanInfo = Introspector.getBeanInfo(typeInfo.getTypeClass());
+ PropertyDescriptor[] propertyDescriptors = beanInfo.getPropertyDescriptors();
Review Comment:
This looks like it might be expensive, can we cache the descriptors per type?
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbTypeInformedElementConverter.java:
##########
@@ -0,0 +1,380 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.NumericTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.PojoTypeInfo;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.dynamodb.table.converter.ArrayAttributeConverterProvider;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverter;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverterProvider;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeValueType;
+import software.amazon.awssdk.enhanced.dynamodb.EnhancedType;
+import software.amazon.awssdk.enhanced.dynamodb.TableSchema;
+import software.amazon.awssdk.enhanced.dynamodb.internal.mapper.BeanAttributeGetter;
+import software.amazon.awssdk.enhanced.dynamodb.mapper.StaticTableSchema;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.beans.BeanInfo;
+import java.beans.IntrospectionException;
+import java.beans.Introspector;
+import java.beans.PropertyDescriptor;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Function;
+
+/**
+ * A {@link ElementConverter} that converts an element to a {@link DynamoDbWriteRequest} using
+ * TypeInformation provided.
+ */
+@PublicEvolving
+public class DynamoDbTypeInformedElementConverter<inputT>
+ implements ElementConverter<inputT, DynamoDbWriteRequest> {
+ private final CompositeType<inputT> typeInfo;
Review Comment:
I personally just prefer `<T>` when it is clear what it means
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35022][Connector/DynamoDB] Add TypeInformed DDB Element Converter as default element converter [flink-connector-aws]
Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 commented on code in PR #136:
URL: https://github.com/apache/flink-connector-aws/pull/136#discussion_r1634758095
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbTypeInformedElementConverter.java:
##########
@@ -0,0 +1,380 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.NumericTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.PojoTypeInfo;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.dynamodb.table.converter.ArrayAttributeConverterProvider;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverter;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverterProvider;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeValueType;
+import software.amazon.awssdk.enhanced.dynamodb.EnhancedType;
+import software.amazon.awssdk.enhanced.dynamodb.TableSchema;
+import software.amazon.awssdk.enhanced.dynamodb.internal.mapper.BeanAttributeGetter;
+import software.amazon.awssdk.enhanced.dynamodb.mapper.StaticTableSchema;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.beans.BeanInfo;
+import java.beans.IntrospectionException;
+import java.beans.Introspector;
+import java.beans.PropertyDescriptor;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Function;
+
+/**
+ * A {@link ElementConverter} that converts an element to a {@link DynamoDbWriteRequest} using
+ * TypeInformation provided.
+ */
+@PublicEvolving
+public class DynamoDbTypeInformedElementConverter<T>
+ implements ElementConverter<T, DynamoDbWriteRequest> {
+
+ private final CompositeType<T> typeInfo;
+ private final boolean ignoreNulls;
+ private final TableSchema<T> tableSchema;
+
+ /**
+ * Creates a {@link DynamoDbTypeInformedElementConverter} that converts an element to a {@link
+ * DynamoDbWriteRequest} using the provided {@link CompositeType}. Usage: {@code new
+ * DynamoDbTypeInformedElementConverter<>(TypeInformation.of(MyPojoClass.class))}
+ *
+ * @param typeInfo The {@link CompositeType} that provides the type information for the element.
+ */
+ public DynamoDbTypeInformedElementConverter(CompositeType<T> typeInfo) {
+ this(typeInfo, true);
+ }
+
+ public DynamoDbTypeInformedElementConverter(CompositeType<T> typeInfo, boolean ignoreNulls) {
+
+ try {
+ this.typeInfo = typeInfo;
+ this.ignoreNulls = ignoreNulls;
+ this.tableSchema = createTableSchema(typeInfo);
+ } catch (IntrospectionException | IllegalStateException | IllegalArgumentException e) {
+ throw new FlinkRuntimeException("Failed to extract DynamoDb table schema", e);
+ }
+ }
+
+ @Override
+ public DynamoDbWriteRequest apply(T input, SinkWriter.Context context) {
+ Preconditions.checkNotNull(tableSchema, "TableSchema is not initialized");
+ try {
+ return DynamoDbWriteRequest.builder()
+ .setType(DynamoDbWriteRequestType.PUT)
+ .setItem(tableSchema.itemToMap(input, ignoreNulls))
+ .build();
+ } catch (ClassCastException | IllegalArgumentException e) {
+ throw new FlinkRuntimeException(
+ String.format(
+ "Failed to convert %s to DynamoDbWriteRequest using %s",
+ input, typeInfo),
+ e);
+ }
+ }
+
+ private <AttributeT> TableSchema<AttributeT> createTableSchema(
+ CompositeType<AttributeT> typeInfo) throws IntrospectionException {
+ if (typeInfo instanceof RowTypeInfo) {
+ return (TableSchema<AttributeT>) createTableSchemaFromRowType((RowTypeInfo) typeInfo);
+ } else if (typeInfo instanceof PojoTypeInfo<?>) {
+ return createTableSchemaFromPojo((PojoTypeInfo<AttributeT>) typeInfo);
+ } else if (typeInfo instanceof TupleTypeInfo<?>) {
+ return createTableSchemaFromTuple((TupleTypeInfo<?>) typeInfo);
+ } else {
+ throw new IllegalArgumentException(String.format("Unsupported TypeInfo %s", typeInfo));
+ }
+ }
+
+ private TableSchema<Row> createTableSchemaFromRowType(RowTypeInfo typeInfo) {
+ StaticTableSchema.Builder<Row> tableSchemaBuilder =
+ StaticTableSchema.builder(typeInfo.getTypeClass());
+
+ String[] fieldNames = typeInfo.getFieldNames();
+ for (int i = 0; i < typeInfo.getArity(); i++) {
+ TypeInformation<?> fieldType = typeInfo.getTypeAt(i);
+ int finalI = i;
+ addAttribute(
+ tableSchemaBuilder,
+ fieldNames[finalI],
+ (AttributeT) -> AttributeT.getField(finalI),
+ (TypeInformation<? super Object>) fieldType);
+ }
+
+ return tableSchemaBuilder.build();
+ }
+
+ private <AttributeT> TableSchema<AttributeT> createTableSchemaFromTuple(
+ TupleTypeInfo<?> typeInfo) {
+ TypeInformation<?>[] fieldTypes = typeInfo.getFieldTypes();
+ String[] fieldNames = typeInfo.getFieldNames();
+
+ StaticTableSchema.Builder<?> tableSchemaBuilder =
+ StaticTableSchema.builder(typeInfo.getTypeClass());
+ for (int i = 0; i < fieldNames.length; i++) {
+ int finalI = i;
+ addAttribute(
+ tableSchemaBuilder,
+ fieldNames[finalI],
+ (tuple) -> ((Tuple) tuple).getField(finalI),
+ (TypeInformation<?>) fieldTypes[i]);
+ }
+
+ return (TableSchema<AttributeT>) tableSchemaBuilder.build();
+ }
+
+ private <AttributeT> TableSchema<AttributeT> createTableSchemaFromPojo(
+ PojoTypeInfo<AttributeT> typeInfo) throws IntrospectionException {
+ StaticTableSchema.Builder<AttributeT> tableSchemaBuilder =
+ StaticTableSchema.builder(typeInfo.getTypeClass());
+ BeanInfo beanInfo = Introspector.getBeanInfo(typeInfo.getTypeClass());
+ PropertyDescriptor[] propertyDescriptors = beanInfo.getPropertyDescriptors();
+ for (PropertyDescriptor propertyDescriptor : propertyDescriptors) {
+ Set<String> fieldNames = new HashSet<>(Arrays.asList(typeInfo.getFieldNames()));
+ if (!fieldNames.contains(propertyDescriptor.getName())) {
+ // Skip properties that are not part of the PojoTypeInfo
+ continue;
+ }
+
+ TypeInformation<?> fieldInfo = typeInfo.getTypeAt(propertyDescriptor.getName());
+ addAttribute(
+ tableSchemaBuilder,
+ propertyDescriptor.getName(),
+ BeanAttributeGetter.create(
+ typeInfo.getTypeClass(), propertyDescriptor.getReadMethod()),
+ fieldInfo);
+ }
+
+ return tableSchemaBuilder.build();
+ }
+
+ private <T, AttributeT> void addAttribute(
+ StaticTableSchema.Builder<T> builder,
+ String fieldName,
+ Function<T, AttributeT> getter,
+ TypeInformation<AttributeT> typeInfo) {
+ builder.addAttribute(
+ typeInfo.getTypeClass(),
+ a ->
+ a.name(fieldName)
+ .getter(getter)
+ .setter((e, o) -> {})
+ .attributeConverter(getAttributeConverter(typeInfo)));
+ }
+
+ private <AttributeT> AttributeConverter<AttributeT> getAttributeConverter(
+ TypeInformation<AttributeT> typeInfo) {
+ if (typeInfo instanceof BasicTypeInfo) {
+ return AttributeConverterProvider.defaultProvider()
+ .converterFor(EnhancedType.of(typeInfo.getTypeClass()));
+ } else if (typeInfo.equals(PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO)) {
+ return getAttributeConverter(
+ AttributeValueType.B,
+ bytes ->
+ bytes instanceof SdkBytes
+ ? AttributeValue.fromB((SdkBytes) bytes)
+ : AttributeValue.fromB(SdkBytes.fromByteArray((byte[]) bytes)));
+ } else if (typeInfo instanceof BasicArrayTypeInfo) {
+ BasicArrayTypeInfo<AttributeT, ?> basicArrayTypeInfo =
+ (BasicArrayTypeInfo<AttributeT, ?>) typeInfo;
+ if (basicArrayTypeInfo.getComponentInfo().equals(BasicTypeInfo.STRING_TYPE_INFO)) {
+ return getAttributeConverter(
+ AttributeValueType.SS,
+ array -> AttributeValue.fromSs(Arrays.asList((String[]) array)));
+ } else if (basicArrayTypeInfo.getComponentInfo() instanceof NumericTypeInfo) {
+ return getAttributeConverter(
+ AttributeValueType.NS,
+ array ->
+ AttributeValue.fromNs(
+ convertObjectArrayToStringList((Object[]) array)));
+ }
+
+ return new ArrayAttributeConverterProvider()
+ .converterFor(EnhancedType.of(typeInfo.getTypeClass()));
+ } else if (typeInfo instanceof ObjectArrayTypeInfo) {
+ return getObjectArrayTypeConverter((ObjectArrayTypeInfo<AttributeT, ?>) typeInfo);
+ } else if (typeInfo instanceof PrimitiveArrayTypeInfo) {
+ PrimitiveArrayTypeInfo<AttributeT> primitiveArrayTypeInfo =
+ (PrimitiveArrayTypeInfo<AttributeT>) typeInfo;
+ if (primitiveArrayTypeInfo.getComponentType() instanceof NumericTypeInfo) {
+ return getAttributeConverter(
+ AttributeValueType.NS,
+ array -> AttributeValue.fromNs(convertPrimitiveArrayToStringList(array)));
+ } else {
+ throw new IllegalArgumentException(
+ String.format(
+ "Unsupported primitive array typeInfo %s",
+ primitiveArrayTypeInfo.getComponentType()));
+ }
Review Comment:
Have we considered using the `.attributeConverterProviders()` method in the table schema builder and reusing the constructs we use to DDB Table API? It might be good to reuse code here instead of duplicating!
```
TableSchema.builder(...)
.attributeConverterProviders(
new ArrayAttributeConverterProvider(),
... (add any more here) ...,
AttributeConverterProvider.defaultProvider());
```
We have an example here:
https://github.com/apache/flink-connector-aws/blob/main/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverter.java#L56-L72
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35022] Add TypeInformed DDB Element Converter [flink-connector-aws]
Posted by "dannycranmer (via GitHub)" <gi...@apache.org>.
dannycranmer commented on code in PR #136:
URL: https://github.com/apache/flink-connector-aws/pull/136#discussion_r1565419487
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbTypeInformedElementConverter.java:
##########
@@ -0,0 +1,245 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.NumericTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * A {@link ElementConverter} that converts an element to a {@link DynamoDbWriteRequest} using
+ * TypeInformation provided.
+ */
+@PublicEvolving
+public class DynamoDbTypeInformedElementConverter<inputT>
+ implements ElementConverter<inputT, DynamoDbWriteRequest> {
+ private final CompositeType<inputT> typeInfo;
+
+ /**
+ * Creates a {@link DynamoDbTypeInformedElementConverter} that converts an element to a {@link
+ * DynamoDbWriteRequest} using the provided {@link CompositeType}. Usage: {@code new
+ * DynamoDbTypeInformedElementConverter<>(TypeInformation.of(MyPojoClass.class))}
+ *
+ * @param typeInfo The {@link CompositeType} that provides the type information for the element.
+ */
+ public DynamoDbTypeInformedElementConverter(CompositeType<inputT> typeInfo) {
+ this.typeInfo = typeInfo;
+ }
+
+ @Override
+ public DynamoDbWriteRequest apply(inputT input, SinkWriter.Context context) {
+ try {
+ return DynamoDbWriteRequest.builder()
+ .setType(DynamoDbWriteRequestType.PUT)
+ .setItem(convertElementUsingTypeInfo(input, typeInfo))
+ .build();
+ } catch (IllegalArgumentException e) {
+ throw new FlinkRuntimeException("Couldn't convert Element to AttributeVal", e);
+ }
+ }
+
+ private <attT> Map<String, AttributeValue> convertElementUsingTypeInfo(
+ attT t, CompositeType<attT> typeInfo) {
+ Map<String, AttributeValue> map = new HashMap<>();
+ for (String fieldKey : typeInfo.getFieldNames()) {
+ TypeInformation<attT> fieldType = typeInfo.getTypeAt(fieldKey);
+ try {
+ Field field = t.getClass().getDeclaredField(fieldKey);
+ field.setAccessible(true);
+ Object fieldVal = field.get(t);
+ checkTypeCompatibility(fieldVal, fieldType);
+ attT fieldValCaster = (attT) fieldVal;
+ map.put(fieldKey, convertValue(fieldValCaster, fieldType));
+ } catch (NoSuchFieldException | IllegalAccessException e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Failed to extract field %s declared in TypeInfo "
+ + "from Object %s",
+ fieldKey, t),
+ e);
+ }
+ }
+
+ return map;
+ }
+
+ private <attrT> AttributeValue convertValue(
+ attrT attribute, TypeInformation<attrT> objectTypeInformation) {
+ if (attribute == null) {
+ return AttributeValue.builder().nul(true).build();
Review Comment:
Not sure this is the correct thing to do here? I believe a common practise is to omit null fields rather than setting to null. Maybe this could be a configuration on the element converter?
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbTypeInformedElementConverter.java:
##########
@@ -0,0 +1,245 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.NumericTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * A {@link ElementConverter} that converts an element to a {@link DynamoDbWriteRequest} using
+ * TypeInformation provided.
+ */
+@PublicEvolving
+public class DynamoDbTypeInformedElementConverter<inputT>
+ implements ElementConverter<inputT, DynamoDbWriteRequest> {
+ private final CompositeType<inputT> typeInfo;
+
+ /**
+ * Creates a {@link DynamoDbTypeInformedElementConverter} that converts an element to a {@link
+ * DynamoDbWriteRequest} using the provided {@link CompositeType}. Usage: {@code new
+ * DynamoDbTypeInformedElementConverter<>(TypeInformation.of(MyPojoClass.class))}
+ *
+ * @param typeInfo The {@link CompositeType} that provides the type information for the element.
+ */
+ public DynamoDbTypeInformedElementConverter(CompositeType<inputT> typeInfo) {
+ this.typeInfo = typeInfo;
+ }
+
+ @Override
+ public DynamoDbWriteRequest apply(inputT input, SinkWriter.Context context) {
+ try {
+ return DynamoDbWriteRequest.builder()
+ .setType(DynamoDbWriteRequestType.PUT)
+ .setItem(convertElementUsingTypeInfo(input, typeInfo))
+ .build();
+ } catch (IllegalArgumentException e) {
+ throw new FlinkRuntimeException("Couldn't convert Element to AttributeVal", e);
+ }
+ }
+
+ private <attT> Map<String, AttributeValue> convertElementUsingTypeInfo(
+ attT t, CompositeType<attT> typeInfo) {
+ Map<String, AttributeValue> map = new HashMap<>();
+ for (String fieldKey : typeInfo.getFieldNames()) {
+ TypeInformation<attT> fieldType = typeInfo.getTypeAt(fieldKey);
+ try {
+ Field field = t.getClass().getDeclaredField(fieldKey);
+ field.setAccessible(true);
+ Object fieldVal = field.get(t);
+ checkTypeCompatibility(fieldVal, fieldType);
+ attT fieldValCaster = (attT) fieldVal;
+ map.put(fieldKey, convertValue(fieldValCaster, fieldType));
+ } catch (NoSuchFieldException | IllegalAccessException e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Failed to extract field %s declared in TypeInfo "
+ + "from Object %s",
+ fieldKey, t),
+ e);
+ }
+ }
+
+ return map;
+ }
+
+ private <attrT> AttributeValue convertValue(
+ attrT attribute, TypeInformation<attrT> objectTypeInformation) {
+ if (attribute == null) {
+ return AttributeValue.builder().nul(true).build();
+ } else if (objectTypeInformation.isBasicType()) {
+ return convertBasicTypeObject(attribute, (BasicTypeInfo<?>) objectTypeInformation);
+ } else if (objectTypeInformation.equals(
+ PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO)) {
+ // special handling for byte array as it has special support in DynamoDB
+ if (attribute instanceof byte[]) {
+ return AttributeValue.fromB(SdkBytes.fromByteArray((byte[]) attribute));
+ } else if (attribute instanceof SdkBytes) {
+ return AttributeValue.fromB((SdkBytes) attribute);
+ } else {
+ throw new IllegalArgumentException(
+ String.format("Failed to extract byte array type from %s", attribute));
+ }
Review Comment:
This block is untested, please add tests
##########
flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/DynamoDbTypeInformedElementConverterTest.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.tuple.Tuple1;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.connector.dynamodb.util.ComplexPayload;
+import org.apache.flink.connector.dynamodb.util.Order;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.flink.connector.dynamodb.sink.DynamoDbWriteRequestType.PUT;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link DynamoDbTypeInformedElementConverter}. */
+public class DynamoDbTypeInformedElementConverterTest {
+
+ @Test
+ public void testSimpleTypeConversion() {
Review Comment:
nit: JUnit 5 coding standards are slightly different from lower versions. In 5 we:
1/ Use package protected class/methods (drop `public`)
2/ Remove the `test` prefix on the methods (`testSimpleTypeConversion()` -> `simpleTypeConversion()`)
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbTypeInformedElementConverter.java:
##########
@@ -0,0 +1,245 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.NumericTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * A {@link ElementConverter} that converts an element to a {@link DynamoDbWriteRequest} using
+ * TypeInformation provided.
+ */
+@PublicEvolving
+public class DynamoDbTypeInformedElementConverter<inputT>
+ implements ElementConverter<inputT, DynamoDbWriteRequest> {
+ private final CompositeType<inputT> typeInfo;
+
+ /**
+ * Creates a {@link DynamoDbTypeInformedElementConverter} that converts an element to a {@link
+ * DynamoDbWriteRequest} using the provided {@link CompositeType}. Usage: {@code new
+ * DynamoDbTypeInformedElementConverter<>(TypeInformation.of(MyPojoClass.class))}
+ *
+ * @param typeInfo The {@link CompositeType} that provides the type information for the element.
+ */
+ public DynamoDbTypeInformedElementConverter(CompositeType<inputT> typeInfo) {
+ this.typeInfo = typeInfo;
+ }
+
+ @Override
+ public DynamoDbWriteRequest apply(inputT input, SinkWriter.Context context) {
+ try {
+ return DynamoDbWriteRequest.builder()
+ .setType(DynamoDbWriteRequestType.PUT)
+ .setItem(convertElementUsingTypeInfo(input, typeInfo))
+ .build();
+ } catch (IllegalArgumentException e) {
+ throw new FlinkRuntimeException("Couldn't convert Element to AttributeVal", e);
+ }
+ }
+
+ private <attT> Map<String, AttributeValue> convertElementUsingTypeInfo(
+ attT t, CompositeType<attT> typeInfo) {
+ Map<String, AttributeValue> map = new HashMap<>();
+ for (String fieldKey : typeInfo.getFieldNames()) {
+ TypeInformation<attT> fieldType = typeInfo.getTypeAt(fieldKey);
+ try {
+ Field field = t.getClass().getDeclaredField(fieldKey);
+ field.setAccessible(true);
+ Object fieldVal = field.get(t);
+ checkTypeCompatibility(fieldVal, fieldType);
+ attT fieldValCaster = (attT) fieldVal;
+ map.put(fieldKey, convertValue(fieldValCaster, fieldType));
+ } catch (NoSuchFieldException | IllegalAccessException e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Failed to extract field %s declared in TypeInfo "
+ + "from Object %s",
+ fieldKey, t),
+ e);
+ }
+ }
+
+ return map;
+ }
+
+ private <attrT> AttributeValue convertValue(
+ attrT attribute, TypeInformation<attrT> objectTypeInformation) {
+ if (attribute == null) {
+ return AttributeValue.builder().nul(true).build();
+ } else if (objectTypeInformation.isBasicType()) {
+ return convertBasicTypeObject(attribute, (BasicTypeInfo<?>) objectTypeInformation);
+ } else if (objectTypeInformation.equals(
+ PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO)) {
+ // special handling for byte array as it has special support in DynamoDB
+ if (attribute instanceof byte[]) {
+ return AttributeValue.fromB(SdkBytes.fromByteArray((byte[]) attribute));
+ } else if (attribute instanceof SdkBytes) {
+ return AttributeValue.fromB((SdkBytes) attribute);
+ } else {
+ throw new IllegalArgumentException(
+ String.format("Failed to extract byte array type from %s", attribute));
+ }
+
+ } else if (isArrayTypeInfo(objectTypeInformation)) {
+ TypeInformation<?> componentTypeInfo =
+ objectTypeInformation instanceof BasicArrayTypeInfo
+ ? ((BasicArrayTypeInfo<?, ?>) objectTypeInformation).getComponentInfo()
+ : objectTypeInformation instanceof PrimitiveArrayTypeInfo
+ ? ((PrimitiveArrayTypeInfo<?>) objectTypeInformation)
+ .getComponentType()
+ : ((ObjectArrayTypeInfo<?, ?>) objectTypeInformation)
+ .getComponentInfo();
+ if (attribute instanceof Collection) {
+ return convertArrayTypeObject(
+ ((Collection<?>) attribute).toArray(), componentTypeInfo, false);
+ }
+
+ return convertArrayTypeObject(
+ attribute,
+ componentTypeInfo,
+ objectTypeInformation instanceof PrimitiveArrayTypeInfo);
+
+ } else if (objectTypeInformation instanceof TupleTypeInfo) {
+ TupleTypeInfo<?> tupleTypeInfo = (TupleTypeInfo<?>) objectTypeInformation;
+ Tuple t = (Tuple) attribute;
+ List<AttributeValue> attributeValues = new ArrayList<>();
+ for (int i = 0; i < tupleTypeInfo.getTotalFields(); i++) {
+ attributeValues.add(convertValue(t.getField(i), tupleTypeInfo.getTypeAt(i)));
+ }
+ return AttributeValue.fromL(attributeValues);
Review Comment:
This block is untested, please add tests
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbTypeInformedElementConverter.java:
##########
@@ -0,0 +1,245 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.NumericTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * A {@link ElementConverter} that converts an element to a {@link DynamoDbWriteRequest} using
+ * TypeInformation provided.
+ */
+@PublicEvolving
+public class DynamoDbTypeInformedElementConverter<inputT>
+ implements ElementConverter<inputT, DynamoDbWriteRequest> {
+ private final CompositeType<inputT> typeInfo;
+
+ /**
+ * Creates a {@link DynamoDbTypeInformedElementConverter} that converts an element to a {@link
+ * DynamoDbWriteRequest} using the provided {@link CompositeType}. Usage: {@code new
+ * DynamoDbTypeInformedElementConverter<>(TypeInformation.of(MyPojoClass.class))}
+ *
+ * @param typeInfo The {@link CompositeType} that provides the type information for the element.
+ */
+ public DynamoDbTypeInformedElementConverter(CompositeType<inputT> typeInfo) {
+ this.typeInfo = typeInfo;
+ }
+
+ @Override
+ public DynamoDbWriteRequest apply(inputT input, SinkWriter.Context context) {
+ try {
+ return DynamoDbWriteRequest.builder()
+ .setType(DynamoDbWriteRequestType.PUT)
+ .setItem(convertElementUsingTypeInfo(input, typeInfo))
+ .build();
+ } catch (IllegalArgumentException e) {
+ throw new FlinkRuntimeException("Couldn't convert Element to AttributeVal", e);
+ }
+ }
+
+ private <attT> Map<String, AttributeValue> convertElementUsingTypeInfo(
+ attT t, CompositeType<attT> typeInfo) {
+ Map<String, AttributeValue> map = new HashMap<>();
+ for (String fieldKey : typeInfo.getFieldNames()) {
+ TypeInformation<attT> fieldType = typeInfo.getTypeAt(fieldKey);
+ try {
+ Field field = t.getClass().getDeclaredField(fieldKey);
+ field.setAccessible(true);
+ Object fieldVal = field.get(t);
+ checkTypeCompatibility(fieldVal, fieldType);
+ attT fieldValCaster = (attT) fieldVal;
+ map.put(fieldKey, convertValue(fieldValCaster, fieldType));
+ } catch (NoSuchFieldException | IllegalAccessException e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Failed to extract field %s declared in TypeInfo "
+ + "from Object %s",
+ fieldKey, t),
+ e);
+ }
+ }
+
+ return map;
+ }
+
+ private <attrT> AttributeValue convertValue(
+ attrT attribute, TypeInformation<attrT> objectTypeInformation) {
+ if (attribute == null) {
+ return AttributeValue.builder().nul(true).build();
+ } else if (objectTypeInformation.isBasicType()) {
+ return convertBasicTypeObject(attribute, (BasicTypeInfo<?>) objectTypeInformation);
+ } else if (objectTypeInformation.equals(
+ PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO)) {
+ // special handling for byte array as it has special support in DynamoDB
+ if (attribute instanceof byte[]) {
+ return AttributeValue.fromB(SdkBytes.fromByteArray((byte[]) attribute));
+ } else if (attribute instanceof SdkBytes) {
+ return AttributeValue.fromB((SdkBytes) attribute);
+ } else {
+ throw new IllegalArgumentException(
+ String.format("Failed to extract byte array type from %s", attribute));
+ }
+
+ } else if (isArrayTypeInfo(objectTypeInformation)) {
+ TypeInformation<?> componentTypeInfo =
+ objectTypeInformation instanceof BasicArrayTypeInfo
+ ? ((BasicArrayTypeInfo<?, ?>) objectTypeInformation).getComponentInfo()
+ : objectTypeInformation instanceof PrimitiveArrayTypeInfo
+ ? ((PrimitiveArrayTypeInfo<?>) objectTypeInformation)
+ .getComponentType()
+ : ((ObjectArrayTypeInfo<?, ?>) objectTypeInformation)
+ .getComponentInfo();
+ if (attribute instanceof Collection) {
+ return convertArrayTypeObject(
+ ((Collection<?>) attribute).toArray(), componentTypeInfo, false);
+ }
+
+ return convertArrayTypeObject(
+ attribute,
+ componentTypeInfo,
+ objectTypeInformation instanceof PrimitiveArrayTypeInfo);
+
+ } else if (objectTypeInformation instanceof TupleTypeInfo) {
+ TupleTypeInfo<?> tupleTypeInfo = (TupleTypeInfo<?>) objectTypeInformation;
+ Tuple t = (Tuple) attribute;
+ List<AttributeValue> attributeValues = new ArrayList<>();
+ for (int i = 0; i < tupleTypeInfo.getTotalFields(); i++) {
+ attributeValues.add(convertValue(t.getField(i), tupleTypeInfo.getTypeAt(i)));
+ }
+ return AttributeValue.fromL(attributeValues);
+
+ } else if (objectTypeInformation instanceof CompositeType) {
+ return AttributeValue.fromM(
+ convertElementUsingTypeInfo(
+ attribute, (CompositeType<attrT>) objectTypeInformation));
+
+ } else {
+ throw new IllegalArgumentException(
+ String.format("Unsupported TypeInfo %s", objectTypeInformation));
+ }
+ }
+
+ private boolean isArrayTypeInfo(TypeInformation<?> typeInformation) {
+ return typeInformation instanceof BasicArrayTypeInfo
+ || typeInformation instanceof PrimitiveArrayTypeInfo
+ || typeInformation instanceof ObjectArrayTypeInfo;
+ }
+
+ private <attrT> AttributeValue convertArrayTypeObject(
+ Object attribute, TypeInformation<attrT> componentTypeInfo, boolean isPrimitive) {
+
+ if (componentTypeInfo.getTypeClass().isAssignableFrom(String.class)) {
+ return AttributeValue.fromSs(convertObjectArrayToStringList((Object[]) attribute));
+ } else if (componentTypeInfo instanceof NumericTypeInfo && !isPrimitive) {
+ return AttributeValue.fromNs(convertObjectArrayToStringList((Object[]) attribute));
+ } else if (componentTypeInfo instanceof NumericTypeInfo) {
+ if (attribute instanceof int[]) {
+ return AttributeValue.fromNs(
+ (Arrays.stream((int[]) attribute)
+ .boxed()
+ .map(Object::toString)
+ .collect(Collectors.toList())));
+ } else if (attribute instanceof double[]) {
+ return AttributeValue.fromNs(
+ (Arrays.stream((double[]) attribute)
+ .boxed()
+ .map(Object::toString)
+ .collect(Collectors.toList())));
+ } else if (attribute instanceof long[]) {
+ return AttributeValue.fromNs(
+ (Arrays.stream((long[]) attribute)
+ .boxed()
+ .map(Object::toString)
+ .collect(Collectors.toList())));
+ } else {
+ throw new IllegalArgumentException(
+ String.format(
+ "Unsupported primitive numeric typeInfo %s",
+ componentTypeInfo.getTypeClass()));
+ }
+ } else if (!isPrimitive) {
+ List<AttributeValue> attributeValueList = new ArrayList<>();
+ for (Object attributeItem : (Object[]) attribute) {
+ checkTypeCompatibility(attributeItem, componentTypeInfo);
+ attributeValueList.add(convertValue(((attrT) attributeItem), componentTypeInfo));
+ }
+ return AttributeValue.fromL(attributeValueList);
+ } else {
+ throw new IllegalArgumentException(
+ String.format(
+ "Unsupported primitive typeInfo %s", componentTypeInfo.getTypeClass()));
+ }
Review Comment:
This block is untested, please add tests
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbTypeInformedElementConverter.java:
##########
@@ -0,0 +1,245 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.NumericTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * A {@link ElementConverter} that converts an element to a {@link DynamoDbWriteRequest} using
+ * TypeInformation provided.
+ */
+@PublicEvolving
+public class DynamoDbTypeInformedElementConverter<inputT>
+ implements ElementConverter<inputT, DynamoDbWriteRequest> {
+ private final CompositeType<inputT> typeInfo;
+
+ /**
+ * Creates a {@link DynamoDbTypeInformedElementConverter} that converts an element to a {@link
+ * DynamoDbWriteRequest} using the provided {@link CompositeType}. Usage: {@code new
+ * DynamoDbTypeInformedElementConverter<>(TypeInformation.of(MyPojoClass.class))}
+ *
+ * @param typeInfo The {@link CompositeType} that provides the type information for the element.
+ */
+ public DynamoDbTypeInformedElementConverter(CompositeType<inputT> typeInfo) {
+ this.typeInfo = typeInfo;
+ }
+
+ @Override
+ public DynamoDbWriteRequest apply(inputT input, SinkWriter.Context context) {
+ try {
+ return DynamoDbWriteRequest.builder()
+ .setType(DynamoDbWriteRequestType.PUT)
+ .setItem(convertElementUsingTypeInfo(input, typeInfo))
+ .build();
+ } catch (IllegalArgumentException e) {
+ throw new FlinkRuntimeException("Couldn't convert Element to AttributeVal", e);
+ }
+ }
+
+ private <attT> Map<String, AttributeValue> convertElementUsingTypeInfo(
+ attT t, CompositeType<attT> typeInfo) {
+ Map<String, AttributeValue> map = new HashMap<>();
+ for (String fieldKey : typeInfo.getFieldNames()) {
+ TypeInformation<attT> fieldType = typeInfo.getTypeAt(fieldKey);
+ try {
+ Field field = t.getClass().getDeclaredField(fieldKey);
Review Comment:
As per the [coding guidelines](https://flink.apache.org/how-to-contribute/code-style-and-quality-java/#java-reflection) we should not use reflection. Can we instead use the `TypeExtractor` class as suggested in the docs?
Another note here is that loading the class information for each record is expensive. Suggest that we lazily cache the type info based on the class.
##########
flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/DynamoDbTypeInformedElementConverterTest.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.tuple.Tuple1;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.connector.dynamodb.util.ComplexPayload;
+import org.apache.flink.connector.dynamodb.util.Order;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.flink.connector.dynamodb.sink.DynamoDbWriteRequestType.PUT;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link DynamoDbTypeInformedElementConverter}. */
+public class DynamoDbTypeInformedElementConverterTest {
+
+ @Test
+ public void testSimpleTypeConversion() {
+ DynamoDbTypeInformedElementConverter<Order> elementConverter =
+ new DynamoDbTypeInformedElementConverter<>(
+ (CompositeType<Order>) TypeInformation.of(Order.class));
+
+ Order order = new Order("orderId", 1, 2.0);
+
+ DynamoDbWriteRequest actual = elementConverter.apply(order, null);
+
+ assertThat(actual.getType()).isEqualTo(PUT);
+ assertThat(actual.getItem()).containsOnlyKeys("orderId", "quantity", "total");
+ assertThat(actual.getItem().get("orderId").s()).isEqualTo("orderId");
+ assertThat(actual.getItem().get("quantity").n()).isEqualTo("1");
+ assertThat(actual.getItem().get("total").n()).isEqualTo("2.0");
+ }
+
+ @Test
+ public void testComplexTypeConversion() {
+ TypeInformation<ComplexPayload> typeInformation = TypeInformation.of(ComplexPayload.class);
+ DynamoDbTypeInformedElementConverter<ComplexPayload> elementConverter =
+ new DynamoDbTypeInformedElementConverter<>(
+ (CompositeType<ComplexPayload>) typeInformation);
+
+ ComplexPayload payload =
+ new ComplexPayload(
+ "stringFieldVal",
+ new String[] {"stringArrayFieldVal1", "stringArrayFieldVal2"},
+ new int[] {10, 20},
+ new ComplexPayload.InnerPayload(true, new byte[] {1, 0, 10}));
+
+ DynamoDbWriteRequest actual = elementConverter.apply(payload, null);
+
+ assertThat(actual.getType()).isEqualTo(PUT);
+ assertThat(actual.getItem())
+ .containsOnlyKeys(
+ "stringField", "stringArrayField", "intArrayField", "innerPayload");
+ assertThat(actual.getItem().get("stringArrayField").ss())
+ .containsExactly("stringArrayFieldVal1", "stringArrayFieldVal2");
+ assertThat(actual.getItem().get("intArrayField").ns()).containsExactly("10", "20");
+
+ assertThat(actual.getItem().get("innerPayload").m()).isNotNull();
+ Map<String, AttributeValue> innerPayload = actual.getItem().get("innerPayload").m();
+
+ assertThat(innerPayload).containsOnlyKeys("primitiveBooleanField", "byteArrayField");
+ assertThat(innerPayload.get("primitiveBooleanField").bool()).isTrue();
+ assertThat(innerPayload.get("byteArrayField").b())
+ .isEqualTo(SdkBytes.fromByteArray(new byte[] {1, 0, 10}));
+ }
+
+ @Test
+ public void testCollectionTypeConversion() {
+ DynamoDbTypeInformedElementConverter<Tuple> elementConverter =
+ new DynamoDbTypeInformedElementConverter<>(
+ new TupleTypeInfo<>(
+ BasicTypeInfo.STRING_TYPE_INFO,
+ BasicArrayTypeInfo.STRING_ARRAY_TYPE_INFO));
+ List<String> stringArray = new ArrayList<>();
+ stringArray.add("stringArrayFieldVal1");
+ stringArray.add("stringArrayFieldVal2");
+
+ DynamoDbWriteRequest actual =
+ elementConverter.apply(Tuple2.of("stringVal", stringArray), null);
+
+ assertThat(actual.getType()).isEqualTo(PUT);
+ assertThat(actual.getItem()).containsOnlyKeys("f0", "f1");
+ assertThat(actual.getItem().get("f0").s()).isEqualTo("stringVal");
+ assertThat(actual.getItem().get("f1").ss())
+ .containsExactly("stringArrayFieldVal1", "stringArrayFieldVal2");
+ }
+
+ @Test
+ public void testUnsupportedTypeIsWrappedByFlinkException() {
+ DynamoDbTypeInformedElementConverter<Tuple> elementConverter =
+ new DynamoDbTypeInformedElementConverter<>(
+ new TupleTypeInfo<>(
+ BasicTypeInfo.DATE_TYPE_INFO,
+ BasicArrayTypeInfo.STRING_ARRAY_TYPE_INFO));
+
+ Date dateVal = Date.from(Instant.now());
+
+ Throwable throwable =
+ Assertions.assertThrows(
+ FlinkRuntimeException.class,
+ () -> {
+ elementConverter.apply(Tuple2.of(dateVal, "stringVal"), null);
+ });
+
+ assertThat(throwable).hasCauseInstanceOf(IllegalArgumentException.class);
+ assertThat(throwable.getCause())
+ .hasMessageContaining("Unsupported BasicTypeInfo " + BasicTypeInfo.DATE_TYPE_INFO);
Review Comment:
Can you use the AssertJ lib here instead?
```
Assertions.assertThatExceptionOfType(FlinkRuntimeException.class)
.isThrownBy(() -> elementConverter.apply(Tuple1.of("nan"), null))
.withMessageContaining("Incompatible type when converting nan, using Integer type information.")
.withRootCauseInstanceOf(IllegalArgumentException.class);
```
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbTypeInformedElementConverter.java:
##########
@@ -0,0 +1,245 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.NumericTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * A {@link ElementConverter} that converts an element to a {@link DynamoDbWriteRequest} using
+ * TypeInformation provided.
+ */
+@PublicEvolving
+public class DynamoDbTypeInformedElementConverter<inputT>
+ implements ElementConverter<inputT, DynamoDbWriteRequest> {
+ private final CompositeType<inputT> typeInfo;
+
+ /**
+ * Creates a {@link DynamoDbTypeInformedElementConverter} that converts an element to a {@link
+ * DynamoDbWriteRequest} using the provided {@link CompositeType}. Usage: {@code new
+ * DynamoDbTypeInformedElementConverter<>(TypeInformation.of(MyPojoClass.class))}
+ *
+ * @param typeInfo The {@link CompositeType} that provides the type information for the element.
+ */
+ public DynamoDbTypeInformedElementConverter(CompositeType<inputT> typeInfo) {
+ this.typeInfo = typeInfo;
+ }
+
+ @Override
+ public DynamoDbWriteRequest apply(inputT input, SinkWriter.Context context) {
+ try {
+ return DynamoDbWriteRequest.builder()
+ .setType(DynamoDbWriteRequestType.PUT)
+ .setItem(convertElementUsingTypeInfo(input, typeInfo))
+ .build();
+ } catch (IllegalArgumentException e) {
+ throw new FlinkRuntimeException("Couldn't convert Element to AttributeVal", e);
+ }
+ }
+
+ private <attT> Map<String, AttributeValue> convertElementUsingTypeInfo(
+ attT t, CompositeType<attT> typeInfo) {
+ Map<String, AttributeValue> map = new HashMap<>();
+ for (String fieldKey : typeInfo.getFieldNames()) {
+ TypeInformation<attT> fieldType = typeInfo.getTypeAt(fieldKey);
+ try {
+ Field field = t.getClass().getDeclaredField(fieldKey);
+ field.setAccessible(true);
+ Object fieldVal = field.get(t);
+ checkTypeCompatibility(fieldVal, fieldType);
+ attT fieldValCaster = (attT) fieldVal;
+ map.put(fieldKey, convertValue(fieldValCaster, fieldType));
+ } catch (NoSuchFieldException | IllegalAccessException e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Failed to extract field %s declared in TypeInfo "
+ + "from Object %s",
+ fieldKey, t),
+ e);
+ }
+ }
+
+ return map;
+ }
+
+ private <attrT> AttributeValue convertValue(
+ attrT attribute, TypeInformation<attrT> objectTypeInformation) {
+ if (attribute == null) {
+ return AttributeValue.builder().nul(true).build();
+ } else if (objectTypeInformation.isBasicType()) {
+ return convertBasicTypeObject(attribute, (BasicTypeInfo<?>) objectTypeInformation);
+ } else if (objectTypeInformation.equals(
+ PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO)) {
+ // special handling for byte array as it has special support in DynamoDB
+ if (attribute instanceof byte[]) {
+ return AttributeValue.fromB(SdkBytes.fromByteArray((byte[]) attribute));
+ } else if (attribute instanceof SdkBytes) {
+ return AttributeValue.fromB((SdkBytes) attribute);
+ } else {
+ throw new IllegalArgumentException(
+ String.format("Failed to extract byte array type from %s", attribute));
+ }
+
+ } else if (isArrayTypeInfo(objectTypeInformation)) {
+ TypeInformation<?> componentTypeInfo =
+ objectTypeInformation instanceof BasicArrayTypeInfo
+ ? ((BasicArrayTypeInfo<?, ?>) objectTypeInformation).getComponentInfo()
+ : objectTypeInformation instanceof PrimitiveArrayTypeInfo
+ ? ((PrimitiveArrayTypeInfo<?>) objectTypeInformation)
+ .getComponentType()
+ : ((ObjectArrayTypeInfo<?, ?>) objectTypeInformation)
+ .getComponentInfo();
+ if (attribute instanceof Collection) {
+ return convertArrayTypeObject(
+ ((Collection<?>) attribute).toArray(), componentTypeInfo, false);
+ }
+
+ return convertArrayTypeObject(
+ attribute,
+ componentTypeInfo,
+ objectTypeInformation instanceof PrimitiveArrayTypeInfo);
+
+ } else if (objectTypeInformation instanceof TupleTypeInfo) {
+ TupleTypeInfo<?> tupleTypeInfo = (TupleTypeInfo<?>) objectTypeInformation;
+ Tuple t = (Tuple) attribute;
+ List<AttributeValue> attributeValues = new ArrayList<>();
+ for (int i = 0; i < tupleTypeInfo.getTotalFields(); i++) {
+ attributeValues.add(convertValue(t.getField(i), tupleTypeInfo.getTypeAt(i)));
+ }
+ return AttributeValue.fromL(attributeValues);
+
+ } else if (objectTypeInformation instanceof CompositeType) {
+ return AttributeValue.fromM(
+ convertElementUsingTypeInfo(
+ attribute, (CompositeType<attrT>) objectTypeInformation));
+
+ } else {
+ throw new IllegalArgumentException(
+ String.format("Unsupported TypeInfo %s", objectTypeInformation));
+ }
+ }
+
+ private boolean isArrayTypeInfo(TypeInformation<?> typeInformation) {
+ return typeInformation instanceof BasicArrayTypeInfo
+ || typeInformation instanceof PrimitiveArrayTypeInfo
+ || typeInformation instanceof ObjectArrayTypeInfo;
+ }
+
+ private <attrT> AttributeValue convertArrayTypeObject(
+ Object attribute, TypeInformation<attrT> componentTypeInfo, boolean isPrimitive) {
+
+ if (componentTypeInfo.getTypeClass().isAssignableFrom(String.class)) {
+ return AttributeValue.fromSs(convertObjectArrayToStringList((Object[]) attribute));
+ } else if (componentTypeInfo instanceof NumericTypeInfo && !isPrimitive) {
+ return AttributeValue.fromNs(convertObjectArrayToStringList((Object[]) attribute));
+ } else if (componentTypeInfo instanceof NumericTypeInfo) {
+ if (attribute instanceof int[]) {
+ return AttributeValue.fromNs(
+ (Arrays.stream((int[]) attribute)
Review Comment:
As a big fan of the stream API, it pains me to say that we [should not be using it for performance critical code](https://flink.apache.org/how-to-contribute/code-style-and-quality-java/#java-streams). Since this would be called for each record I would consider this performance critical/data-intensive. Please convert to standard loops, can extract to new methods for readability.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35022] Add TypeInformed DDB Element Converter [flink-connector-aws]
Posted by "dannycranmer (via GitHub)" <gi...@apache.org>.
dannycranmer commented on PR #136:
URL: https://github.com/apache/flink-connector-aws/pull/136#issuecomment-2056352344
Please change commit message to include the component: `[FLINK-35022][Connectors/DynamoDB]`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35022][Connector/DynamoDB] Add TypeInformed DDB Element Converter as default element converter [flink-connector-aws]
Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 commented on code in PR #136:
URL: https://github.com/apache/flink-connector-aws/pull/136#discussion_r1634758095
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbTypeInformedElementConverter.java:
##########
@@ -0,0 +1,380 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.NumericTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.PojoTypeInfo;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.dynamodb.table.converter.ArrayAttributeConverterProvider;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverter;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeConverterProvider;
+import software.amazon.awssdk.enhanced.dynamodb.AttributeValueType;
+import software.amazon.awssdk.enhanced.dynamodb.EnhancedType;
+import software.amazon.awssdk.enhanced.dynamodb.TableSchema;
+import software.amazon.awssdk.enhanced.dynamodb.internal.mapper.BeanAttributeGetter;
+import software.amazon.awssdk.enhanced.dynamodb.mapper.StaticTableSchema;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+import java.beans.BeanInfo;
+import java.beans.IntrospectionException;
+import java.beans.Introspector;
+import java.beans.PropertyDescriptor;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Function;
+
+/**
+ * A {@link ElementConverter} that converts an element to a {@link DynamoDbWriteRequest} using
+ * TypeInformation provided.
+ */
+@PublicEvolving
+public class DynamoDbTypeInformedElementConverter<T>
+ implements ElementConverter<T, DynamoDbWriteRequest> {
+
+ private final CompositeType<T> typeInfo;
+ private final boolean ignoreNulls;
+ private final TableSchema<T> tableSchema;
+
+ /**
+ * Creates a {@link DynamoDbTypeInformedElementConverter} that converts an element to a {@link
+ * DynamoDbWriteRequest} using the provided {@link CompositeType}. Usage: {@code new
+ * DynamoDbTypeInformedElementConverter<>(TypeInformation.of(MyPojoClass.class))}
+ *
+ * @param typeInfo The {@link CompositeType} that provides the type information for the element.
+ */
+ public DynamoDbTypeInformedElementConverter(CompositeType<T> typeInfo) {
+ this(typeInfo, true);
+ }
+
+ public DynamoDbTypeInformedElementConverter(CompositeType<T> typeInfo, boolean ignoreNulls) {
+
+ try {
+ this.typeInfo = typeInfo;
+ this.ignoreNulls = ignoreNulls;
+ this.tableSchema = createTableSchema(typeInfo);
+ } catch (IntrospectionException | IllegalStateException | IllegalArgumentException e) {
+ throw new FlinkRuntimeException("Failed to extract DynamoDb table schema", e);
+ }
+ }
+
+ @Override
+ public DynamoDbWriteRequest apply(T input, SinkWriter.Context context) {
+ Preconditions.checkNotNull(tableSchema, "TableSchema is not initialized");
+ try {
+ return DynamoDbWriteRequest.builder()
+ .setType(DynamoDbWriteRequestType.PUT)
+ .setItem(tableSchema.itemToMap(input, ignoreNulls))
+ .build();
+ } catch (ClassCastException | IllegalArgumentException e) {
+ throw new FlinkRuntimeException(
+ String.format(
+ "Failed to convert %s to DynamoDbWriteRequest using %s",
+ input, typeInfo),
+ e);
+ }
+ }
+
+ private <AttributeT> TableSchema<AttributeT> createTableSchema(
+ CompositeType<AttributeT> typeInfo) throws IntrospectionException {
+ if (typeInfo instanceof RowTypeInfo) {
+ return (TableSchema<AttributeT>) createTableSchemaFromRowType((RowTypeInfo) typeInfo);
+ } else if (typeInfo instanceof PojoTypeInfo<?>) {
+ return createTableSchemaFromPojo((PojoTypeInfo<AttributeT>) typeInfo);
+ } else if (typeInfo instanceof TupleTypeInfo<?>) {
+ return createTableSchemaFromTuple((TupleTypeInfo<?>) typeInfo);
+ } else {
+ throw new IllegalArgumentException(String.format("Unsupported TypeInfo %s", typeInfo));
+ }
+ }
+
+ private TableSchema<Row> createTableSchemaFromRowType(RowTypeInfo typeInfo) {
+ StaticTableSchema.Builder<Row> tableSchemaBuilder =
+ StaticTableSchema.builder(typeInfo.getTypeClass());
+
+ String[] fieldNames = typeInfo.getFieldNames();
+ for (int i = 0; i < typeInfo.getArity(); i++) {
+ TypeInformation<?> fieldType = typeInfo.getTypeAt(i);
+ int finalI = i;
+ addAttribute(
+ tableSchemaBuilder,
+ fieldNames[finalI],
+ (AttributeT) -> AttributeT.getField(finalI),
+ (TypeInformation<? super Object>) fieldType);
+ }
+
+ return tableSchemaBuilder.build();
+ }
+
+ private <AttributeT> TableSchema<AttributeT> createTableSchemaFromTuple(
+ TupleTypeInfo<?> typeInfo) {
+ TypeInformation<?>[] fieldTypes = typeInfo.getFieldTypes();
+ String[] fieldNames = typeInfo.getFieldNames();
+
+ StaticTableSchema.Builder<?> tableSchemaBuilder =
+ StaticTableSchema.builder(typeInfo.getTypeClass());
+ for (int i = 0; i < fieldNames.length; i++) {
+ int finalI = i;
+ addAttribute(
+ tableSchemaBuilder,
+ fieldNames[finalI],
+ (tuple) -> ((Tuple) tuple).getField(finalI),
+ (TypeInformation<?>) fieldTypes[i]);
+ }
+
+ return (TableSchema<AttributeT>) tableSchemaBuilder.build();
+ }
+
+ private <AttributeT> TableSchema<AttributeT> createTableSchemaFromPojo(
+ PojoTypeInfo<AttributeT> typeInfo) throws IntrospectionException {
+ StaticTableSchema.Builder<AttributeT> tableSchemaBuilder =
+ StaticTableSchema.builder(typeInfo.getTypeClass());
+ BeanInfo beanInfo = Introspector.getBeanInfo(typeInfo.getTypeClass());
+ PropertyDescriptor[] propertyDescriptors = beanInfo.getPropertyDescriptors();
+ for (PropertyDescriptor propertyDescriptor : propertyDescriptors) {
+ Set<String> fieldNames = new HashSet<>(Arrays.asList(typeInfo.getFieldNames()));
+ if (!fieldNames.contains(propertyDescriptor.getName())) {
+ // Skip properties that are not part of the PojoTypeInfo
+ continue;
+ }
+
+ TypeInformation<?> fieldInfo = typeInfo.getTypeAt(propertyDescriptor.getName());
+ addAttribute(
+ tableSchemaBuilder,
+ propertyDescriptor.getName(),
+ BeanAttributeGetter.create(
+ typeInfo.getTypeClass(), propertyDescriptor.getReadMethod()),
+ fieldInfo);
+ }
+
+ return tableSchemaBuilder.build();
+ }
+
+ private <T, AttributeT> void addAttribute(
+ StaticTableSchema.Builder<T> builder,
+ String fieldName,
+ Function<T, AttributeT> getter,
+ TypeInformation<AttributeT> typeInfo) {
+ builder.addAttribute(
+ typeInfo.getTypeClass(),
+ a ->
+ a.name(fieldName)
+ .getter(getter)
+ .setter((e, o) -> {})
+ .attributeConverter(getAttributeConverter(typeInfo)));
+ }
+
+ private <AttributeT> AttributeConverter<AttributeT> getAttributeConverter(
+ TypeInformation<AttributeT> typeInfo) {
+ if (typeInfo instanceof BasicTypeInfo) {
+ return AttributeConverterProvider.defaultProvider()
+ .converterFor(EnhancedType.of(typeInfo.getTypeClass()));
+ } else if (typeInfo.equals(PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO)) {
+ return getAttributeConverter(
+ AttributeValueType.B,
+ bytes ->
+ bytes instanceof SdkBytes
+ ? AttributeValue.fromB((SdkBytes) bytes)
+ : AttributeValue.fromB(SdkBytes.fromByteArray((byte[]) bytes)));
+ } else if (typeInfo instanceof BasicArrayTypeInfo) {
+ BasicArrayTypeInfo<AttributeT, ?> basicArrayTypeInfo =
+ (BasicArrayTypeInfo<AttributeT, ?>) typeInfo;
+ if (basicArrayTypeInfo.getComponentInfo().equals(BasicTypeInfo.STRING_TYPE_INFO)) {
+ return getAttributeConverter(
+ AttributeValueType.SS,
+ array -> AttributeValue.fromSs(Arrays.asList((String[]) array)));
+ } else if (basicArrayTypeInfo.getComponentInfo() instanceof NumericTypeInfo) {
+ return getAttributeConverter(
+ AttributeValueType.NS,
+ array ->
+ AttributeValue.fromNs(
+ convertObjectArrayToStringList((Object[]) array)));
+ }
+
+ return new ArrayAttributeConverterProvider()
+ .converterFor(EnhancedType.of(typeInfo.getTypeClass()));
+ } else if (typeInfo instanceof ObjectArrayTypeInfo) {
+ return getObjectArrayTypeConverter((ObjectArrayTypeInfo<AttributeT, ?>) typeInfo);
+ } else if (typeInfo instanceof PrimitiveArrayTypeInfo) {
+ PrimitiveArrayTypeInfo<AttributeT> primitiveArrayTypeInfo =
+ (PrimitiveArrayTypeInfo<AttributeT>) typeInfo;
+ if (primitiveArrayTypeInfo.getComponentType() instanceof NumericTypeInfo) {
+ return getAttributeConverter(
+ AttributeValueType.NS,
+ array -> AttributeValue.fromNs(convertPrimitiveArrayToStringList(array)));
+ } else {
+ throw new IllegalArgumentException(
+ String.format(
+ "Unsupported primitive array typeInfo %s",
+ primitiveArrayTypeInfo.getComponentType()));
+ }
Review Comment:
Have we considered using the `.attributeConverterProviders()` method in the table schema builder and reusing the constructs we use to DDB Table API?
```
TableSchema.builder(...)
.attributeConverterProviders(new ArrayAttributeConverterProvider(), ... (add any more here) ..., AttributeConverterProvider.defaultProvider());
```
We have an example here:
https://github.com/apache/flink-connector-aws/blob/main/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverter.java#L56-L72
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35022][Connector/DynamoDB] Add TypeInformed DDB Element Converter [flink-connector-aws]
Posted by "dannycranmer (via GitHub)" <gi...@apache.org>.
dannycranmer commented on PR #136:
URL: https://github.com/apache/flink-connector-aws/pull/136#issuecomment-2077164916
@vahmed-hamdy can you also update the docs please?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org