public RowData getRowDataFromRecordData(RecordData recordData, boolean primaryKeyOnly) {
        if (primaryKeyOnly) {
            GenericRowData genericRowData = new GenericRowData(primaryKeyColumnIndexes.size() + 1);
            genericRowData.setField(0, StringData.fromString(tableId.toString()));
            for (int i = 0; i < primaryKeyColumnIndexes.size(); i++) {
                genericRowData.setField(
                        i + 1,
                        fieldGetters
                                .get(primaryKeyColumnIndexes.get(i))
                                .getFieldOrNull(recordData));
            }
            return genericRowData;
        } else {
            GenericRowData genericRowData = new GenericRowData(recordData.getArity());
            for (int i = 0; i < recordData.getArity(); i++) {
                genericRowData.setField(i, fieldGetters.get(i).getFieldOrNull(recordData));
            }
            return genericRowData;
        }
    }
@Test
    public void testGetRowDataFromRecordData() {
        Schema schema =
                Schema.newBuilder()
                        .physicalColumn(
                                "col1",
                                org.apache.flink.cdc.common.types.DataTypes.STRING().notNull())
                        .physicalColumn(
                                "boolean", org.apache.flink.cdc.common.types.DataTypes.BOOLEAN())
                        .physicalColumn(
                                "binary", org.apache.flink.cdc.common.types.DataTypes.BINARY(3))
                        .physicalColumn(
                                "varbinary",
                                org.apache.flink.cdc.common.types.DataTypes.VARBINARY(10))
                        .physicalColumn(
                                "bytes", org.apache.flink.cdc.common.types.DataTypes.BYTES())
                        .physicalColumn(
                                "tinyint", org.apache.flink.cdc.common.types.DataTypes.TINYINT())
                        .physicalColumn(
                                "smallint", org.apache.flink.cdc.common.types.DataTypes.SMALLINT())
                        .physicalColumn("int", org.apache.flink.cdc.common.types.DataTypes.INT())
                        .physicalColumn(
                                "big_int", org.apache.flink.cdc.common.types.DataTypes.BIGINT())
                        .physicalColumn(
                                "float", org.apache.flink.cdc.common.types.DataTypes.FLOAT())
                        .physicalColumn(
                                "double", org.apache.flink.cdc.common.types.DataTypes.DOUBLE())
                        .physicalColumn(
                                "decimal",
                                org.apache.flink.cdc.common.types.DataTypes.DECIMAL(6, 3))
                        .physicalColumn("char", org.apache.flink.cdc.common.types.DataTypes.CHAR(5))
                        .physicalColumn(
                                "varchar", org.apache.flink.cdc.common.types.DataTypes.VARCHAR(10))
                        .physicalColumn(
                                "string", org.apache.flink.cdc.common.types.DataTypes.STRING())
                        .physicalColumn("date", org.apache.flink.cdc.common.types.DataTypes.DATE())
                        .physicalColumn("time", org.apache.flink.cdc.common.types.DataTypes.TIME())
                        .physicalColumn(
                                "time_with_precision",
                                org.apache.flink.cdc.common.types.DataTypes.TIME(6))
                        .physicalColumn(
                                "timestamp",
                                org.apache.flink.cdc.common.types.DataTypes.TIMESTAMP())
                        .physicalColumn(
                                "timestamp_with_precision",
                                org.apache.flink.cdc.common.types.DataTypes.TIMESTAMP(3))
                        .physicalColumn(
                                "timestamp_ltz",
                                org.apache.flink.cdc.common.types.DataTypes.TIMESTAMP_LTZ())
                        .physicalColumn(
                                "timestamp_ltz_with_precision",
                                org.apache.flink.cdc.common.types.DataTypes.TIMESTAMP_LTZ(3))
                        .physicalColumn(
                                "null_string", org.apache.flink.cdc.common.types.DataTypes.STRING())
                        .primaryKey("col1")
                        .build();
        TableSchemaInfo tableSchemaInfo =
                new TableSchemaInfo(
                        TableId.parse("testDatabase.testTable"), schema, null, ZoneId.of("UTC+8"));
        Object[] testData =
                new Object[] {
                    BinaryStringData.fromString("pk"),
                    true,
                    new byte[] {1, 2},
                    new byte[] {3, 4},
                    new byte[] {5, 6, 7},
                    (byte) 1,
                    (short) 2,
                    3,
                    4L,
                    5.1f,
                    6.2,
                    DecimalData.fromBigDecimal(new BigDecimal("7.123"), 6, 3),
                    BinaryStringData.fromString("test1"),
                    BinaryStringData.fromString("test2"),
                    BinaryStringData.fromString("test3"),
                    100,
                    200,
                    300,
                    TimestampData.fromTimestamp(
                            java.sql.Timestamp.valueOf("2023-01-01 00:00:00.000")),
                    TimestampData.fromTimestamp(java.sql.Timestamp.valueOf("2023-01-01 00:00:00")),
                    LocalZonedTimestampData.fromInstant(Instant.parse("2023-01-01T00:00:00.000Z")),
                    LocalZonedTimestampData.fromInstant(Instant.parse("2023-01-01T00:00:00.000Z")),
                    null
                };
        BinaryRecordData recordData =
                new BinaryRecordDataGenerator(schema.getColumnDataTypes().toArray(new DataType[0]))
                        .generate(testData);

        Assertions.assertEquals(
                GenericRowData.ofKind(
                        RowKind.INSERT,
                        org.apache.flink.table.data.binary.BinaryStringData.fromString("pk"),
                        true,
                        new byte[] {1, 2},
                        new byte[] {3, 4},
                        new byte[] {5, 6, 7},
                        (byte) 1,
                        (short) 2,
                        3,
                        4L,
                        5.1f,
                        6.2,
                        org.apache.flink.table.data.DecimalData.fromBigDecimal(
                                new BigDecimal("7.123"), 6, 3),
                        org.apache.flink.table.data.binary.BinaryStringData.fromString("test1"),
                        org.apache.flink.table.data.binary.BinaryStringData.fromString("test2"),
                        org.apache.flink.table.data.binary.BinaryStringData.fromString("test3"),
                        100,
                        200,
                        300,
                        org.apache.flink.table.data.TimestampData.fromTimestamp(
                                Timestamp.valueOf("2023-01-01 00:00:00.000")),
                        org.apache.flink.table.data.TimestampData.fromTimestamp(
                                Timestamp.valueOf("2023-01-01 00:00:00")),
                        // plus 8 hours.
                        org.apache.flink.table.data.TimestampData.fromInstant(
                                Instant.parse("2023-01-01T08:00:00.000Z")),
                        org.apache.flink.table.data.TimestampData.fromInstant(
                                Instant.parse("2023-01-01T08:00:00.000Z")),
                        null),
                tableSchemaInfo.getRowDataFromRecordData(recordData, false));
    }