Made code improvements

Author: Sreeja Chintalapati

hadoop-ozone/iceberg/src/main/java/org/apache/hadoop/ozone/iceberg/RewriteTablePathOzoneAction.java

@@ -30,6 +30,7 @@
 import java.util.concurrent.TimeUnit;
 import org.apache.iceberg.BaseTable;
 import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.PartitionStatisticsFile;
 import org.apache.iceberg.RewriteTablePathUtil;
 import org.apache.iceberg.RewriteTablePathUtil.RewriteResult;
 import org.apache.iceberg.Snapshot;
@@ -144,8 +145,9 @@ private void validateInputs() {
               "Source prefix cannot be the same as target prefix (%s)", sourcePrefix));
     }
 
-    validateAndSetEndVersion();
-    validateAndSetStartVersion();
+    TableMetadata tableMetadata = ((HasTableOperations) table).operations().current();
+    validateAndSetEndVersion(tableMetadata);
+    validateAndSetStartVersion(tableMetadata);
 
     if (stagingDir == null) {
       stagingDir =
@@ -158,9 +160,7 @@ private void validateInputs() {
     }
   }
 
-  private void validateAndSetEndVersion() {
-    TableMetadata tableMetadata = ((HasTableOperations) table).operations().current();
-
+  private void validateAndSetEndVersion(TableMetadata tableMetadata) {
     if (endVersionName == null) {
       Objects.requireNonNull(
           tableMetadata.metadataFileLocation(), "Metadata file location should not be null");
@@ -170,9 +170,7 @@ private void validateAndSetEndVersion() {
     }
   }
 
-  private void validateAndSetStartVersion() {
-    TableMetadata tableMetadata = ((HasTableOperations) table).operations().current();
-
+  private void validateAndSetStartVersion(TableMetadata tableMetadata) {
     if (startVersionName != null) {
       this.startVersionName = validateVersion(tableMetadata, startVersionName);
     }
@@ -184,9 +182,12 @@ private String validateVersion(TableMetadata tableMetadata, String versionFileNa
       versionFile = tableMetadata.metadataFileLocation();
     }
 
-    for (MetadataLogEntry log : tableMetadata.previousFiles()) {
-      if (versionInFilePath(log.file(), versionFileName)) {
-        versionFile = log.file();
+    if (versionFile == null) {
+      for (MetadataLogEntry log : tableMetadata.previousFiles()) {
+        if (versionInFilePath(log.file(), versionFileName)) {
+          versionFile = log.file();
+          break;
+        }
       }
     }
 
@@ -216,8 +217,8 @@ private String rebuildMetadata() {
     TableMetadata endMetadata =
             ((HasTableOperations) newStaticTable(endVersionName, table.io())).operations().current();
 
-    if (endMetadata.partitionStatisticsFiles() != null
-            && !endMetadata.partitionStatisticsFiles().isEmpty()) {
+    List<PartitionStatisticsFile> partitionStats = endMetadata.partitionStatisticsFiles();
+    if (partitionStats != null && !partitionStats.isEmpty()) {
       throw new IllegalArgumentException("Partition statistics files are not supported yet.");
     }
 
@@ -301,7 +302,7 @@ private Set<Pair<String, String>> statsFileCopyPlan(List<StatisticsFile> beforeS
       StatisticsFile before = beforeStats.get(i);
       StatisticsFile after = afterStats.get(i);
       if (before.fileSizeInBytes() != after.fileSizeInBytes()) {
-        throw new IllegalArgumentException("Before and after path rewrite, statistic files count should be same");
+        throw new IllegalArgumentException("Before and after path rewrite, statistic files size should be same");
       }
       result.add(Pair.of(before.path(), after.path()));
     }

hadoop-ozone/iceberg/src/test/java/org/apache/hadoop/ozone/iceberg/TestRewriteTablePathOzoneAction.java

@@ -67,8 +67,6 @@ class TestRewriteTablePathOzoneAction {
   private String targetPrefix = null;
   private Table table = null;
 
-  private final HadoopTables tables = new HadoopTables(new Configuration());
-
   @TempDir
   private java.nio.file.Path tableDir;
   @TempDir
@@ -261,6 +259,7 @@ private static Set<Pair<String, String>> readCsvPairs(Table tbl, String fileList
   }
 
   private Table createTable(String location) {
+    HadoopTables tables = new HadoopTables(new Configuration());
     Table tbl = tables.create(SCHEMA, PartitionSpec.unpartitioned(), new HashMap<>(), location);
     for (int i = 0; i < COMMITS; i++) {
       String dataPath = location + "/data/batch-" + i + ".parquet";