Core, Flink, Spark, KafkaConnect: Remove usage of deprecated path API (…

…apache#11744)

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

@@ -850,7 +850,7 @@ public FileContent content() {
 
     @Override
     public CharSequence path() {
-      return deleteFile.path();
+      return deleteFile.location();
     }
 
     @Override

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

@@ -416,7 +416,7 @@ public FileContent content() {
 
     @Override
     public CharSequence path() {
-      return wrapped.path();
+      return wrapped.location();
     }
 
     @Override

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

@@ -486,7 +486,7 @@ public FileContent content() {
 
     @Override
     public CharSequence path() {
-      return wrapped.path();
+      return wrapped.location();
     }
 
     @Override

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

@@ -500,7 +500,7 @@ public FileContent content() {
 
     @Override
     public CharSequence path() {
-      return wrapped.path();
+      return wrapped.location();
     }
 
     @Override

core/src/test/java/org/apache/iceberg/TestRewriteManifests.java

@@ -164,7 +164,7 @@ public void testReplaceManifestsSeparate() {
 
     // cluster by path will split the manifest into two
 
-    table.rewriteManifests().clusterBy(file -> file.location()).commit();
+    table.rewriteManifests().clusterBy(ContentFile::location).commit();
 
     List<ManifestFile> manifests = table.currentSnapshot().allManifests(table.io());
     assertThat(manifests).hasSize(2);

data/src/test/java/org/apache/iceberg/data/TestDataFileIndexStatsFilters.java

@@ -451,7 +451,7 @@ private void assertDeletes(FileScanTask task, DeleteFile... expectedDeleteFiles)
   }
 
   private CharSequenceSet deletePaths(FileScanTask task) {
-    return CharSequenceSet.of(Iterables.transform(task.deletes(), ContentFile::path));
+    return CharSequenceSet.of(Iterables.transform(task.deletes(), ContentFile::location));
   }
 
   private List<FileScanTask> planTasks() throws IOException {

data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java

@@ -178,7 +178,7 @@ public void testInsertDuplicatedKey() throws IOException {
 
     // Check records in the data file.
     DataFile dataFile = result.dataFiles()[0];
-    assertThat(readRecordsAsList(table.schema(), dataFile.path()))
+    assertThat(readRecordsAsList(table.schema(), dataFile.location()))
         .isEqualTo(
             ImmutableList.of(
                 createRecord(1, "aaa"),
@@ -192,13 +192,13 @@ public void testInsertDuplicatedKey() throws IOException {
 
     // Check records in the pos-delete file.
     Schema posDeleteSchema = DeleteSchemaUtil.pathPosSchema();
-    assertThat(readRecordsAsList(posDeleteSchema, posDeleteFile.path()))
+    assertThat(readRecordsAsList(posDeleteSchema, posDeleteFile.location()))
         .isEqualTo(
             ImmutableList.of(
-                posRecord.copy("file_path", dataFile.path(), "pos", 0L),
-                posRecord.copy("file_path", dataFile.path(), "pos", 1L),
-                posRecord.copy("file_path", dataFile.path(), "pos", 2L),
-                posRecord.copy("file_path", dataFile.path(), "pos", 3L)));
+                posRecord.copy("file_path", dataFile.location(), "pos", 0L),
+                posRecord.copy("file_path", dataFile.location(), "pos", 1L),
+                posRecord.copy("file_path", dataFile.location(), "pos", 2L),
+                posRecord.copy("file_path", dataFile.location(), "pos", 3L)));
   }
 
   @TestTemplate
@@ -226,13 +226,13 @@ public void testUpsertSameRow() throws IOException {
 
     // Check records in the data file.
     DataFile dataFile = result.dataFiles()[0];
-    assertThat(readRecordsAsList(table.schema(), dataFile.path()))
+    assertThat(readRecordsAsList(table.schema(), dataFile.location()))
         .isEqualTo(ImmutableList.of(record, record));
 
     // Check records in the pos-delete file.
     DeleteFile posDeleteFile = result.deleteFiles()[0];
-    assertThat(readRecordsAsList(DeleteSchemaUtil.pathPosSchema(), posDeleteFile.path()))
-        .isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
+    assertThat(readRecordsAsList(DeleteSchemaUtil.pathPosSchema(), posDeleteFile.location()))
+        .isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.location(), "pos", 0L)));
 
     deltaWriter =
         createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema, DeleteGranularity.PARTITION);
@@ -312,24 +312,24 @@ public void testUpsertData() throws IOException {
 
     // Check records in the data file.
     DataFile dataFile = result.dataFiles()[0];
-    assertThat(readRecordsAsList(table.schema(), dataFile.path()))
+    assertThat(readRecordsAsList(table.schema(), dataFile.location()))
         .isEqualTo(
             ImmutableList.of(
                 createRecord(5, "aaa"), createRecord(6, "aaa"), createRecord(7, "ccc")));
 
     // Check records in the eq-delete file.
     DeleteFile eqDeleteFile = result.deleteFiles()[0];
     assertThat(eqDeleteFile.content()).isEqualTo(FileContent.EQUALITY_DELETES);
-    assertThat(readRecordsAsList(eqDeleteRowSchema, eqDeleteFile.path()))
+    assertThat(readRecordsAsList(eqDeleteRowSchema, eqDeleteFile.location()))
         .isEqualTo(
             ImmutableList.of(keyFunc.apply("aaa"), keyFunc.apply("ccc"), keyFunc.apply("bbb")));
 
     // Check records in the pos-delete file.
     DeleteFile posDeleteFile = result.deleteFiles()[1];
     Schema posDeleteSchema = DeleteSchemaUtil.pathPosSchema();
     assertThat(posDeleteFile.content()).isEqualTo(FileContent.POSITION_DELETES);
-    assertThat(readRecordsAsList(posDeleteSchema, posDeleteFile.path()))
-        .isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
+    assertThat(readRecordsAsList(posDeleteSchema, posDeleteFile.location()))
+        .isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.location(), "pos", 0L)));
   }
 
   @TestTemplate
@@ -397,15 +397,15 @@ public void testUpsertDataWithFullRowSchema() throws IOException {
 
     // Check records in the data file.
     DataFile dataFile = result.dataFiles()[0];
-    assertThat(readRecordsAsList(table.schema(), dataFile.path()))
+    assertThat(readRecordsAsList(table.schema(), dataFile.location()))
         .isEqualTo(
             ImmutableList.of(
                 createRecord(5, "aaa"), createRecord(6, "aaa"), createRecord(7, "ccc")));
 
     // Check records in the eq-delete file.
     DeleteFile eqDeleteFile = result.deleteFiles()[0];
     assertThat(eqDeleteFile.content()).isEqualTo(FileContent.EQUALITY_DELETES);
-    assertThat(readRecordsAsList(eqDeleteRowSchema, eqDeleteFile.path()))
+    assertThat(readRecordsAsList(eqDeleteRowSchema, eqDeleteFile.location()))
         .isEqualTo(
             ImmutableList.of(
                 createRecord(3, "aaa"), createRecord(4, "ccc"), createRecord(2, "bbb")));
@@ -414,8 +414,8 @@ public void testUpsertDataWithFullRowSchema() throws IOException {
     DeleteFile posDeleteFile = result.deleteFiles()[1];
     Schema posDeleteSchema = DeleteSchemaUtil.pathPosSchema();
     assertThat(posDeleteFile.content()).isEqualTo(FileContent.POSITION_DELETES);
-    assertThat(readRecordsAsList(posDeleteSchema, posDeleteFile.path()))
-        .isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
+    assertThat(readRecordsAsList(posDeleteSchema, posDeleteFile.location()))
+        .isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.location(), "pos", 0L)));
   }
 
   @TestTemplate

flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/RowDataFileScanTaskReader.java

@@ -220,7 +220,7 @@ private static class FlinkDeleteFilter extends DeleteFilter<RowData> {
         Schema tableSchema,
         Schema requestedSchema,
         InputFilesDecryptor inputFilesDecryptor) {
-      super(task.file().path().toString(), task.deletes(), tableSchema, requestedSchema);
+      super(task.file().location(), task.deletes(), tableSchema, requestedSchema);
       this.requiredRowType = FlinkSchemaUtil.convert(requiredSchema());
       this.asStructLike = new RowDataWrapper(requiredRowType, requiredSchema().asStruct());
       this.inputFilesDecryptor = inputFilesDecryptor;

flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/split/IcebergSourceSplit.java

@@ -107,7 +107,7 @@ private String toString(Collection<FileScanTask> files) {
             .map(
                 fileScanTask ->
                     MoreObjects.toStringHelper(fileScanTask)
-                        .add("file", fileScanTask.file().path().toString())
+                        .add("file", fileScanTask.file().location())
                         .add("start", fileScanTask.start())
                         .add("length", fileScanTask.length())
                         .toString())

flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestBucketPartitionerFlinkIcebergSink.java

@@ -184,7 +184,7 @@ private TableTestStats extractPartitionResults(TableSchemaType tableSchemaType)
       for (FileScanTask scanTask : fileScanTasks) {
         long recordCountInFile = scanTask.file().recordCount();
 
-        String[] splitFilePath = scanTask.file().path().toString().split("/");
+        String[] splitFilePath = scanTask.file().location().split("/");
         // Filename example: 00007-0-a7d3a29a-33e9-4740-88f4-0f494397d60c-00001.parquet
         // Writer ID: .......^^^^^
         String filename = splitFilePath[splitFilePath.length - 1];

flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java

@@ -241,6 +241,6 @@ public void testDeleteStats() throws Exception {
         new String(
             deleteFile.lowerBounds().get(MetadataColumns.DELETE_FILE_PATH.fieldId()).array());
     DataFile dataFile = table.currentSnapshot().addedDataFiles(table.io()).iterator().next();
-    assumeThat(fromStat).isEqualTo(dataFile.path().toString());
+    assumeThat(fromStat).isEqualTo(dataFile.location());
   }
 }

flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestTaskWriters.java

@@ -114,7 +114,7 @@ public void testCloseTwice() throws IOException {
 
       FileSystem fs = FileSystem.get(CONF);
       for (DataFile dataFile : dataFiles) {
-        assertThat(fs.exists(new Path(dataFile.path().toString()))).isTrue();
+        assertThat(fs.exists(new Path(dataFile.location()))).isTrue();
       }
     }
   }
@@ -133,7 +133,7 @@ public void testAbort() throws IOException {
 
       FileSystem fs = FileSystem.get(CONF);
       for (DataFile dataFile : dataFiles) {
-        assertThat(fs.exists(new Path(dataFile.path().toString()))).isFalse();
+        assertThat(fs.exists(new Path(dataFile.location()))).isFalse();
       }
     }
   }
