[iceberg] minor improvements (apache#2015)

Author: nastra

fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergLakeCommitter.java

@@ -212,13 +212,13 @@ private long commit(SnapshotUpdate<?> snapshotUpdate, Map<String, String> snapsh
     public void abort(IcebergCommittable committable) {
         List<String> dataFilesToDelete =
                 committable.getDataFiles().stream()
-                        .map(file -> file.path().toString())
+                        .map(ContentFile::location)
                         .collect(Collectors.toList());
         CatalogUtil.deleteFiles(icebergTable.io(), dataFilesToDelete, "data file", true);
 
         List<String> deleteFilesToDelete =
                 committable.getDeleteFiles().stream()
-                        .map(file -> file.path().toString())
+                        .map(ContentFile::location)
                         .collect(Collectors.toList());
         CatalogUtil.deleteFiles(icebergTable.io(), deleteFilesToDelete, "delete file", true);
     }

fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/bucketing/IcebergBucketingFunctionTest.java

@@ -31,7 +31,6 @@
 import org.apache.iceberg.types.Types;
 import org.junit.jupiter.api.Test;
 
-import java.io.IOException;
 import java.math.BigDecimal;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
@@ -44,7 +43,7 @@
 class IcebergBucketingFunctionTest {
 
     @Test
-    void testIntegerHash() throws IOException {
+    void testIntegerHash() {
         int testValue = 42;
         int bucketNum = 10;
 
@@ -69,7 +68,7 @@ void testIntegerHash() throws IOException {
     }
 
     @Test
-    void testLongHash() throws IOException {
+    void testLongHash() {
         long testValue = 1234567890123456789L;
         int bucketNum = 10;
 
@@ -94,7 +93,7 @@ void testLongHash() throws IOException {
     }
 
     @Test
-    void testStringHash() throws IOException {
+    void testStringHash() {
         String testValue = "Hello Iceberg, Fluss this side!";
         int bucketNum = 10;
 
@@ -120,7 +119,7 @@ void testStringHash() throws IOException {
     }
 
     @Test
-    void testDecimalHash() throws IOException {
+    void testDecimalHash() {
         BigDecimal testValue = new BigDecimal("123.45");
         Decimal decimal = Decimal.fromBigDecimal(testValue, 10, 2);
         int bucketNum = 10;
@@ -148,7 +147,7 @@ void testDecimalHash() throws IOException {
     }
 
     @Test
-    void testTimestampEncodingHash() throws IOException {
+    void testTimestampEncodingHash() {
         // Iceberg expects microseconds for TIMESTAMP type
         long millis = 1698235273182L;
         int nanos = 123456;
@@ -179,7 +178,7 @@ void testTimestampEncodingHash() throws IOException {
     }
 
     @Test
-    void testDateHash() throws IOException {
+    void testDateHash() {
         int dateValue = 19655;
         int bucketNum = 10;
 
@@ -204,7 +203,7 @@ void testDateHash() throws IOException {
     }
 
     @Test
-    void testTimeHashing() throws IOException {
+    void testTimeHashing() {
         // Fluss stores time as int (milliseconds since midnight)
         int timeMillis = 34200000;
         long timeMicros = timeMillis * 1000L; // Convert to microseconds for Iceberg
@@ -231,7 +230,7 @@ void testTimeHashing() throws IOException {
     }
 
     @Test
-    void testBinaryEncoding() throws IOException {
+    void testBinaryEncoding() {
         byte[] testValue = "Hello i only understand binary data".getBytes();
         int bucketNum = 10;
 

fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/source/IcebergRecordReaderTest.java

@@ -153,16 +153,16 @@ void testReadLogTable(boolean isPartitioned) throws Exception {
         List<Row> projectExpect = new ArrayList<>();
         try (CloseableIterable<FileScanTask> fileScanTasks = tableScan.planFiles()) {
             for (FileScanTask task : fileScanTasks) {
-                org.apache.iceberg.io.CloseableIterator<Record> iterator =
-                        reader.open(task).iterator();
-                IcebergRecordAsFlussRow recordAsFlussRow = new IcebergRecordAsFlussRow();
-                projectExpect.addAll(
-                        convertToFlinkRow(
-                                projectFieldGetters,
-                                TransformingCloseableIterator.transform(
-                                        CloseableIterator.wrap(iterator),
-                                        recordAsFlussRow::replaceIcebergRecord)));
-                iterator.close();
+                try (org.apache.iceberg.io.CloseableIterator<Record> iterator =
+                        reader.open(task).iterator()) {
+                    IcebergRecordAsFlussRow recordAsFlussRow = new IcebergRecordAsFlussRow();
+                    projectExpect.addAll(
+                            convertToFlinkRow(
+                                    projectFieldGetters,
+                                    TransformingCloseableIterator.transform(
+                                            CloseableIterator.wrap(iterator),
+                                            recordAsFlussRow::replaceIcebergRecord)));
+                }
             }
         }
         assertThat(projectActual).containsExactlyInAnyOrderElementsOf(projectExpect);

fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/testutils/FlinkIcebergTieringTestBase.java

@@ -437,7 +437,7 @@ private CloseableIterator<Record> getIcebergRows(
 
             for (DataFile file : files) {
                 Iterable<Record> iterable =
-                        Parquet.read(table.io().newInputFile(file.path().toString()))
+                        Parquet.read(table.io().newInputFile(file.location()))
                                 .project(table.schema())
                                 .createReaderFunc(
                                         fileSchema ->
@@ -471,7 +471,7 @@ private TableScan filterByPartition(TableScan tableScan, Map<String, String> par
     }
 
     protected void checkSnapshotPropertyInIceberg(
-            TablePath tablePath, Map<String, String> expectedProperties) throws Exception {
+            TablePath tablePath, Map<String, String> expectedProperties) {
         org.apache.iceberg.Table table = icebergCatalog.loadTable(toIceberg(tablePath));
         Snapshot snapshot = table.currentSnapshot();
         assertThat(snapshot.summary()).containsAllEntriesOf(expectedProperties);