Lance 底层通过 lance-core（Rust）直接读取存储，其 {@code object_store} 只支持 + * 一组硬编码 scheme（{@code file/s3/gs/az/abfss/oss/cos/hf/memory}）。对于 + * {@code tbdsfs://} / {@code hdfs://} 等 Hadoop 生态的 scheme，Lance 无法识别。 + * + *
本工具类的职责：在 {@code Dataset.open(uri)} 调用之前，如果 {@code uri} 是 + * Hadoop 兼容的 scheme（且不在 Lance 原生支持列表内），则： + *
使用条件：需要在 classpath 中提供对应 scheme 的 Hadoop {@code FileSystem} + * 实现（例如 tbdsfs 需要 {@code tbdsfs-hadoop-*.jar}）。这些 jar 在 TBDS 集群的 + * {@code /usr/local/service/flink/lib/} 下由集群提供，因此 lance-flink 本身 + * 无需绑定。 + * + *
限制：当前实现为整目录 读时全量拷贝，对小/中等规模 lance dataset + * 有效；对超大 dataset 建议后续演进为 range-read 或增量同步策略。 + * + *
本类线程安全；对同一源路径的并发 resolve 请求会串行化为一次下载。
+ */
+public final class LanceHadoopPathResolver {
+
+ private static final Logger LOG = LoggerFactory.getLogger(LanceHadoopPathResolver.class);
+
+ /** Lance native object_store 支持的 scheme，遇到这些直接透传。 */
+ private static final Set 处于 Flink TaskManager 环境中时，由于 YARN 容器不一定可以看到宿主机的
+ * {@code core-site.xml}，上述两种方式作为兽底的配置注入通道。
+ */
+ static Configuration buildHadoopConfiguration(Configuration userConf) {
+ Configuration conf = userConf != null ? userConf : new Configuration();
+
+ // 显式加载 core-site.xml / hdfs-site.xml。在 Flink YARN TaskManager 容器内，
+ // 由于 Flink 的 child-first 类加载器隔离，new Configuration() 未必能通过
+ // context classloader 加载到宿主机的 core-site.xml，导致 tbdsfs.meta 等配置
+ // 缺失（tbdsfs 的 Go 库会因此报 invalid uri 并 fatal）。这里按标准路径兜底加载。
+ loadHadoopSiteXmls(conf);
+
+ // 1. 从系统属性注入 lance.hadoop.* -> hadoop conf key
+ java.util.Properties props = System.getProperties();
+ for (String key : props.stringPropertyNames()) {
+ if (key.startsWith("lance.hadoop.")) {
+ String hadoopKey = key.substring("lance.hadoop.".length());
+ String value = props.getProperty(key);
+ conf.set(hadoopKey, value);
+ LOG.info("Injected Hadoop conf from system property: {} = {}", hadoopKey, value);
+ }
+ }
+
+ // 2. 从环境变量注入 LANCE_HADOOP_XXX -> hadoop conf key
+ java.util.Map Flink SQL 里通过 {@code SET 'flink.hadoop.xxx' = 'yyy'} 设置的配置会以
+ * {@code flink.hadoop.} 前缀进入 Flink 的 {@code Configuration}。这里把这些
+ * 前缀剥掉后注入 Hadoop {@code Configuration}（例如 {@code flink.hadoop.tbdsfs.meta}
+ * → {@code tbdsfs.meta}）。
+ *
+ * 这解决了在 YARN TaskManager 容器内 {@code new Configuration()} 因类加载器
+ * 隔离而加载不到宿主机的 {@code core-site.xml}（进而拿不到 {@code tbdsfs.meta}）
+ * 的问题——tbdsfs 的 Go 库在 {@code meta} 为空时会 fallback 到把 name 当 URI，
+ * 报 {@code invalid uri: /internal} 并直接 fatal 退出。
+ *
+ * @param flinkConf Flink 运行时配置（可为 {@code null}）
+ * @return 注入了 {@code flink.hadoop.*} 配置的 Hadoop {@code Configuration}
+ */
+ public static Configuration buildHadoopConfigurationFromFlink(
+ org.apache.flink.configuration.Configuration flinkConf) {
+ Configuration conf = new Configuration();
+ if (flinkConf == null) {
+ return conf;
+ }
+ for (java.util.Map.Entry 该方法只做"缺省填充"，永远不会覆盖用户已经显式设置的值。
+ */
+ static void applySchemeSpecificDefaults(Configuration conf, String sourceUri, String scheme) {
+ if (conf == null || scheme == null) return;
+ if (!"tbdsfs".equals(scheme)) return;
+
+ // 1. 确保 fs.tbdsfs.impl 存在
+ if (conf.get("fs.tbdsfs.impl") == null) {
+ conf.set("fs.tbdsfs.impl", "io.tbdsfs.TbdsFileSystem");
+ LOG.info("Applied default fs.tbdsfs.impl = io.tbdsfs.TbdsFileSystem");
+ }
+
+ // 2. 如果 tbdsfs.name 未配置，从 URI authority 推导
+ if (conf.get("tbdsfs.name") == null) {
+ try {
+ URI u = new URI(sourceUri);
+ String authority = u.getAuthority();
+ if (authority != null && !authority.isEmpty()) {
+ conf.set("tbdsfs.name", authority);
+ LOG.info("Applied default tbdsfs.name = {} (from URI authority)", authority);
+ }
+ } catch (URISyntaxException ignore) {
+ // fall through; tbdsfs will raise its own error if truly missing
+ }
+ }
+ }
+
+ /**
+ * 规范化 URI：如果输入形如 {@code scheme:/authority/path}（单斜杠，authority 与 path
+ * 之间没有明确分隔），将其转换为 {@code scheme://authority/path}（双斜杠）。
+ * 常见触发场景是 Flink Table 反序列化 URI 时 {@code new Path(str)} 会丢失一个斜杠。
+ * 幂等：对已经形如 {@code scheme://...} 或 {@code scheme:///...}（无 authority）的
+ * URI 不做任何改动。
+ */
+ static String normalizeUri(String uri) {
+ if (uri == null) return null;
+ int colon = uri.indexOf(':');
+ if (colon <= 0 || colon >= uri.length() - 1) return uri;
+ String rest = uri.substring(colon + 1);
+ // 如果已经是 "//..."（含空 authority 的 "///..."），保持原样
+ if (rest.startsWith("//")) return uri;
+ // 只处理 "scheme:/xxx" 且第 2 个字符不是 '/'（否则已经是双斜杠了）
+ if (rest.startsWith("/") && !rest.startsWith("//")) {
+ String scheme = uri.substring(0, colon);
+ return scheme + ":/" + rest; // 把 "scheme:/xxx" 变为 "scheme://xxx"
+ }
+ return uri;
+ }
+
+ /** 提取 scheme。返回小写 scheme，如果没有 scheme 返回 {@code null}。 */
+ static String extractScheme(String uri) {
+ try {
+ URI u = new URI(uri);
+ String s = u.getScheme();
+ return s == null ? null : s.toLowerCase(Locale.ROOT);
+ } catch (URISyntaxException e) {
+ // 非标 URI（比如 Windows 路径 C:\），当作本地路径
+ return null;
+ }
+ }
+
+ static boolean isNativeSupportedScheme(String scheme) {
+ return scheme != null && LANCE_NATIVE_SCHEMES.contains(scheme.toLowerCase(Locale.ROOT));
+ }
+
+ /**
+ * 显式加载 Hadoop 的 {@code core-site.xml} / {@code hdfs-site.xml}。
+ * 用于兜底 Flink YARN 容器内 {@code new Configuration()} 因类加载器隔离而
+ * 加载不到宿主机 site 文件的问题。
+ */
+ private static void loadHadoopSiteXmls(Configuration conf) {
+ java.nio.file.Path confDir = resolveHadoopConfDir();
+ if (confDir == null) {
+ return;
+ }
+ addSiteXmlIfExists(conf, confDir.resolve("core-site.xml"));
+ addSiteXmlIfExists(conf, confDir.resolve("hdfs-site.xml"));
+ }
+
+ private static void addSiteXmlIfExists(Configuration conf, java.nio.file.Path siteXml) {
+ if (Files.isRegularFile(siteXml)) {
+ conf.addResource(new Path(siteXml.toUri()));
+ LOG.info("Explicitly loaded Hadoop site config: {}", siteXml);
+ }
+ }
+
+ /** 探测 Hadoop 配置目录：HADOOP_CONF_DIR → 系统属性 → TBDS 标准路径。 */
+ private static java.nio.file.Path resolveHadoopConfDir() {
+ String env = System.getenv("HADOOP_CONF_DIR");
+ if (env != null && !env.isEmpty()) {
+ return java.nio.file.Paths.get(env);
+ }
+ String prop = System.getProperty("hadoop.conf.dir");
+ if (prop != null && !prop.isEmpty()) {
+ return java.nio.file.Paths.get(prop);
+ }
+ java.nio.file.Path standard = java.nio.file.Paths.get("/usr/local/service/hadoop/etc/hadoop");
+ if (Files.isDirectory(standard)) {
+ return standard;
+ }
+ return null;
+ }
+
+ private static java.nio.file.Path resolveCacheRoot(String userSpecified) {
+ if (userSpecified != null && !userSpecified.isEmpty()) {
+ return java.nio.file.Paths.get(userSpecified);
+ }
+ String tmpDir = System.getProperty("java.io.tmpdir", "/tmp");
+ return java.nio.file.Paths.get(tmpDir, "lance-hadoop-cache");
+ }
+
+ /** 把 URI 转为可作为文件系统目录名的安全字符串。 */
+ private static String sanitize(String uri) {
+ return uri.replaceAll("[^A-Za-z0-9._-]", "_");
+ }
+
+ /** 缓存目录是否已经就绪（存在 {@code _versions/} 子目录，Lance dataset 的标志）。 */
+ private static boolean isCacheReady(java.nio.file.Path targetDir) {
+ if (!Files.isDirectory(targetDir)) return false;
+ java.nio.file.Path versionsDir = targetDir.resolve("_versions");
+ return Files.isDirectory(versionsDir);
+ }
+
+ /**
+ * 递归把 {@code srcDir} 下的所有文件同步到 {@code destDir}（本地目录）。
+ * 保留相对目录结构。
+ */
+ private static void downloadDirectory(Path srcDir, java.nio.file.Path destDir,
+ Configuration conf) throws IOException {
+ FileSystem fs = srcDir.getFileSystem(conf);
+ LOG.info("Resolved Hadoop FileSystem for {}: uri={}, impl={}",
+ srcDir, fs.getUri(), fs.getClass().getName());
+ FileStatus rootStatus = fs.getFileStatus(srcDir);
+ if (!rootStatus.isDirectory()) {
+ throw new IOException("Lance dataset path is not a directory: " + srcDir);
+ }
+ Files.createDirectories(destDir);
+
+ long fileCount = 0L;
+ long byteCount = 0L;
+ int skippedSentinelFiles = 0;
+ RemoteIterator Focus areas:
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ */
+class LanceHadoopPathResolverTest {
+
+ @TempDir
+ Path tempDir;
+
+ @Test
+ void extractSchemeReturnsLowercaseScheme() {
+ assertThat(LanceHadoopPathResolver.extractScheme("S3://bucket/foo")).isEqualTo("s3");
+ assertThat(LanceHadoopPathResolver.extractScheme("TBDSFS://svc/foo")).isEqualTo("tbdsfs");
+ assertThat(LanceHadoopPathResolver.extractScheme("hdfs://nn/p")).isEqualTo("hdfs");
+ assertThat(LanceHadoopPathResolver.extractScheme("file:///tmp/x")).isEqualTo("file");
+ }
+
+ @Test
+ void extractSchemeReturnsNullForBarePath() {
+ assertThat(LanceHadoopPathResolver.extractScheme("/tmp/foo/bar")).isNull();
+ assertThat(LanceHadoopPathResolver.extractScheme("relative/path")).isNull();
+ }
+
+ @Test
+ void normalizeUriRestoresMissingSlashInAuthority() {
+ // 场景 1：Flink Path 反序列化把 "tbdsfs://internal/x" 变成 "tbdsfs:/internal/x"，
+ // normalizeUri 应恢复
+ assertThat(LanceHadoopPathResolver.normalizeUri("tbdsfs:/internal/lance_poc/db1"))
+ .isEqualTo("tbdsfs://internal/lance_poc/db1");
+ // 场景 2：hdfs://ns1/path 也一样
+ assertThat(LanceHadoopPathResolver.normalizeUri("hdfs:/ns1/path"))
+ .isEqualTo("hdfs://ns1/path");
+ }
+
+ @Test
+ void normalizeUriIsIdempotent() {
+ // 已经是双斜杠不变
+ assertThat(LanceHadoopPathResolver.normalizeUri("tbdsfs://internal/x"))
+ .isEqualTo("tbdsfs://internal/x");
+ // 空 authority 的 "scheme:///" 也不变
+ assertThat(LanceHadoopPathResolver.normalizeUri("file:///tmp/x"))
+ .isEqualTo("file:///tmp/x");
+ // 本地路径不变
+ assertThat(LanceHadoopPathResolver.normalizeUri("/tmp/x")).isEqualTo("/tmp/x");
+ assertThat(LanceHadoopPathResolver.normalizeUri(null)).isNull();
+ }
+
+ @Test
+ void isNativeSupportedSchemeRecognizesLanceSchemes() {
+ assertThat(LanceHadoopPathResolver.isNativeSupportedScheme("s3")).isTrue();
+ assertThat(LanceHadoopPathResolver.isNativeSupportedScheme("S3")).isTrue();
+ assertThat(LanceHadoopPathResolver.isNativeSupportedScheme("file")).isTrue();
+ assertThat(LanceHadoopPathResolver.isNativeSupportedScheme("cos")).isTrue();
+ assertThat(LanceHadoopPathResolver.isNativeSupportedScheme("tbdsfs")).isFalse();
+ assertThat(LanceHadoopPathResolver.isNativeSupportedScheme("hdfs")).isFalse();
+ assertThat(LanceHadoopPathResolver.isNativeSupportedScheme(null)).isFalse();
+ }
+
+ @Test
+ void resolveForReadPassesThroughNullAndEmpty() {
+ assertThat(LanceHadoopPathResolver.resolveForRead(null, null, null)).isNull();
+ assertThat(LanceHadoopPathResolver.resolveForRead("", null, null)).isEmpty();
+ }
+
+ @Test
+ void resolveForReadPassesThroughLocalPath() {
+ String local = "/tmp/lance/dataset.lance";
+ String resolved = LanceHadoopPathResolver.resolveForRead(local, null, null);
+ assertThat(resolved).isEqualTo(local);
+ }
+
+ @Test
+ void resolveForReadPassesThroughFileUri() {
+ String fileUri = "file:///tmp/lance/dataset.lance";
+ String resolved = LanceHadoopPathResolver.resolveForRead(fileUri, null, null);
+ assertThat(resolved).isEqualTo(fileUri);
+ }
+
+ @Test
+ void resolveForReadPassesThroughS3Uri() {
+ String s3 = "s3://bucket/prefix/dataset.lance";
+ String resolved = LanceHadoopPathResolver.resolveForRead(s3, null, null);
+ assertThat(resolved).isEqualTo(s3);
+ }
+
+ @Test
+ void resolveForReadPassesThroughCosUri() {
+ String cos = "cos://bucket/prefix/dataset.lance";
+ String resolved = LanceHadoopPathResolver.resolveForRead(cos, null, null);
+ assertThat(resolved).isEqualTo(cos);
+ }
+
+ /**
+ * 使用自定义 scheme {@code mytestfs://}（lance 不原生支持），并通过 Hadoop
+ * 配置将该 scheme 重定向到 {@link org.apache.hadoop.fs.LocalFileSystem}，从而完整
+ * 走通 Hadoop FS 下载链路，且不依赖真实 HDFS。
+ */
+ @Test
+ void resolveForReadDownloadsFromHadoopFileSystem() throws IOException {
+ // 1. 构造一个“源” lance dataset 目录（用本地磁盘模拟）
+ Path srcDir = tempDir.resolve("src_dataset.lance");
+ Path versionsDir = srcDir.resolve("_versions");
+ Path dataDir = srcDir.resolve("data");
+ Files.createDirectories(versionsDir);
+ Files.createDirectories(dataDir);
+ Files.write(versionsDir.resolve("1.manifest"), new byte[] {1, 2, 3, 4});
+ Files.write(dataDir.resolve("part-0.lance"), "hello".getBytes());
+ Files.write(srcDir.resolve("_latest.manifest"), new byte[] {9});
+
+ // 2. 配置：把自定义 scheme mytestfs 重定向到 Hadoop LocalFileSystem
+ Configuration conf = new Configuration(false);
+ conf.set("fs.mytestfs.impl", LocalFsWithMytestfsScheme.class.getName());
+
+ // 3. 使用 mytestfs:// 前缀访问源目录，触发 Hadoop FS 下载路径
+ String srcUri = "mytestfs://" + srcDir.toAbsolutePath();
+ Path cacheRoot = tempDir.resolve("cache");
+
+ String resolved = LanceHadoopPathResolver.resolveForRead(
+ srcUri, conf, cacheRoot.toAbsolutePath().toString());
+
+ // 4. 验证返回的是本地 file:// 路径，并且包含预期的文件
+ assertThat(resolved).startsWith("file://");
+ Path resolvedDir = Path.of(resolved.substring("file://".length()));
+ assertThat(Files.isDirectory(resolvedDir)).isTrue();
+ assertThat(Files.isDirectory(resolvedDir.resolve("_versions"))).isTrue();
+ assertThat(Files.isRegularFile(resolvedDir.resolve("_versions/1.manifest"))).isTrue();
+ assertThat(Files.isRegularFile(resolvedDir.resolve("data/part-0.lance"))).isTrue();
+ assertThat(Files.readString(resolvedDir.resolve("data/part-0.lance"))).isEqualTo("hello");
+ }
+
+ /**
+ * 二次 resolve 相同 URI 时应命中缓存，不重复下载。
+ */
+ @Test
+ void resolveForReadReusesCacheOnSecondCall() throws IOException {
+ Path srcDir = tempDir.resolve("cached_dataset.lance");
+ Path versionsDir = srcDir.resolve("_versions");
+ Files.createDirectories(versionsDir);
+ Files.write(versionsDir.resolve("1.manifest"), new byte[] {1});
+
+ Configuration conf = new Configuration(false);
+ conf.set("fs.mytestfs.impl", LocalFsWithMytestfsScheme.class.getName());
+
+ String srcUri = "mytestfs://" + srcDir.toAbsolutePath();
+ Path cacheRoot = tempDir.resolve("cache2");
+
+ String first = LanceHadoopPathResolver.resolveForRead(
+ srcUri, conf, cacheRoot.toAbsolutePath().toString());
+ // 修改缓存目录里的文件，模拟“已存在的缓存”
+ Path firstDir = Path.of(first.substring("file://".length()));
+ Path sentinel = firstDir.resolve("_versions/1.manifest");
+ Files.write(sentinel, new byte[] {42});
+
+ String second = LanceHadoopPathResolver.resolveForRead(
+ srcUri, conf, cacheRoot.toAbsolutePath().toString());
+ // 第二次 resolve 应返回同一个目录，且 sentinel 未被重新下载覆盖
+ assertThat(second).isEqualTo(first);
+ assertThat(Files.readAllBytes(sentinel)).containsExactly(42);
+ }
+
+ /**
+ * 将 Hadoop {@link org.apache.hadoop.fs.LocalFileSystem} 包装为 {@code mytestfs://} scheme，
+ * 测试专用。它仅重写 {@link #getScheme()} 以及将传入的 {@code mytestfs://