[FLINK-38889][pipeline][kafka] Support serializing complex types(MAP, ARRAY, ROW) to JSON (Debezium / Canal) (#4221)

Co-authored-by: guoxuanlin <guoxuanlin@tencent.com>
Co-authored-by: yuxiqian <34335406+yuxiqian@users.noreply.github.com>

Author: 3 people

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/TableSchemaInfo.java

@@ -21,22 +21,15 @@
 import org.apache.flink.cdc.common.data.RecordData;
 import org.apache.flink.cdc.common.event.TableId;
 import org.apache.flink.cdc.common.schema.Schema;
-import org.apache.flink.cdc.common.types.DataType;
-import org.apache.flink.cdc.common.types.DataTypeChecks;
-import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.cdc.connectors.kafka.json.utils.RecordDataConverter;
 import org.apache.flink.table.data.GenericRowData;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.data.StringData;
-import org.apache.flink.table.data.TimestampData;
-import org.apache.flink.table.data.binary.BinaryStringData;
 
 import java.time.ZoneId;
 import java.util.ArrayList;
 import java.util.List;
 
-import static org.apache.flink.cdc.common.types.DataTypeChecks.getPrecision;
-import static org.apache.flink.cdc.common.types.DataTypeChecks.getScale;
-
 /** maintain the {@link SerializationSchema} of a specific {@link TableId}. */
 public class TableSchemaInfo {
 
@@ -96,96 +89,7 @@ public RowData getRowDataFromRecordData(RecordData recordData, boolean primaryKe
     }
 
     private static List<RecordData.FieldGetter> createFieldGetters(Schema schema, ZoneId zoneId) {
-        List<RecordData.FieldGetter> fieldGetters = new ArrayList<>(schema.getColumns().size());
-        for (int i = 0; i < schema.getColumns().size(); i++) {
-            fieldGetters.add(createFieldGetter(schema.getColumns().get(i).getType(), i, zoneId));
-        }
-        return fieldGetters;
-    }
-
-    private static RecordData.FieldGetter createFieldGetter(
-            DataType fieldType, int fieldPos, ZoneId zoneId) {
-        final RecordData.FieldGetter fieldGetter;
-        // ordered by type root definition
-        switch (fieldType.getTypeRoot()) {
-            case CHAR:
-            case VARCHAR:
-                fieldGetter =
-                        record ->
-                                BinaryStringData.fromString(record.getString(fieldPos).toString());
-                break;
-            case BOOLEAN:
-                fieldGetter = record -> record.getBoolean(fieldPos);
-                break;
-            case BINARY:
-            case VARBINARY:
-                fieldGetter = record -> record.getBinary(fieldPos);
-                break;
-            case DECIMAL:
-                final int decimalPrecision = getPrecision(fieldType);
-                final int decimalScale = getScale(fieldType);
-                fieldGetter =
-                        record ->
-                                DecimalData.fromBigDecimal(
-                                        record.getDecimal(fieldPos, decimalPrecision, decimalScale)
-                                                .toBigDecimal(),
-                                        decimalPrecision,
-                                        decimalScale);
-                break;
-            case TINYINT:
-                fieldGetter = record -> record.getByte(fieldPos);
-                break;
-            case SMALLINT:
-                fieldGetter = record -> record.getShort(fieldPos);
-                break;
-            case INTEGER:
-                fieldGetter = record -> record.getInt(fieldPos);
-                break;
-            case DATE:
-                fieldGetter = record -> (int) record.getDate(fieldPos).toEpochDay();
-                break;
-            case TIME_WITHOUT_TIME_ZONE:
-                fieldGetter = record -> (int) record.getTime(fieldPos).toMillisOfDay();
-                break;
-            case BIGINT:
-                fieldGetter = record -> record.getLong(fieldPos);
-                break;
-            case FLOAT:
-                fieldGetter = record -> record.getFloat(fieldPos);
-                break;
-            case DOUBLE:
-                fieldGetter = record -> record.getDouble(fieldPos);
-                break;
-            case TIMESTAMP_WITHOUT_TIME_ZONE:
-                fieldGetter =
-                        record ->
-                                TimestampData.fromTimestamp(
-                                        record.getTimestamp(fieldPos, getPrecision(fieldType))
-                                                .toTimestamp());
-                break;
-            case TIMESTAMP_WITH_TIME_ZONE:
-            case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
-                fieldGetter =
-                        record ->
-                                TimestampData.fromInstant(
-                                        record.getLocalZonedTimestampData(
-                                                        fieldPos,
-                                                        DataTypeChecks.getPrecision(fieldType))
-                                                .toInstant());
-                break;
-            default:
-                throw new IllegalArgumentException(
-                        "don't support type of " + fieldType.getTypeRoot());
-        }
-        if (!fieldType.isNullable()) {
-            return fieldGetter;
-        }
-        return row -> {
-            if (row.isNullAt(fieldPos)) {
-                return null;
-            }
-            return fieldGetter.getFieldOrNull(row);
-        };
+        return RecordDataConverter.createFieldGetters(schema, zoneId);
     }
 
     public Schema getSchema() {