feat: support reading Lance tables on Hadoop-family filesystems (tbdsfs/hdfs)

lance-flink-1.18/pom.xml

@@ -80,6 +80,29 @@
             <artifactId>flink-connector-base</artifactId>
         </dependency>
 
+        <!-- Hadoop（用于 tbdsfs / hdfs 等 Hadoop 兼容 FileSystem 前置缓存下载；
+             运行时由集群提供，因此 scope=provided 不打进 fat jar） -->
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-common</artifactId>
+            <version>3.3.6</version>
+            <scope>provided</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-log4j12</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>log4j</groupId>
+                    <artifactId>log4j</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-reload4j</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
         <!-- 日志 -->
         <dependency>
             <groupId>org.slf4j</groupId>

lance-flink-1.19/pom.xml

@@ -80,6 +80,29 @@
             <artifactId>flink-connector-base</artifactId>
         </dependency>
 
+        <!-- Hadoop（用于 tbdsfs / hdfs 等 Hadoop 兼容 FileSystem 前置缓存下载；
+             运行时由集群提供，因此 scope=provided 不打进 fat jar） -->
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-common</artifactId>
+            <version>3.3.6</version>
+            <scope>provided</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-log4j12</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>log4j</groupId>
+                    <artifactId>log4j</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-reload4j</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
         <!-- 日志 -->
         <dependency>
             <groupId>org.slf4j</groupId>

lance-flink-1.20/pom.xml

@@ -80,6 +80,29 @@
             <artifactId>flink-connector-base</artifactId>
         </dependency>
 
+        <!-- Hadoop（用于 tbdsfs / hdfs 等 Hadoop 兼容 FileSystem 前置缓存下载；
+             运行时由集群提供，因此 scope=provided 不打进 fat jar） -->
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-common</artifactId>
+            <version>3.3.6</version>
+            <scope>provided</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-log4j12</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>log4j</groupId>
+                    <artifactId>log4j</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-reload4j</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
         <!-- 日志 -->
         <dependency>
             <groupId>org.slf4j</groupId>

pom.xml

@@ -323,10 +323,16 @@
                                     <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"/>
                                 </transformers>
                                 <relocations>
-                                    <relocation>
-                                        <pattern>org.apache.arrow</pattern>
-                                        <shadedPattern>org.apache.flink.connector.lance.shaded.arrow</shadedPattern>
-                                    </relocation>
+                                    <!--
+                                      NOTE: Do NOT relocate org.apache.arrow.
+
+                                      lance-core's Rust JNI hard-codes Arrow C Data Interface
+                                      class names (e.g. org.apache.arrow.c.jni.PrivateData,
+                                      org.apache.arrow.c.Data, org.apache.arrow.c.ArrowArray)
+                                      and resolves them via FindClass() at runtime. Relocating
+                                      org.apache.arrow therefore breaks the JNI bridge and fails
+                                      with: Could not find class Lorg/apache/arrow/c/jni/PrivateData;
+                                    -->
                                 </relocations>
                             </configuration>
                         </execution>

src/main/java/org/apache/flink/connector/lance/LanceSource.java

