Detect Source Table Format

xtable-api/pom.xml

@@ -29,6 +29,10 @@
     <name>XTable Project API</name>
 
     <dependencies>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+        </dependency>
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-annotations</artifactId>
@@ -88,5 +92,37 @@
             <groupId>org.mockito</groupId>
             <artifactId>mockito-junit-jupiter</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.iceberg</groupId>
+            <artifactId>iceberg-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.iceberg</groupId>
+            <artifactId>iceberg-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.delta</groupId>
+            <artifactId>delta-core_2.12</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.delta</groupId>
+            <artifactId>delta-core_2.12</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.delta</groupId>
+            <artifactId>delta-core_2.12</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.delta</groupId>
+            <artifactId>delta-core_2.12</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.delta</groupId>
+            <artifactId>delta-kernel-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.delta</groupId>
+            <artifactId>delta-kernel-defaults</artifactId>
+        </dependency>
     </dependencies>
 </project>

xtable-api/src/main/java/org/apache/xtable/conversion/ExternalTable.java

@@ -24,6 +24,7 @@
 import lombok.Getter;
 import lombok.NonNull;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 
 import com.google.common.base.Preconditions;
@@ -50,22 +51,26 @@ class ExternalTable {
   /** Optional, additional properties that can be used to define interactions with the table */
   protected final Properties additionalProperties;
 
+  protected final Configuration hadoopConf;
+
   ExternalTable(
       @NonNull String name,
       @NonNull String formatName,
       @NonNull String basePath,
       String[] namespace,
       CatalogConfig catalogConfig,
-      Properties additionalProperties) {
+      Properties additionalProperties,
+      Configuration hadoopConf) {
     this.name = name;
     this.formatName = formatName;
     this.basePath = sanitizeBasePath(basePath);
     this.namespace = namespace;
     this.catalogConfig = catalogConfig;
     this.additionalProperties = additionalProperties;
+    this.hadoopConf = hadoopConf;
   }
 
-  protected String sanitizeBasePath(String tableBasePath) {
+  protected static String sanitizeBasePath(String tableBasePath) {
     Path path = new Path(tableBasePath);
     Preconditions.checkArgument(path.isAbsolute(), "Table base path must be absolute");
     if (path.isAbsoluteAndSchemeAuthorityNull()) {

xtable-api/src/main/java/org/apache/xtable/conversion/SourceTable.java

@@ -18,13 +18,17 @@
 
 package org.apache.xtable.conversion;
 
+import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.util.Properties;
 
 import lombok.Builder;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.NonNull;
 
+import org.apache.hadoop.conf.Configuration;
+
 @EqualsAndHashCode(callSuper = true)
 @Getter
 public class SourceTable extends ExternalTable {
@@ -39,8 +43,41 @@ public SourceTable(
       String dataPath,
       String[] namespace,
       CatalogConfig catalogConfig,
-      Properties additionalProperties) {
-    super(name, formatName, basePath, namespace, catalogConfig, additionalProperties);
+      Properties additionalProperties,
+      Configuration hadoopConf) {
+    super(name, formatName, basePath, namespace, catalogConfig, additionalProperties, hadoopConf);
     this.dataPath = dataPath == null ? this.getBasePath() : sanitizeBasePath(dataPath);
   }
+
+  public static SourceTable withDetectedFormat(
+      String name,
+      String basePath,
+      String dataPath,
+      String[] namespace,
+      CatalogConfig catalogConfig,
+      Properties additionalProperties,
+      Configuration hadoopConf) {
+
+    Configuration resolvedConf = hadoopConf != null ? hadoopConf : new Configuration();
+    String detectedFormat = resolveFormatOrThrow(basePath, resolvedConf);
+
+    return SourceTable.builder()
+        .name(name)
+        .formatName(detectedFormat)
+        .basePath(basePath)
+        .dataPath(dataPath)
+        .namespace(namespace)
+        .catalogConfig(catalogConfig)
+        .additionalProperties(additionalProperties)
+        .hadoopConf(resolvedConf)
+        .build();
+  }
+
+  private static String resolveFormatOrThrow(String basePath, Configuration hadoopConf) {
+    try {
+      return SourceTableFormatDetector.detectFormat(basePath, hadoopConf);
+    } catch (IOException e) {
+      throw new UncheckedIOException("Failed to auto-detect source table format", e);
+    }
+  }
 }

xtable-api/src/main/java/org/apache/xtable/conversion/SourceTableFormatDetector.java

@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.conversion;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.hadoop.HadoopTables;
+
+import io.delta.kernel.Snapshot;
+import io.delta.kernel.defaults.engine.DefaultEngine;
+import io.delta.kernel.engine.Engine;
+import io.delta.kernel.internal.SnapshotImpl;
+
+import org.apache.xtable.model.storage.TableFormat;
+
+@Log4j2
+public class SourceTableFormatDetector {
+  // private constructor to prevent instantiation
+  private SourceTableFormatDetector() {
+    throw new UnsupportedOperationException("This class cannot be instantiated");
+  }
+  // helper method to detect input format
+  public static String detectFormat(String pathStr, Configuration conf) throws IOException {
+    String sanitizeBasePath = ExternalTable.sanitizeBasePath(pathStr);
+    Path basePath = new Path(sanitizeBasePath);
+    FileSystem fs = basePath.getFileSystem(conf);
+
+    List<String> matches = new ArrayList<>();
+
+    if (fs.exists(new Path(basePath, "_delta_log"))) {
+      matches.add(TableFormat.DELTA);
+    }
+
+    if (fs.exists(new Path(basePath, ".hoodie"))) {
+      matches.add(TableFormat.HUDI);
+    }
+    try {
+      HadoopTables tables = new HadoopTables(conf);
+      org.apache.iceberg.Table table = tables.load(pathStr);
+      if (table != null) {
+        matches.add(TableFormat.ICEBERG);
+      }
+    } catch (NoSuchTableException e) {
+      log.debug("No Iceberg table found at path: {}", pathStr);
+    } catch (Exception e) {
+      log.debug("Unexpected error while probing for Iceberg table at path: {}", pathStr, e);
+    }
+    if (matches.size() == 1) {
+      return matches.get(0);
+    }
+
+    if (matches.size() > 1) {
+      log.info(
+          "Multiple formats detected: {}. Resolving target sync vs original source...", matches);
+      return inferSourceFromSyncMetadata(basePath, fs, matches, conf);
+    }
+    throw new IllegalArgumentException("Unable to detect table format for path: " + pathStr);
+  }
+
+  // checks source format from a XTable target
+  private static String inferSourceFromSyncMetadata(
+      Path basePath, FileSystem fs, List<String> detectedFormats, Configuration conf) {
+    try {
+      // check Hudi metadata if present
+      if (detectedFormats.contains(TableFormat.HUDI)) {
+        // check XTable target for hudi property file
+        Path hoodieMeta = new Path(basePath, ".hoodie");
+        if (fs.exists(new Path(hoodieMeta, "hoodie.properties"))) {
+          return TableFormat.HUDI;
+        }
+      }
+
+      if (detectedFormats.contains(TableFormat.ICEBERG)) {
+        HadoopTables tables = new HadoopTables(conf);
+        org.apache.iceberg.Table table = tables.load(basePath.toString());
+        // check for target property tag during a sync run
+        if (table.properties().containsKey("xtable.conversion.target")) {
+          detectedFormats.remove(TableFormat.ICEBERG);
+          // remove Hudi if it was detected as potential source, as it is not the source
+          if (detectedFormats.contains(TableFormat.HUDI)) {
+            detectedFormats.remove(TableFormat.HUDI);
+          }
+          if (detectedFormats.size() == 1) {
+            return detectedFormats.get(0);
+          }
+        }
+      }
+      if (detectedFormats.contains(TableFormat.DELTA)) {
+        try {
+
+          Engine kernelEngine = DefaultEngine.create(conf);
+
+          io.delta.kernel.Table table =
+              io.delta.kernel.Table.forPath(kernelEngine, basePath.toString());
+          Snapshot snapshot = table.getLatestSnapshot(kernelEngine);
+
+          if (snapshot instanceof SnapshotImpl) {
+            Map<String, String> config = ((SnapshotImpl) snapshot).getMetadata().getConfiguration();
+
+            if (config != null && config.containsKey("xtable.conversion.target")) {
+              detectedFormats.remove(TableFormat.DELTA);
+              // remove Hudi if it was detected as potential source, as it is not the source
+              if (detectedFormats.contains(TableFormat.HUDI)) {
+                detectedFormats.remove(TableFormat.HUDI);
+              }
+              if (detectedFormats.size() == 1) {
+                return detectedFormats.get(0);
+              }
+            }
+          }
+        } catch (Exception e) {
+          log.debug(
+              "Failed to verify Delta log metadata via internal XTable engine at path: {}",
+              basePath,
+              e);
+        }
+      }
+    } catch (Exception e) {
+      log.warn("Failed to parse table metadata properties during conflict resolution step", e);
+    }
+
+    return null;
+  }
+}

xtable-api/src/main/java/org/apache/xtable/conversion/TargetTable.java

@@ -26,6 +26,8 @@
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
 
+import org.apache.hadoop.conf.Configuration;
+
 @Getter
 @EqualsAndHashCode(callSuper = true)
 public class TargetTable extends ExternalTable {
@@ -39,8 +41,16 @@ public TargetTable(
       String[] namespace,
       CatalogConfig catalogConfig,
       Duration metadataRetention,
-      Properties additionalProperties) {
-    super(name, formatName, basePath, namespace, catalogConfig, additionalProperties);
+      Properties additionalProperties,
+      Configuration hadoopConf) {
+    super(
+        name,
+        formatName,
+        basePath,
+        namespace,
+        catalogConfig,
+        additionalProperties,
+        hadoopConf != null ? hadoopConf : new Configuration());
     this.metadataRetention =
         metadataRetention == null ? Duration.of(7, ChronoUnit.DAYS) : metadataRetention;
   }

xtable-api/src/test/java/org/apache/xtable/conversion/TestExternalTable.java

@@ -21,36 +21,39 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
+import org.apache.hadoop.conf.Configuration;
 import org.junit.jupiter.api.Test;
 
 public class TestExternalTable {
+  Configuration hadoopConf = new Configuration();
+
   @Test
   void sanitizePath() {
     ExternalTable tooManySlashes =
-        new ExternalTable("name", "hudi", "s3://bucket//path", null, null, null);
+        new ExternalTable("name", "hudi", "s3://bucket//path", null, null, null, hadoopConf);
     assertEquals("s3://bucket/path", tooManySlashes.getBasePath());
 
     ExternalTable localFilePath =
-        new ExternalTable("name", "hudi", "/local/data//path", null, null, null);
+        new ExternalTable("name", "hudi", "/local/data//path", null, null, null, hadoopConf);
     assertEquals("file:///local/data/path", localFilePath.getBasePath());
 
     ExternalTable properLocalFilePath =
-        new ExternalTable("name", "hudi", "file:///local/data//path", null, null, null);
+        new ExternalTable("name", "hudi", "file:///local/data//path", null, null, null, hadoopConf);
     assertEquals("file:///local/data/path", properLocalFilePath.getBasePath());
   }
 
   @Test
   void errorIfRequiredArgsNotSet() {
     assertThrows(
         NullPointerException.class,
-        () -> new ExternalTable("name", "hudi", null, null, null, null));
+        () -> new ExternalTable("name", "hudi", null, null, null, null, hadoopConf));
 
     assertThrows(
         NullPointerException.class,
-        () -> new ExternalTable("name", null, "file://bucket/path", null, null, null));
+        () -> new ExternalTable("name", null, "file://bucket/path", null, null, null, hadoopConf));
 
     assertThrows(
         NullPointerException.class,
-        () -> new ExternalTable(null, "hudi", "file://bucket/path", null, null, null));
+        () -> new ExternalTable(null, "hudi", "file://bucket/path", null, null, null, hadoopConf));
   }
 }

xtable-core/src/main/java/org/apache/xtable/conversion/ConversionUtils.java

@@ -28,6 +28,7 @@ public static SourceTable convertToSourceTable(TargetTable table) {
         table.getBasePath(),
         table.getNamespace(),
         table.getCatalogConfig(),
-        table.getAdditionalProperties());
+        table.getAdditionalProperties(),
+        table.hadoopConf);
   }
 }