@Override
    public DynamicTableSource createDynamicTableSource(Context context) {
        final FactoryUtil.TableFactoryHelper helper =
                FactoryUtil.createTableFactoryHelper(this, context);
        helper.validateExcept(DebeziumOptions.DEBEZIUM_OPTIONS_PREFIX);

        final ReadableConfig config = helper.getOptions();
        String hostname = config.get(HOSTNAME);
        String username = config.get(USERNAME);
        String password = config.get(PASSWORD);
        String databaseName = config.get(DATABASE_NAME);
        checkNotNull(databaseName);
        // During the incremental phase, Debezium uses the uppercase database name.
        // However, during the snapshot phase, the database name is user-configurable.
        // To avoid inconsistencies between the database names in the snapshot and incremental phases,
        // it is necessary to convert the database name to uppercase when constructing the Oracle Source.
        // For more details, please refer to: https://github.com/apache/inlong/pull/7865.
        databaseName = databaseName.toUpperCase();
        String tableName = config.get(TABLE_NAME);
        String schemaName = config.get(SCHEMA_NAME);
        int port = config.get(PORT);
        StartupOptions startupOptions = getStartupOptions(config);
        final boolean sourceMultipleEnable = config.get(SOURCE_MULTIPLE_ENABLE);
        ResolvedSchema physicalSchema = context.getCatalogTable().getResolvedSchema();
        String inlongMetric = config.getOptional(INLONG_METRIC).orElse(null);
        String inlongAudit = config.get(INLONG_AUDIT);
        boolean enableParallelRead = config.get(SCAN_INCREMENTAL_SNAPSHOT_ENABLED);
        int splitSize = config.get(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE);
        int splitMetaGroupSize = config.get(CHUNK_META_GROUP_SIZE);
        int fetchSize = config.get(SCAN_SNAPSHOT_FETCH_SIZE);
        Duration connectTimeout = config.get(CONNECT_TIMEOUT);
        int connectMaxRetries = config.get(CONNECT_MAX_RETRIES);
        int connectionPoolSize = config.get(CONNECTION_POOL_SIZE);
        double distributionFactorUpper = config.get(SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND);
        double distributionFactorLower = config.get(SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND);
        String chunkKeyColumn =
                config.getOptional(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN).orElse(null);
        String serverTimezone = config.get(SERVER_TIME_ZONE);

        if (enableParallelRead) {
            validateIntegerOption(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize, 1);
            validateIntegerOption(SCAN_SNAPSHOT_FETCH_SIZE, fetchSize, 1);
            validateIntegerOption(CHUNK_META_GROUP_SIZE, splitMetaGroupSize, 1);
            validateIntegerOption(CONNECTION_POOL_SIZE, connectionPoolSize, 1);
            validateIntegerOption(CONNECT_MAX_RETRIES, connectMaxRetries, 0);
            validateDistributionFactorUpper(distributionFactorUpper);
            validateDistributionFactorLower(distributionFactorLower);
        }
        return new OracleTableSource(
                physicalSchema,
                port,
                hostname,
                databaseName,
                tableName,
                schemaName,
                username,
                password,
                getDebeziumProperties(context.getCatalogTable().getOptions()),
                startupOptions,
                sourceMultipleEnable,
                inlongMetric,
                inlongAudit,
                enableParallelRead,
                splitSize,
                splitMetaGroupSize,
                fetchSize,
                connectTimeout,
                connectMaxRetries,
                connectionPoolSize,
                distributionFactorUpper,
                distributionFactorLower,
                chunkKeyColumn);
    }
Factory for creating configured instance of {@link OracleTableSource}.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/table/OracleTableSourceFactory.java