@@ -128,7 +128,7 @@ public void open(Configuration parameters) throws Exception {
         Path path = Paths.get(datasetPath);
         try {
             // Honor read.version / read.as-of-timestamp for time-travel reads (issue #5).
-            this.dataset = LanceOpener.open(path.toString(), allocator, options);
+            this.dataset = LanceOpener.open(path.toString(), allocator, options, parameters);
         } catch (Exception e) {
             throw new IOException("Cannot open Lance dataset: " + datasetPath, e);
         }

src/main/java/org/apache/flink/connector/lance/config/LanceOptions.java

@@ -25,7 +25,9 @@
 import java.io.Serializable;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 
 /**
@@ -394,6 +396,7 @@ public static MetricType fromValue(String value) {
     private final Integer vectorRefineFactor;
     private final String defaultDatabase;
     private final String warehouse;
+    private final Map<String, String> hadoopConfig;
 
     private LanceOptions(Builder builder) {
         this.path = builder.path;
@@ -421,6 +424,9 @@ private LanceOptions(Builder builder) {
         this.vectorRefineFactor = builder.vectorRefineFactor;
         this.defaultDatabase = builder.defaultDatabase;
         this.warehouse = builder.warehouse;
+        this.hadoopConfig = builder.hadoopConfig == null
+                ? Collections.emptyMap()
+                : Collections.unmodifiableMap(new HashMap<>(builder.hadoopConfig));
     }
 
     // ==================== Getter Methods ====================
@@ -525,6 +531,16 @@ public String getWarehouse() {
         return warehouse;
     }
 
+    /**
+     * Extra Hadoop configuration key/value pairs (e.g. {@code tbdsfs.meta}) that should be
+     * injected into the Hadoop {@link org.apache.hadoop.conf.Configuration} used to resolve
+     * Hadoop-family dataset paths (tbdsfs/hdfs). Populated from SQL {@code WITH} options
+     * prefixed with {@code hadoop.}.
+     */
+    public Map<String, String> getHadoopConfig() {
+        return hadoopConfig;
+    }
+
     // ==================== Builder ====================
 
     public static Builder builder() {
@@ -631,6 +647,7 @@ public static class Builder {
         private Integer vectorRefineFactor;
         private String defaultDatabase = "default";
         private String warehouse;
+        private Map<String, String> hadoopConfig;
 
         public Builder path(String path) {
             this.path = path;
@@ -757,6 +774,11 @@ public Builder warehouse(String warehouse) {
             return this;
         }
 
+        public Builder hadoopConfig(Map<String, String> hadoopConfig) {
+            this.hadoopConfig = hadoopConfig;
+            return this;
+        }
+
         /**
          * Build LanceOptions instance with validation
          */

src/main/java/org/apache/flink/connector/lance/table/LanceDynamicTableFactory.java

@@ -28,7 +28,9 @@
 import org.apache.flink.table.factories.DynamicTableSourceFactory;
 import org.apache.flink.table.factories.FactoryUtil;
 
+import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Map;
 import java.util.Set;
 
 /**
@@ -187,10 +189,11 @@ public Set<ConfigOption<?>> optionalOptions() {
     @Override
     public DynamicTableSource createDynamicTableSource(Context context) {
         FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
-        helper.validate();
+        Map<String, String> tableOptions = context.getCatalogTable().getOptions();
+        helper.validateExcept(extractHadoopOptionKeys(tableOptions));
 
         ReadableConfig config = helper.getOptions();
-        LanceOptions options = buildLanceOptions(config);
+        LanceOptions options = buildLanceOptions(config, tableOptions);
 
         return new LanceDynamicTableSource(
                 options,
@@ -201,21 +204,34 @@ public DynamicTableSource createDynamicTableSource(Context context) {
     @Override
     public DynamicTableSink createDynamicTableSink(Context context) {
         FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
-        helper.validate();
+        Map<String, String> tableOptions = context.getCatalogTable().getOptions();
+        helper.validateExcept(extractHadoopOptionKeys(tableOptions));
 
         ReadableConfig config = helper.getOptions();
-        LanceOptions options = buildLanceOptions(config);
+        LanceOptions options = buildLanceOptions(config, tableOptions);
 
         return new LanceDynamicTableSink(
                 options,
                 context.getCatalogTable().getResolvedSchema().toPhysicalRowDataType()
         );
     }
 
+    /**
+     * 提取以 {@code hadoop.} 为前缀的选项 key，供 {@code validateExcept} 跳过校验。
+     */
+    private String[] extractHadoopOptionKeys(Map<String, String> tableOptions) {
+        if (tableOptions == null) {
+            return new String[0];
+        }
+        return tableOptions.keySet().stream()
+                .filter(key -> key != null && key.startsWith("hadoop."))
+                .toArray(String[]::new);
+    }
+
     /**
      * Build LanceOptions from configuration
      */
-    private LanceOptions buildLanceOptions(ReadableConfig config) {
+    private LanceOptions buildLanceOptions(ReadableConfig config, Map<String, String> tableOptions) {
         LanceOptions.Builder builder = LanceOptions.builder();
 
         // Common configuration
@@ -248,6 +264,31 @@ private LanceOptions buildLanceOptions(ReadableConfig config) {
         builder.vectorMetric(LanceOptions.MetricType.fromValue(config.get(VECTOR_METRIC)));
         builder.vectorNprobes(config.get(VECTOR_NPROBES));
 
+        // Hadoop-family FileSystem 配置（例如 hadoop.tbdsfs.meta），用于 tbdsfs/hdfs 路径解析
+        Map<String, String> hadoopConfig = extractHadoopConfig(tableOptions);
+        if (!hadoopConfig.isEmpty()) {
+            builder.hadoopConfig(hadoopConfig);
+        }
+
         return builder.build();
     }
+
+    /**
+     * 从表 DDL 的原始 WITH 参数中提取以 {@code hadoop.} 为前缀的配置项，去掉前缀后作为
+     * Hadoop Configuration 的 key/value（例如 {@code 'hadoop.tbdsfs.meta' = 'zk://...'}
+     * → {@code tbdsfs.meta = zk://...}）。用于绕过集群部分节点 core-site.xml 缺配置的问题。
+     */
+    private Map<String, String> extractHadoopConfig(Map<String, String> tableOptions) {
+        Map<String, String> result = new HashMap<>();
+        if (tableOptions == null) {
+            return result;
+        }
+        for (Map.Entry<String, String> entry : tableOptions.entrySet()) {
+            String key = entry.getKey();
+            if (key != null && key.startsWith("hadoop.") && entry.getValue() != null) {
+                result.put(key.substring("hadoop.".length()), entry.getValue());
+            }
+        }
+        return result;
+    }
 }

src/main/java/org/apache/flink/connector/lance/table/LanceDynamicTableSource.java

@@ -141,6 +141,11 @@ LanceOptions buildRuntimeOptions(RowType rowType) {
                 .readBatchSize(options.getReadBatchSize())
                 .readFilter(buildFilterExpression());
 
+        // 携带 hadoop.* 配置（如 tbdsfs.meta），避免投影/过滤下推重建 options 时丢失
+        if (options.getHadoopConfig() != null && !options.getHadoopConfig().isEmpty()) {
+            optionsBuilder.hadoopConfig(options.getHadoopConfig());
+        }
+
         // Carry over time-travel options from the SQL WITH clause (issue #5).
         // Without this the readVersion / readAsOfTimestamp get dropped when the planner
         // rebuilds options during projection/filter push-down.