@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())
                        .setMigrateAll(sourceMultipleEnable)
                        .setServerTimeZone(ZoneId.of(serverTimeZone))
                        .setAppendSource(appendSource)
                        .setValidator(new RowKindValidator(rowKindFiltered))
                        .build();

        if (enableParallelRead) {
            LOGGER.info("in  PostgreSQLTableSource, enableParallelRead is true");
            PostgresSourceBuilder.PostgresIncrementalSource<RowData> parallelSource =
                    new PostgresSourceBuilder<RowData>()
                            .hostname(hostname)
                            .port(port)
                            .database(database)
                            .tableList(tableName)
                            .username(username)
                            .password(password)
                            .inlongMetric(inlongMetric)
                            .inlongAudit(inlongAudit)
                            .decodingPluginName(pluginName)
                            .slotName(slotName)
                            .debeziumProperties(dbzProperties)
                            .deserializer(deserializer)
                            .splitSize(splitSize)
                            .splitMetaGroupSize(splitMetaGroupSize)
                            .distributionFactorUpper(distributionFactorUpper)
                            .distributionFactorLower(distributionFactorLower)
                            .fetchSize(fetchSize)
                            .connectTimeout(connectTimeout)
                            .connectMaxRetries(connectMaxRetries)
                            .connectionPoolSize(connectionPoolSize)
                            .startupOptions(startupOptions)
                            .chunkKeyColumn(chunkKeyColumn)
                            .heartbeatInterval(heartbeatInterval)
                            .build();
            return SourceProvider.of(parallelSource);
        } else {
            LOGGER.info("in  PostgreSQLTableSource, enableParallelRead is false");
            DebeziumSourceFunction<RowData> sourceFunction =
                    PostgreSQLSource.<RowData>builder()
                            .hostname(hostname)
                            .port(port)
                            .database(database)
                            .schemaList(schemaName.split(","))
                            .tableList(tableName)
                            .username(username)
                            .password(password)
                            .decodingPluginName(pluginName)
                            .slotName(slotName)
                            .debeziumProperties(dbzProperties)
                            .deserializer(deserializer)
                            .inlongMetric(inlongMetric)
                            .inlongAudit(inlongAudit)
                            .migrateAll(sourceMultipleEnable)
                            .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.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/table/PostgreSQLTableSource.java