loserwang1024 commented on code in PR #4108:
URL: https://github.com/apache/flink-cdc/pull/4108#discussion_r2296625362
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresFullTypesITCase.java:
##########
@@ -263,7 +264,7 @@ public void testTimeTypesWithTemporalModeAdaptive() throws
Exception {
.tableList("inventory.time_types")
.startupOptions(StartupOptions.initial())
.debeziumProperties(debeziumProps)
- .serverTimeZone("UTC");
+ .serverTimeZone("UTC+8");
Review Comment:
What this modification want to test?
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java:
##########
@@ -472,6 +482,57 @@ protected Object convertToMap(Object dbzObj, Schema
schema) throws Exception {
return new GenericMapData(convertedMap);
}
+ protected Object convertToArray(Object dbzObj, Schema schema) throws
Exception {
+ if (dbzObj == null) {
+ return null;
+ }
+
+ Schema elementSchema = schema.valueSchema();
+ DataType elementType = schemaDataTypeInference.infer(null,
elementSchema);
+ DeserializationRuntimeConverter elementConverter =
createConverter(elementType);
+
+ if (dbzObj instanceof java.util.List) {
+ java.util.List<?> list = (java.util.List<?>) dbzObj;
+ Object[] array = new Object[list.size()];
+
+ for (int i = 0; i < list.size(); i++) {
+ Object element = list.get(i);
+ if (element != null) {
+ if (elementSchema.type() == Schema.Type.ARRAY) {
+ array[i] = convertToArray(element, elementSchema);
Review Comment:
Add a test to test this nested array?
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java:
##########
@@ -472,6 +482,57 @@ protected Object convertToMap(Object dbzObj, Schema
schema) throws Exception {
return new GenericMapData(convertedMap);
}
+ protected Object convertToArray(Object dbzObj, Schema schema) throws
Exception {
+ if (dbzObj == null) {
+ return null;
+ }
+
+ Schema elementSchema = schema.valueSchema();
+ DataType elementType = schemaDataTypeInference.infer(null,
elementSchema);
+ DeserializationRuntimeConverter elementConverter =
createConverter(elementType);
+
+ if (dbzObj instanceof java.util.List) {
+ java.util.List<?> list = (java.util.List<?>) dbzObj;
+ Object[] array = new Object[list.size()];
+
+ for (int i = 0; i < list.size(); i++) {
+ Object element = list.get(i);
+ if (element != null) {
+ if (elementSchema.type() == Schema.Type.ARRAY) {
+ array[i] = convertToArray(element, elementSchema);
+ } else {
+ array[i] = elementConverter.convert(element,
elementSchema);
+ }
+ }
+ }
+
+ return new GenericArrayData(array);
+ } else if (dbzObj.getClass().isArray()) {
+ Object[] inputArray = (Object[]) dbzObj;
+ Object[] convertedArray = new Object[inputArray.length];
+
+ for (int i = 0; i < inputArray.length; i++) {
+ if (inputArray[i] != null) {
+ if (elementSchema.type() == Schema.Type.ARRAY) {
+ convertedArray[i] = convertToArray(inputArray[i],
elementSchema);
+ } else {
+ convertedArray[i] =
elementConverter.convert(inputArray[i], elementSchema);
+ }
+ }
+ }
+
+ return new GenericArrayData(convertedArray);
+ } else {
+ Object[] array = new Object[1];
Review Comment:
Add a test for length == 1?
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java:
##########
@@ -472,6 +482,57 @@ protected Object convertToMap(Object dbzObj, Schema
schema) throws Exception {
return new GenericMapData(convertedMap);
}
+ protected Object convertToArray(Object dbzObj, Schema schema) throws
Exception {
+ if (dbzObj == null) {
+ return null;
+ }
+
+ Schema elementSchema = schema.valueSchema();
+ DataType elementType = schemaDataTypeInference.infer(null,
elementSchema);
+ DeserializationRuntimeConverter elementConverter =
createConverter(elementType);
+
+ if (dbzObj instanceof java.util.List) {
+ java.util.List<?> list = (java.util.List<?>) dbzObj;
+ Object[] array = new Object[list.size()];
+
+ for (int i = 0; i < list.size(); i++) {
+ Object element = list.get(i);
+ if (element != null) {
Review Comment:
When this element will be null? What about thrown an exception if element is
null?
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresFullTypesITCase.java:
##########
@@ -703,6 +704,59 @@ public void testJsonTypes() throws Exception {
Assertions.assertThat(recordFields(snapshotRecord,
JSON_TYPES)).isEqualTo(expectedSnapshot);
}
+ @Test
+ public void testArrayTypes() throws Exception {
+ initializePostgresTable(POSTGIS_CONTAINER, "column_type_test");
+
+ PostgresSourceConfigFactory configFactory =
+ (PostgresSourceConfigFactory)
+ new PostgresSourceConfigFactory()
+ .hostname(POSTGIS_CONTAINER.getHost())
+
.port(POSTGIS_CONTAINER.getMappedPort(POSTGRESQL_PORT))
+ .username(TEST_USER)
+ .password(TEST_PASSWORD)
+
.databaseList(POSTGRES_CONTAINER.getDatabaseName())
+ .tableList("inventory.array_types")
+ .startupOptions(StartupOptions.initial())
+ .serverTimeZone("UTC");
+ configFactory.database(POSTGRES_CONTAINER.getDatabaseName());
+ configFactory.slotName(slotName);
+ configFactory.decodingPluginName("pgoutput");
+
+ FlinkSourceProvider sourceProvider =
+ (FlinkSourceProvider)
+ new
PostgresDataSource(configFactory).getEventSourceProvider();
+
+ CloseableIterator<Event> events =
+ env.fromSource(
+ sourceProvider.getSource(),
+ WatermarkStrategy.noWatermarks(),
+ PostgresDataSourceFactory.IDENTIFIER,
+ new EventTypeInfo())
+ .executeAndCollect();
+
+ List<Event> snapshotResults = fetchResultsAndCreateTableEvent(events,
1).f0;
+ RecordData snapshotRecord = ((DataChangeEvent)
snapshotResults.get(0)).after();
+
+ Object[] actualSnapshotObjects = recordFields(snapshotRecord,
ARRAY_TYPES);
+
+ Assertions.assertThat(actualSnapshotObjects[0]).isEqualTo(1); // id字段
Review Comment:
Use English?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
