@Override
    public Map<String, String> tableOptions() {
        Map<String, String> options = super.tableOptions();

        options.put(DATA_TYPE, dataType);
        options.put(REDIS_MODE, clusterMode);
        options.put(SCHEMA_MAPPING_MODE, schemaMapMode);
        if (CLUSTER_MODE_STANDALONE.equals(clusterMode)) {
            options.put(VALUE_HOST, host);
            options.put(VALUE_PORT, String.valueOf(port));
            if (StringUtils.isNotBlank(password)) {
                options.put(PASSWORD, password);
            }
        } else if (CLUSTER_MODE_CLUSTER.equals(clusterMode)) {
            options.put(CLUSTER_NODES, clusterNodes);
            if (StringUtils.isNotBlank(password)) {
                options.put(CLUSTER_PASSWORD, password);
            }
        } else {
            options.put(MASTER_NAME, masterName);
            options.put(SENTINELS_INFO, sentinelsInfo);
            if (StringUtils.isNotBlank(password)) {
                options.put(SENTINELS_PASSWORD, password);
            }
        }
        if (database != null) {
            options.put(DATABASE, String.valueOf(database));
        }
        if (maxIdle != null) {
            options.put(MAX_IDLE, String.valueOf(maxIdle));
        }
        if (maxRetries != null) {
            options.put(SINK_MAX_RETRIES, String.valueOf(maxRetries));
        }

        if (maxTotal != null) {
            options.put(MAX_TOTAL, String.valueOf(maxTotal));
        }
        if (minIdle != null) {
            options.put(MIN_IDLE, String.valueOf(minIdle));
        }
        if (soTimeout != null) {
            options.put(SO_TIMEOUT, String.valueOf(soTimeout));
        }
        if (timeout != null) {
            options.put(TIMEOUT, String.valueOf(timeout));
        }
        if (ttl != null) {
            options.put(EXPIRE_TIME, ttl + "s");
        }

        options.putAll(format.generateOptions(false));

        // If the extend attributes starts with .ddl,
        // it will be passed to the ddl statement of the table
        Map<String, String> properties = getProperties();
        if (properties != null) {
            properties.forEach((keyName, ddlValue) -> {
                if (StringUtils.isNoneBlank(keyName) && keyName.startsWith(DDL_ATTR_PREFIX)) {
                    String ddlKeyName = keyName.substring(DDL_ATTR_PREFIX.length());
                    options.put(ddlKeyName, ddlValue);
                }
            });
        }

        options.put(CONNECTOR_KEY, CONNECTOR_REDIS_INLONG);

        return options;
    }
The prefix of ddl attr parsed from frontend advanced properties.
inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/load/RedisLoadNode.java