@Override
    public Map<String, String> tableOptions() {
        Map<String, String> map = super.tableOptions();
        map.put("connector", "filesystem");
        map.put("path", path);
        map.put("format", format);
        if (null == getProperties() || !getProperties().containsKey(sourceMonitorInterval)) {
            map.put(sourceMonitorInterval, Duration.ofMinutes(1L).toString());
        }
        return map;
    }
The feature version of flink is greater than or equal to 1.16.x for stream table
inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/extract/FileSystemExtractNode.java