Address Vinish remaining comments

Author: rahil-c

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

@@ -18,6 +18,18 @@
 
 package org.apache.xtable.conversion;
 
+import static org.apache.xtable.model.storage.TableFormat.DELTA;
+import static org.apache.xtable.model.storage.TableFormat.HUDI;
+import static org.apache.xtable.model.storage.TableFormat.ICEBERG;
+
+import org.apache.hadoop.conf.Configuration;
+
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+
+import org.apache.xtable.delta.DeltaConversionSourceProvider;
+import org.apache.xtable.hudi.HudiConversionSourceProvider;
+import org.apache.xtable.iceberg.IcebergConversionSourceProvider;
+
 public class ConversionUtils {
 
   public static SourceTable convertToSourceTable(TargetTable table) {
@@ -30,4 +42,26 @@ public static SourceTable convertToSourceTable(TargetTable table) {
         table.getCatalogConfig(),
         table.getAdditionalProperties());
   }
+
+  public static ConversionSourceProvider<?> getConversionSourceProvider(
+      String sourceTableFormat, Configuration hadoopConf) {
+    if (sourceTableFormat.equalsIgnoreCase(HUDI)) {
+      ConversionSourceProvider<HoodieInstant> hudiConversionSourceProvider =
+          new HudiConversionSourceProvider();
+      hudiConversionSourceProvider.init(hadoopConf);
+      return hudiConversionSourceProvider;
+    } else if (sourceTableFormat.equalsIgnoreCase(DELTA)) {
+      ConversionSourceProvider<Long> deltaConversionSourceProvider =
+          new DeltaConversionSourceProvider();
+      deltaConversionSourceProvider.init(hadoopConf);
+      return deltaConversionSourceProvider;
+    } else if (sourceTableFormat.equalsIgnoreCase(ICEBERG)) {
+      ConversionSourceProvider<org.apache.iceberg.Snapshot> icebergConversionSourceProvider =
+          new IcebergConversionSourceProvider();
+      icebergConversionSourceProvider.init(hadoopConf);
+      return icebergConversionSourceProvider;
+    } else {
+      throw new IllegalArgumentException("Unsupported source format: " + sourceTableFormat);
+    }
+  }
 }

xtable-service/src/main/java/org/apache/xtable/service/ConversionService.java

@@ -18,27 +18,32 @@
 
 package org.apache.xtable.service;
 
-import static org.apache.xtable.model.storage.TableFormat.DELTA;
-import static org.apache.xtable.model.storage.TableFormat.HUDI;
-import static org.apache.xtable.model.storage.TableFormat.ICEBERG;
+import static org.apache.xtable.conversion.ConversionUtils.convertToSourceTable;
 
 import java.util.ArrayList;
 import java.util.List;
 
-import org.apache.commons.lang3.tuple.Pair;
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.conf.Configuration;
+
+import org.apache.iceberg.SchemaParser;
 
 import com.google.common.annotations.VisibleForTesting;
 
+import org.apache.xtable.avro.AvroSchemaConverter;
 import org.apache.xtable.conversion.ConversionConfig;
 import org.apache.xtable.conversion.ConversionController;
-import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.ConversionUtils;
 import org.apache.xtable.conversion.SourceTable;
 import org.apache.xtable.conversion.TargetTable;
+import org.apache.xtable.iceberg.IcebergSchemaExtractor;
+import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.schema.SparkSchemaExtractor;
 import org.apache.xtable.service.models.ConvertTableRequest;
 import org.apache.xtable.service.models.ConvertTableResponse;
 import org.apache.xtable.service.models.ConvertedTable;
-import org.apache.xtable.service.spark.SparkHolder;
-import org.apache.xtable.service.utils.ConversionServiceUtil;
 
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
@@ -50,40 +55,61 @@
  * a source table, generating target tables, and then executing the conversion via a designated
  * conversion controller.
  */