@@ -155,7 +155,7 @@ public void testCompleteFiles() throws IOException {
 
       FileSystem fs = FileSystem.get(CONF);
       for (DataFile dataFile : dataFiles) {
-        assertThat(fs.exists(new Path(dataFile.path().toString()))).isTrue();
+        assertThat(fs.exists(new Path(dataFile.location()))).isTrue();
       }
 
       AppendFiles appendFiles = table.newAppend();

flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/source/enumerator/TestContinuousSplitPlannerImpl.java

@@ -161,10 +161,9 @@ public void testTableScanThenIncrementalWithNonEmptyTable() throws Exception {
     assertThat(split.task().files()).hasSize(2);
     Set<String> discoveredFiles =
         split.task().files().stream()
-            .map(fileScanTask -> fileScanTask.file().path().toString())
+            .map(fileScanTask -> fileScanTask.file().location())
             .collect(Collectors.toSet());
-    Set<String> expectedFiles =
-        ImmutableSet.of(dataFile1.path().toString(), dataFile2.path().toString());
+    Set<String> expectedFiles = ImmutableSet.of(dataFile1.location(), dataFile2.location());
     assertThat(discoveredFiles).containsExactlyInAnyOrderElementsOf(expectedFiles);
 
     IcebergEnumeratorPosition lastPosition = initialResult.toPosition();
@@ -244,10 +243,10 @@ public void testIncrementalFromLatestSnapshotWithNonEmptyTable() throws Exceptio
     assertThat(split.task().files()).hasSize(1);
     Set<String> discoveredFiles =
         split.task().files().stream()
-            .map(fileScanTask -> fileScanTask.file().path().toString())
+            .map(fileScanTask -> fileScanTask.file().location())
             .collect(Collectors.toSet());
     // should discover dataFile2 appended in snapshot2
-    Set<String> expectedFiles = ImmutableSet.of(dataFile2.path().toString());
+    Set<String> expectedFiles = ImmutableSet.of(dataFile2.location());
     assertThat(discoveredFiles).containsExactlyElementsOf(expectedFiles);
 
     IcebergEnumeratorPosition lastPosition = secondResult.toPosition();
@@ -316,11 +315,10 @@ public void testIncrementalFromEarliestSnapshotWithNonEmptyTable() throws Except
     assertThat(split.task().files()).hasSize(2);
     Set<String> discoveredFiles =
         split.task().files().stream()
-            .map(fileScanTask -> fileScanTask.file().path().toString())
+            .map(fileScanTask -> fileScanTask.file().location())
             .collect(Collectors.toSet());
     // should discover files appended in both snapshot1 and snapshot2
-    Set<String> expectedFiles =
-        ImmutableSet.of(dataFile1.path().toString(), dataFile2.path().toString());
+    Set<String> expectedFiles = ImmutableSet.of(dataFile1.location(), dataFile2.location());
     assertThat(discoveredFiles).containsExactlyInAnyOrderElementsOf(expectedFiles);
 
     IcebergEnumeratorPosition lastPosition = secondResult.toPosition();
@@ -406,10 +404,10 @@ public void testIncrementalFromSnapshotId() throws Exception {
     assertThat(split.task().files()).hasSize(1);
     Set<String> discoveredFiles =
         split.task().files().stream()
-            .map(fileScanTask -> fileScanTask.file().path().toString())
+            .map(fileScanTask -> fileScanTask.file().location())
             .collect(Collectors.toSet());
     // should  discover dataFile2 appended in snapshot2
-    Set<String> expectedFiles = ImmutableSet.of(dataFile2.path().toString());
+    Set<String> expectedFiles = ImmutableSet.of(dataFile2.location());
     assertThat(discoveredFiles).containsExactlyElementsOf(expectedFiles);
 
     IcebergEnumeratorPosition lastPosition = secondResult.toPosition();
@@ -489,10 +487,10 @@ public void testIncrementalFromSnapshotTimestamp() throws Exception {
     assertThat(split.task().files()).hasSize(1);
     Set<String> discoveredFiles =
         split.task().files().stream()
-            .map(fileScanTask -> fileScanTask.file().path().toString())
+            .map(fileScanTask -> fileScanTask.file().location())
             .collect(Collectors.toSet());
     // should discover dataFile2 appended in snapshot2
-    Set<String> expectedFiles = ImmutableSet.of(dataFile2.path().toString());
+    Set<String> expectedFiles = ImmutableSet.of(dataFile2.location());
     assertThat(discoveredFiles).containsExactlyElementsOf(expectedFiles);
 
     IcebergEnumeratorPosition lastPosition = secondResult.toPosition();
@@ -529,12 +527,12 @@ public void testMaxPlanningSnapshotCount() throws Exception {
     ContinuousEnumerationResult secondResult = splitPlanner.planSplits(initialResult.toPosition());
     // should discover dataFile1 appended in snapshot1
     verifyMaxPlanningSnapshotCountResult(
-        secondResult, null, snapshot1, ImmutableSet.of(dataFile1.path().toString()));
+        secondResult, null, snapshot1, ImmutableSet.of(dataFile1.location()));
 
     ContinuousEnumerationResult thirdResult = splitPlanner.planSplits(secondResult.toPosition());
     // should discover dataFile2 appended in snapshot2
     verifyMaxPlanningSnapshotCountResult(
-        thirdResult, snapshot1, snapshot2, ImmutableSet.of(dataFile2.path().toString()));
+        thirdResult, snapshot1, snapshot2, ImmutableSet.of(dataFile2.location()));
   }
 
   @Test
@@ -670,7 +668,7 @@ private void verifyMaxPlanningSnapshotCountResult(
     assertThat(split.task().files()).hasSize(1);
     Set<String> discoveredFiles =
         split.task().files().stream()
-            .map(fileScanTask -> fileScanTask.file().path().toString())
+            .map(fileScanTask -> fileScanTask.file().location())
             .collect(Collectors.toSet());
     assertThat(discoveredFiles).containsExactlyElementsOf(expectedFiles);
   }

flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/source/RowDataFileScanTaskReader.java

@@ -220,7 +220,7 @@ private static class FlinkDeleteFilter extends DeleteFilter<RowData> {
         Schema tableSchema,
         Schema requestedSchema,
         InputFilesDecryptor inputFilesDecryptor) {
-      super(task.file().path().toString(), task.deletes(), tableSchema, requestedSchema);
+      super(task.file().location(), task.deletes(), tableSchema, requestedSchema);
       this.requiredRowType = FlinkSchemaUtil.convert(requiredSchema());
       this.asStructLike = new RowDataWrapper(requiredRowType, requiredSchema().asStruct());
       this.inputFilesDecryptor = inputFilesDecryptor;

flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/source/split/IcebergSourceSplit.java

@@ -107,7 +107,7 @@ private String toString(Collection<FileScanTask> files) {
             .map(
                 fileScanTask ->
                     MoreObjects.toStringHelper(fileScanTask)
-                        .add("file", fileScanTask.file().path().toString())
+                        .add("file", fileScanTask.file().location())
                         .add("start", fileScanTask.start())
                         .add("length", fileScanTask.length())
                         .toString())

flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java

@@ -646,11 +646,11 @@ public void testSetCurrentAndCherryPickSnapshotId() {
   private void validateTableFiles(Table tbl, DataFile... expectedFiles) {
     tbl.refresh();
     Set<CharSequence> expectedFilePaths =
-        Arrays.stream(expectedFiles).map(DataFile::path).collect(Collectors.toSet());
+        Arrays.stream(expectedFiles).map(DataFile::location).collect(Collectors.toSet());
     Set<CharSequence> actualFilePaths =
         StreamSupport.stream(tbl.newScan().planFiles().spliterator(), false)
             .map(FileScanTask::file)
-            .map(ContentFile::path)
+            .map(ContentFile::location)
             .collect(Collectors.toSet());
     assertThat(actualFilePaths).as("Files should match").isEqualTo(expectedFilePaths);
   }

flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/TestHelpers.java

@@ -602,7 +602,7 @@ public static void assertEquals(ContentFile<?> expected, ContentFile<?> actual)
     assertThat(actual).isNotNull();
     assertThat(actual.specId()).as("SpecId").isEqualTo(expected.specId());
     assertThat(actual.content()).as("Content").isEqualTo(expected.content());
-    assertThat(actual.path()).as("Path").isEqualTo(expected.path());
+    assertThat(actual.location()).as("Location").isEqualTo(expected.location());
     assertThat(actual.format()).as("Format").isEqualTo(expected.format());
     assertThat(actual.partition().size())
         .as("Partition size")

flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/actions/TestRewriteDataFilesAction.java

@@ -383,7 +383,7 @@ public void testRewriteAvoidRepeateCompress() throws IOException {
     assertThat(dataFilesRewrote).hasSize(2);
     // the biggest file do not be rewrote
     List rewroteDataFileNames =
-        dataFilesRewrote.stream().map(ContentFile::path).collect(Collectors.toList());
+        dataFilesRewrote.stream().map(ContentFile::location).collect(Collectors.toList());
     assertThat(rewroteDataFileNames).contains(file.getAbsolutePath());
 
     // Assert the table records as expected.

flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/TestBucketPartitionerFlinkIcebergSink.java

@@ -184,7 +184,7 @@ private TableTestStats extractPartitionResults(TableSchemaType tableSchemaType)
       for (FileScanTask scanTask : fileScanTasks) {
         long recordCountInFile = scanTask.file().recordCount();
 
-        String[] splitFilePath = scanTask.file().path().toString().split("/");
+        String[] splitFilePath = scanTask.file().location().split("/");
         // Filename example: 00007-0-a7d3a29a-33e9-4740-88f4-0f494397d60c-00001.parquet
         // Writer ID: .......^^^^^
         String filename = splitFilePath[splitFilePath.length - 1];

flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java

@@ -241,6 +241,6 @@ public void testDeleteStats() throws Exception {
         new String(
             deleteFile.lowerBounds().get(MetadataColumns.DELETE_FILE_PATH.fieldId()).array());
     DataFile dataFile = table.currentSnapshot().addedDataFiles(table.io()).iterator().next();
-    assumeThat(fromStat).isEqualTo(dataFile.path().toString());
+    assumeThat(fromStat).isEqualTo(dataFile.location());
   }
 }

flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkV2.java

@@ -224,7 +224,7 @@ public void testDeleteStats() throws Exception {
         new String(
             deleteFile.lowerBounds().get(MetadataColumns.DELETE_FILE_PATH.fieldId()).array());
     DataFile dataFile = table.currentSnapshot().addedDataFiles(table.io()).iterator().next();
-    assumeThat(fromStat).isEqualTo(dataFile.path().toString());
+    assumeThat(fromStat).isEqualTo(dataFile.location());
   }
 
   protected void testChangeLogs(