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

        options.put(READ_AS_STREAMING, ENABLE_CODE);
        options.put(READ_STREAMING_CHECK_INTERVAL, String.valueOf(checkIntervalInMinus));

        // Synchronization to Metastore is enabled by default,
        // which can be modified in the front-end configuration
        options.put(HUDI_OPTION_HIVE_SYNC_ENABLED, ENABLE_CODE);
        options.put(HUDI_OPTION_HIVE_SYNC_MODE, HUDI_OPTION_HIVE_SYNC_MODE_HMS_VALUE);
        options.put(HUDI_OPTION_HIVE_SYNC_DB, dbName);
        options.put(HUDI_OPTION_HIVE_SYNC_TABLE, tableName);
        options.put(HUDI_OPTION_HIVE_SYNC_METASTORE_URIS, uri);

        // If the extend attributes starts with .ddl,
        // it will be passed to the ddl statement of the table
        if (CollectionUtils.isNotEmpty(extList)) {
            extList.forEach(ext -> {
                String keyName = ext.get(EXTEND_ATTR_KEY_NAME);
                if (StringUtils.isNoneBlank(keyName) &&
                        keyName.startsWith(DDL_ATTR_PREFIX)) {
                    String ddlKeyName = keyName.substring(DDL_ATTR_PREFIX.length());
                    String ddlValue = ext.get(EXTEND_ATTR_VALUE_NAME);
                    options.put(ddlKeyName, ddlValue);
                }
            });
        }

        String path = String.format("%s/%s.db/%s", warehouse, dbName, tableName);
        options.put(HUDI_OPTION_DEFAULT_PATH, path);

        // read options
        options.put(READ_START_COMMIT, readStartCommit);
        options.put(READ_STREAMING_SKIP_COMPACT, String.valueOf(readStreamingSkipCompaction));

        options.put(HUDI_OPTION_DATABASE_NAME, dbName);
        options.put(HUDI_OPTION_TABLE_NAME, tableName);
        options.put(CONNECTOR_KEY, CONNECTOR);

        return options;
    }
Hudi extract node for extract data from hudi
inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/extract/HudiExtractNode.java