@Override
    public Map<String, String> tableOptions() {
        Map<String, String> options = super.tableOptions();
        if (extractMode == ExtractMode.CDC) {
            options.put("connector", "mysql-cdc-inlong");
            options.put("hostname", hostname);
            options.put("database-name", database);
            if (rowKindsFiltered != null) {
                List<String> rowKinds = rowKindsFiltered.stream().map(RowKindEnum::shortString)
                        .collect(Collectors.toList());
                options.put("row-kinds-filtered", StringUtils.join(rowKinds, "&"));
            }
            if (port != null) {
                options.put("port", port.toString());
            }
            if (serverId != null) {
                options.put("server-id", serverId.toString());
            }
            if (incrementalSnapshotEnabled != null) {
                options.put("scan.incremental.snapshot.enabled", incrementalSnapshotEnabled.toString());
            }
            if (serverTimeZone != null) {
                options.put("server-time-zone", serverTimeZone);
            }
        } else {
            options.put("connector", "jdbc-inlong");
            options.put("url", url);
            Preconditions.checkState(tableNames.size() == 1,
                    "Only support one table for scan extract mode");
        }
        options.put("username", username);
        options.put("password", password);
        String formatTable =
                tableNames.size() == 1 ? tableNames.get(0) : String.format("(%s)", StringUtils.join(tableNames, "|"));
        options.put("table-name", String.format("%s", formatTable));
        return options;
    }
Base constructor

@param id The id of node
@param name The name of node
@param fields The field list of node
@param watermarkField The watermark field of node only used for {@link ExtractMode#CDC}
@param properties The properties connect to mysql
@param primaryKey The primark key connect to mysql
@param tableNames The table names connect to mysql
@param hostname The hostname connect to mysql only used for {@link ExtractMode#CDC}
@param username The username connect to mysql
@param password The password connect to mysql
@param database The database connect to mysql only used for {@link ExtractMode#CDC}
@param port The port connect to mysql only used for {@link ExtractMode#CDC}
@param serverId The server id connect to mysql only used for {@link ExtractMode#CDC}
@param incrementalSnapshotEnabled The incremental snapshot enabled connect to mysql
        only used for {@link ExtractMode#CDC}
@param serverTimeZone The server time zone connect to mysql only used for {@link ExtractMode#CDC}
@param extractMode The extract mode connect mysql {@link ExtractMode}
@param url The url connect to mysql only used for {@link ExtractMode#SCAN}
inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/extract/MySqlExtractNode.java