@Override
    public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) {
        RowType physicalDataType =
                (RowType) physicalSchema.toPhysicalRowDataType().getLogicalType();
        MetadataConverter[] metadataConverters = getMetadataConverters();
        TypeInformation<RowData> typeInfo = scanContext.createTypeInformation(producedDataType);

        DebeziumDeserializationSchema<RowData> deserializer =
                RowDataDebeziumDeserializeSchema.newBuilder()
                        .setPhysicalRowType(physicalDataType)
                        .setMetadataConverters(metadataConverters)
                        .setResultTypeInfo(typeInfo)
                        .setUserDefinedConverterFactory(
                                PostgreSQLDeserializationConverterFactory.instance())
                        .setValueValidator(new PostgresValueValidator(schemaName, tableName))
                        .setChangelogMode(changelogMode)
                        .build();
        DebeziumSourceFunction<RowData> sourceFunction =
                PostgreSQLSource.<RowData>builder()
                        .hostname(hostname)
                        .port(port)
                        .database(database)
                        .schemaList(schemaName)
                        .tableList(schemaName + "." + tableName)
                        .username(username)
                        .password(password)
                        .decodingPluginName(pluginName)
                        .slotName(slotName)
                        .debeziumProperties(dbzProperties)
                        .deserializer(deserializer)
                        .metricOption(metricOption)
                        .build();
        return SourceFunctionProvider.of(sourceFunction, false);
    }
Metadata that is appended at the end of a physical source row.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/postgre/PostgreSQLTableSource.java