return iceberg metadata fields

Author: rahil-c

xtable-service/README.md

@@ -24,11 +24,14 @@ See XTable's `spec` module for more details: https://github.com/apache/incubator
 
 ## How to run the service locally
 
+#### Before running the service, ensure that you have the required credentials set in your enviroment needed to read and write to cloud storage.
+
 To run the service locally, first ensure you have built the project with 
 ```sh
 mvn clean install -DskipTests
 ```
 
+
 Then you can run start the quarkus service using the following command:
 ```sh
 mvn quarkus:dev -pl xtable-service

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

@@ -26,10 +26,17 @@
 import java.util.Collections;
 import java.util.List;
 
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hudi.common.table.timeline.HoodieInstant;
 
+import org.apache.iceberg.BaseTable;
 import org.apache.iceberg.Snapshot;
 
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.hadoop.HadoopTables;
 import org.apache.xtable.conversion.ConversionConfig;
 import org.apache.xtable.conversion.ConversionController;
 import org.apache.xtable.conversion.ConversionSourceProvider;
@@ -79,9 +86,13 @@ public ConvertTableResponse runSync(ConvertTableRequest request) {
         getConversionSourceProvider(request.getSourceFormat());
     conversionController.sync(conversionConfig, conversionSourceProvider);
 
-    InternalTable internalTable = new InternalTable("ICEBERG", request.getSourceTablePath() + "/metadata", null);
-    ConvertTableResponse response = new ConvertTableResponse(Collections.singletonList(internalTable));
-    return response;
+    Pair<String, String> responseFields = getIcebergSchemaAndMetadataPath(request.getSourceTablePath(), sparkHolder.jsc().hadoopConfiguration());
+
+    InternalTable internalTable =
+        new InternalTable(
+            "ICEBERG",
+                responseFields.getLeft(), responseFields.getRight());
+      return new ConvertTableResponse(Collections.singletonList(internalTable));
   }
 
   private ConversionSourceProvider<?> getConversionSourceProvider(String sourceTableFormat) {
@@ -104,4 +115,13 @@ private ConversionSourceProvider<?> getConversionSourceProvider(String sourceTab
       throw new IllegalArgumentException("Unsupported source format: " + sourceTableFormat);
     }
   }
+
+  public static Pair<String, String> getIcebergSchemaAndMetadataPath(String tableLocation, Configuration conf) {
+    HadoopTables tables = new HadoopTables(conf);
+    Table table = tables.load(tableLocation);
+    TableOperations ops = ((BaseTable) table).operations();
+    TableMetadata current = ops.current();
+    return Pair.of(current.metadataFileLocation(), current.schema().toString());
+  }
+
 }