review comments round 2

Author: ajantha-bhat

core/src/main/java/org/apache/iceberg/BaseMetastoreTableOperations.java

@@ -331,15 +331,24 @@ private String newTableMetadataFilePath(TableMetadata meta, int newVersion) {
     return metadataFileLocation(meta, String.format("%05d-%s%s", newVersion, UUID.randomUUID(), fileExtension));
   }
 
+  /**
+   * Parse the version from table metadata file name.
+   *
+   * @param metadataLocation table metadata file location
+   * @return version of the table metadata file in success case and
+   * -1 if the version is not parsable (as a sign that the metadata is not part of this catalog)
+   */
   private static int parseVersion(String metadataLocation) {
     int versionStart = metadataLocation.lastIndexOf('/') + 1; // if '/' isn't found, this will be 0
     int versionEnd = metadataLocation.indexOf('-', versionStart);
+    if (versionEnd < 0) {
+      LOG.warn("Found filesystem table's metadata : {}", metadataLocation);
+      return -1;
+    }
+
     try {
       return Integer.valueOf(metadataLocation.substring(versionStart, versionEnd));
-    } catch (NumberFormatException | StringIndexOutOfBoundsException e) {
-      // As per the spec, metastore tables and filesystem tables
-      // will have different representation for metadata file names.
-      // So, StringIndexOutOfBoundsException can happen when parsing a filesystem based metadata file name.
+    } catch (NumberFormatException e) {
       LOG.warn("Unable to parse version from metadata location: {}", metadataLocation, e);
       return -1;
     }

hive-metastore/src/test/java/org/apache/iceberg/hive/HiveTableTest.java

@@ -25,6 +25,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.stream.Collectors;
 import org.apache.avro.generic.GenericData;
 import org.apache.avro.generic.GenericRecordBuilder;
@@ -37,6 +38,7 @@
 import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
 import org.apache.hadoop.hive.serde.serdeConstants;
 import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.BaseTable;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DataFiles;
 import org.apache.iceberg.FileScanTask;
@@ -398,9 +400,10 @@ public void testRegisterHadoopTableToHiveCatalog() throws IOException, TExceptio
     TableIdentifier identifier = TableIdentifier.of(DB_NAME, "table1");
     Table table = hadoopCatalog.createTable(identifier, schema, PartitionSpec.unpartitioned(), Maps.newHashMap());
     // insert some data
-    appendData(table, "file1");
+    String file1Location = appendData(table, "file1");
     List<FileScanTask> tasks = Lists.newArrayList(table.newScan().planFiles());
     Assert.assertEquals("Should scan 1 file", 1, tasks.size());
+    Assert.assertEquals(tasks.get(0).file().path(), file1Location);
 
     // collect metadata file
     List<String> metadataFiles =
@@ -420,18 +423,20 @@ public void testRegisterHadoopTableToHiveCatalog() throws IOException, TExceptio
             () -> catalog.loadTable(identifier));
 
     // register the table to hive catalog using the latest metadata file
-    metadataFiles.sort(Collections.reverseOrder());
-    catalog.registerTable(identifier, "file:" + metadataFiles.get(0));
+    String latestMetadataFile = ((BaseTable) table).operations().current().metadataFileLocation();
+    catalog.registerTable(identifier, "file:" + latestMetadataFile);
     Assert.assertNotNull(metastoreClient.getTable(DB_NAME, "table1"));
 
     // load the table in hive catalog
     table = catalog.loadTable(identifier);
     Assert.assertNotNull(table);
 
     // insert some data
-    appendData(table, "file2");
+    String file2Location = appendData(table, "file2");
     tasks = Lists.newArrayList(table.newScan().planFiles());
     Assert.assertEquals("Should scan 2 files", 2, tasks.size());
+    Set<String> files = tasks.stream().map(task -> task.file().path().toString()).collect(Collectors.toSet());
+    Assert.assertTrue(files.contains(file1Location) && files.contains(file2Location));
   }
 
   private String appendData(Table table, String fileName) throws IOException {