[Bugfix] Populate table name from the identifier in Iceberge conversion

What is the problem?
While converting the iceberg table to other format such as Hudi,
the icerberg source table do not populate the table name. This is
due to iceberg table's behavior as it is treated as Hadoop tables.
This leads to table identified as table-location, leading to confusing
conversation.

Solution:
This commit handles the conversation logic, when icebege table manager
provides HadoopTable, it populate the table name from provided input
TableIdentifier. This ensures that source table name is carried over
to the transformation.

Testing:
- Added unit test to cover this scenario.

Author: Jalpan Randeri

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

@@ -29,6 +29,7 @@
 
 import org.apache.hadoop.conf.Configuration;
 
+import org.apache.iceberg.BaseTable;
 import org.apache.iceberg.CatalogUtil;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
@@ -55,7 +56,7 @@ Table getTable(
       IcebergCatalogConfig catalogConfig, TableIdentifier tableIdentifier, String basePath) {
     return getCatalog(catalogConfig)
         .map(catalog -> catalog.loadTable(tableIdentifier))
-        .orElseGet(() -> getHadoopTables().load(basePath));
+        .orElseGet(() -> loadHadoopTables(getHadoopTables(), basePath, tableIdentifier.name()));
   }
 
   boolean tableExists(
@@ -100,6 +101,14 @@ 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/TestIcebergTableManager.java

@@ -29,6 +29,8 @@
 import java.util.Map;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.hadoop.HadoopTables;
 import org.junit.jupiter.api.Test;
 
 import org.apache.iceberg.PartitionSpec;
@@ -67,6 +69,18 @@ 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";