review feedback - moved to iceberg source converter

Author: jalpan-randeri

xtable-core/src/main/java/org/apache/xtable/iceberg/IcebergConversionSource.java

@@ -35,6 +35,7 @@
 
 import org.apache.hadoop.conf.Configuration;
 
+import org.apache.iceberg.BaseTable;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.PartitionSpec;
@@ -93,10 +94,15 @@ private Table initSourceTable() {
         namespace == null
             ? TableIdentifier.of(tableName)
             : TableIdentifier.of(Namespace.of(namespace), tableName);
-    return tableManager.getTable(
+    Table table = tableManager.getTable(
         (IcebergCatalogConfig) sourceTableConfig.getCatalogConfig(),
         tableIdentifier,
         sourceTableConfig.getBasePath());
+    if (sourceTableConfig.getCatalogConfig() == null && table instanceof BaseTable) {
+      table = new BaseTable(((BaseTable) table).operations(), tableIdentifier.name());
+    }
+
+    return table;
   }
 
   private FileIO initTableOps() {

xtable-core/src/main/java/org/apache/xtable/iceberg/IcebergTableManager.java

@@ -56,7 +56,7 @@ Table getTable(
       IcebergCatalogConfig catalogConfig, TableIdentifier tableIdentifier, String basePath) {
     return getCatalog(catalogConfig)
         .map(catalog -> catalog.loadTable(tableIdentifier))
-        .orElseGet(() -> loadHadoopTables(getHadoopTables(), basePath, tableIdentifier.name()));
+        .orElseGet(() -> getHadoopTables().load(basePath));
   }
 
   boolean tableExists(
@@ -101,14 +101,6 @@ Table getOrCreateTable(
     }
   }
 
-  Table loadHadoopTables(HadoopTables hadoopTables, String basePath, String tableName) {
-    Table table = hadoopTables.load(basePath);
-    if (table instanceof BaseTable) {
-      return new BaseTable(((BaseTable) table).operations(), tableName);
-    }
-    return table;
-  }
-
   private Map<String, String> getDefaultMappingProperties(Schema schema) {
     return Collections.singletonMap(
         TableProperties.DEFAULT_NAME_MAPPING, NameMappingParser.toJson(MappingUtil.create(schema)));

xtable-core/src/test/java/org/apache/xtable/iceberg/ITIcebergConversionSource.java

@@ -126,7 +126,7 @@ void getCurrentTableTest() {
               .build();
       validateTable(
           internalTable,
-          testIcebergTable.getBasePath(),
+          testIcebergTable.getTableName(),
           TableFormat.ICEBERG,
           internalSchema,
           DataLayoutStrategy.FLAT,

xtable-core/src/test/java/org/apache/xtable/iceberg/TestIcebergTableManager.java

@@ -69,18 +69,6 @@ void catalogGetTable() {
     verify(mockCatalog).initialize(catalogName, OPTIONS);
   }
 
-  @Test
-  void catalogGetTableWithoutCatalogConfig() {
-    Table hdfsTable = new BaseTable(null, BASE_PATH);
-    IcebergTableManager tableManager = IcebergTableManager.of(CONFIGURATION);
-    HadoopTables hadoopTables = mock(HadoopTables.class);
-    when(hadoopTables.load(BASE_PATH)).thenReturn(hdfsTable);
-
-    Table expected = new BaseTable(null, IDENTIFIER.name());
-    Table actual = tableManager.loadHadoopTables(hadoopTables, BASE_PATH, IDENTIFIER.name());
-    assertEquals(expected.name(), actual.name());
-  }
-
   @Test
   void catalogGetOrCreateWithExistingTable() {
     String catalogName = "catalog2";