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

        DebeziumDeserializationSchema<RowData> deserializer =
                new MongoDBConnectorDeserializationSchema(
                        physicalDataType, metadataConverters, typeInfo,
                        localTimeZone, new MongoDBRowKindValidator(rowValidator), sourceMultipleEnable);

        String databaseList = null;
        String collectionList = null;
        if (StringUtils.isNotEmpty(database) && StringUtils.isNotEmpty(collection)) {
            // explicitly specified database and collection.
            if (!containsRegexMetaCharacters(database)
                    && !containsRegexMetaCharacters(collection)) {
                checkDatabaseNameValidity(database);
                checkCollectionNameValidity(collection);
                databaseList = database;
                collectionList = database + "." + collection;
            } else {
                databaseList = database;
                collectionList = collection;
            }
        } else if (StringUtils.isNotEmpty(database)) {
            databaseList = collection;
        } else if (StringUtils.isNotEmpty(collection)) {
            collectionList = collection;
        } else {
            // Watching all changes on the cluster by default, we do nothing here
        }

        if (enableParallelRead) {
            MongoDBSourceBuilder<RowData> builder =
                    MongoDBSource.<RowData>builder().hosts(hosts).deserializer(deserializer);

            Optional.ofNullable(databaseList).ifPresent(builder::databaseList);
            Optional.ofNullable(collectionList).ifPresent(builder::collectionList);
            Optional.ofNullable(username).ifPresent(builder::username);
            Optional.ofNullable(password).ifPresent(builder::password);
            Optional.ofNullable(connectionOptions).ifPresent(builder::connectionOptions);
            Optional.ofNullable(copyExisting).ifPresent(builder::copyExisting);
            Optional.ofNullable(batchSize).ifPresent(builder::batchSize);
            Optional.ofNullable(pollMaxBatchSize).ifPresent(builder::pollMaxBatchSize);
            Optional.ofNullable(pollAwaitTimeMillis).ifPresent(builder::pollAwaitTimeMillis);
            Optional.ofNullable(heartbeatIntervalMillis)
                    .ifPresent(builder::heartbeatIntervalMillis);
            Optional.ofNullable(splitMetaGroupSize).ifPresent(builder::splitMetaGroupSize);
            Optional.ofNullable(splitSizeMB).ifPresent(builder::splitSizeMB);
            Optional.ofNullable(inlongMetric).ifPresent(builder::inlongMetric);
            Optional.ofNullable(inlongAudit).ifPresent(builder::inlongAudit);

            return SourceProvider.of(builder.build());
        } else {
            org.apache.inlong.sort.cdc.mongodb.MongoDBSource.Builder<RowData> builder =
                    org.apache.inlong.sort.cdc.mongodb.MongoDBSource.<RowData>builder().hosts(hosts)
                            .deserializer(deserializer);

            Optional.ofNullable(databaseList).ifPresent(builder::databaseList);
            Optional.ofNullable(collectionList).ifPresent(builder::collectionList);
            Optional.ofNullable(username).ifPresent(builder::username);
            Optional.ofNullable(password).ifPresent(builder::password);
            Optional.ofNullable(connectionOptions).ifPresent(builder::connectionOptions);
            Optional.ofNullable(copyExisting).ifPresent(builder::copyExisting);
            Optional.ofNullable(copyExistingQueueSize).ifPresent(builder::copyExistingQueueSize);
            Optional.ofNullable(batchSize).ifPresent(builder::batchSize);
            Optional.ofNullable(pollMaxBatchSize).ifPresent(builder::pollMaxBatchSize);
            Optional.ofNullable(pollAwaitTimeMillis).ifPresent(builder::pollAwaitTimeMillis);
            Optional.ofNullable(heartbeatIntervalMillis).ifPresent(builder::heartbeatIntervalMillis);
            Optional.ofNullable(inlongMetric).ifPresent(builder::inlongMetric);
            Optional.ofNullable(inlongAudit).ifPresent(builder::inlongAudit);
            Optional.ofNullable(sourceMultipleEnable).ifPresent(builder::migrateAll);
            DebeziumSourceFunction<RowData> sourceFunction = builder.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/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/table/MongoDBTableSource.java