+@Log4j2
 @ApplicationScoped
 public class ConversionService {
-  private final SparkHolder sparkHolder;
   private final ConversionController conversionController;
-  private final ConversionServiceUtil conversionServiceUtil;
+  private final ConversionServiceConfig serviceConfig;
+  private final Configuration hadoopConf;
 
   /**
    * Constructs a ConversionService instance with required dependencies.
    *
-   * @param sparkHolder the Spark holder instance containing the Spark context and configuration
-   * @param conversionServiceUtil utility for handling metadata operations
+   * @param serviceConfig the conversion service configuration
    */
   @Inject
-  public ConversionService(SparkHolder sparkHolder, ConversionServiceUtil conversionServiceUtil) {
-    this.sparkHolder = sparkHolder;
-    this.conversionServiceUtil = conversionServiceUtil;
-    this.conversionController = new ConversionController(sparkHolder.jsc().hadoopConfiguration());
+  public ConversionService(ConversionServiceConfig serviceConfig) {
+    this.serviceConfig = serviceConfig;
+    this.hadoopConf = getHadoopConf();
+    this.conversionController = new ConversionController(hadoopConf);
+  }
+
+  private Configuration getHadoopConf() {
+    Configuration conf = new Configuration();
+    String hadoopConfigPath = serviceConfig.getHadoopConfigPath();
+    try {
+      // Load configuration from the specified XML file
+      conf.addResource(hadoopConfigPath);
+
+      // If the resource wasn’t found, log a warning
+      if (conf.size() == 0) {
+        log.warn(
+            "Could not load Hadoop configuration from: {}. Using default Hadoop configuration.",
+            hadoopConfigPath);
+      }
+    } catch (Exception e) {
+      log.error(
+          "Error loading Hadoop configuration from: {}. Exception: {}",
+          hadoopConfigPath,
+          e.getMessage(),
+          e);
+    }
+    return conf;
   }
 
   /**
    * Constructs a ConversionService instance using dependency injection for testing.
    *
-   * @param sparkHolder the Spark holder instance
    * @param conversionController a preconfigured conversion controller
-   * @param conversionServiceUtil utility for handling metadata operations
    */
   @VisibleForTesting
   public ConversionService(
-      SparkHolder sparkHolder,
-      ConversionServiceUtil conversionServiceUtil,
-      ConversionController conversionController) {
-    this.sparkHolder = sparkHolder;
+      ConversionServiceConfig serviceConfig,
+      ConversionController conversionController,
+      Configuration hadoopConf) {
+    this.serviceConfig = serviceConfig;
     this.conversionController = conversionController;
-    this.conversionServiceUtil = conversionServiceUtil;
+    this.hadoopConf = hadoopConf;
   }
 
   /**
@@ -120,37 +146,47 @@ public ConvertTableResponse convertTable(ConvertTableRequest convertTableRequest
     ConversionConfig conversionConfig =
         ConversionConfig.builder().sourceTable(sourceTable).targetTables(targetTables).build();
 
-    ConversionSourceProvider<?> conversionSourceProvider =
-        conversionServiceUtil.getConversionSourceProvider(
-            convertTableRequest.getSourceFormat(), sparkHolder.jsc().hadoopConfiguration());
+    conversionController.sync(
+        conversionConfig,
+        ConversionUtils.getConversionSourceProvider(
+            convertTableRequest.getSourceFormat(), hadoopConf));
 
-    conversionController.sync(conversionConfig, conversionSourceProvider);
-
-    List<ConvertedTable> restTargetTables = new ArrayList<>();
-    for (String targetFormat : convertTableRequest.getTargetFormats()) {
-      if (targetFormat.equals(ICEBERG)) {
-        Pair<String, String> responseFields =
-            conversionServiceUtil.getIcebergSchemaAndMetadataPath(
-                convertTableRequest.getSourceTablePath(), sparkHolder.jsc().hadoopConfiguration());
-        ConvertedTable icebergTable =
-            new ConvertedTable(ICEBERG, responseFields.getLeft(), responseFields.getRight());
-        restTargetTables.add(icebergTable);
-      } else if (targetFormat.equals(HUDI)) {
-        Pair<String, String> responseFields =
-            conversionServiceUtil.getHudiSchemaAndMetadataPath(
-                convertTableRequest.getSourceTablePath(), sparkHolder.jsc().hadoopConfiguration());
-        ConvertedTable hudiTable =
-            new ConvertedTable(HUDI, responseFields.getLeft(), responseFields.getRight());
-        restTargetTables.add(hudiTable);
-      } else if (targetFormat.equals(DELTA)) {
-        Pair<String, String> responseFields =
-            conversionServiceUtil.getDeltaSchemaAndMetadataPath(
-                convertTableRequest.getSourceTablePath(), sparkHolder.spark());
-        ConvertedTable deltaTable =
-            new ConvertedTable(DELTA, responseFields.getLeft(), responseFields.getRight());
-        restTargetTables.add(deltaTable);
+    List<ConvertedTable> convertedTables = new ArrayList<>();
+    for (TargetTable targetTable : targetTables) {
+      InternalTable internalTable =
+          ConversionUtils.getConversionSourceProvider(targetTable.getFormatName(), hadoopConf)
+              .getConversionSourceInstance(convertToSourceTable(targetTable))
+              .getCurrentTable();
+      String schemaString;
+      switch (targetTable.getFormatName()) {
+        case TableFormat.HUDI:
+          schemaString =
+              AvroSchemaConverter.getInstance()
+                  .fromInternalSchema(internalTable.getReadSchema())
+                  .toString();
+          break;
+        case TableFormat.ICEBERG:
+          org.apache.iceberg.Schema iceSchema =
+              IcebergSchemaExtractor.getInstance().toIceberg(internalTable.getReadSchema());
+          schemaString = SchemaParser.toJson(iceSchema);
+          break;
+        case TableFormat.DELTA:
+          schemaString =
+              SparkSchemaExtractor.getInstance()
+                  .fromInternalSchema(internalTable.getReadSchema())
+                  .json();
+          break;
+        default:
+          throw new UnsupportedOperationException(
+              "Unsupported table format: " + targetTable.getFormatName());
       }
+      convertedTables.add(
+          ConvertedTable.builder()
+              .targetFormat(internalTable.getName())
+              .targetSchema(schemaString)
+              .targetMetadataPath(internalTable.getLatestMetdataPath())
+              .build());
     }
-    return new ConvertTableResponse(restTargetTables);
+    return new ConvertTableResponse(convertedTables);
   }
 }