(beam) branch master updated: fix AddFilesIT filter for BigLake (#39533)

[yhu]

This is an automated email from the ASF dual-hosted git repository.

Abacn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new 141804ab568 fix AddFilesIT filter for BigLake (#39533)
141804ab568 is described below

commit 141804ab568d4aca1a21b7eb6ad4d083102b29e0
Author: Abdelrahman Ibrahim <abdoibrahim1...@gmail.com>
AuthorDate: Thu Jul 30 22:18:55 2026 +0300

    fix AddFilesIT filter for BigLake (#39533)
    
    * fix AddFilesIT filter for BigLake
    
    * fix Spotless
---
 .../java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java   | 15 ++++++++++++---
 1 file changed, 12 insertions(+), 3 deletions(-)

diff --git 
a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java
 
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java
index 5ecf1455647..54c9cb8dc93 100644
--- 
a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java
+++ 
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java
@@ -68,6 +68,7 @@ import org.apache.beam.sdk.transforms.Deduplicate;
 import org.apache.beam.sdk.transforms.Filter;
 import org.apache.beam.sdk.transforms.JsonToRow;
 import org.apache.beam.sdk.transforms.MapElements;
+import org.apache.beam.sdk.transforms.SerializableFunction;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PCollectionRowTuple;
 import org.apache.beam.sdk.values.Row;
@@ -246,10 +247,13 @@ public class AddFilesIT {
     // first create a source iceberg table
     catalog.createTable(srcTableId, beamSchemaToIcebergSchema(ROW_SCHEMA), 
SPEC);
 
-    String filter = format("%s/%s/data/", namespace, srcTableName);
+    // BigLake may write under {namespace}/{table}/{id}/data/... rather than 
the Hive-style
+    // {namespace}/{table}/data/... layout, so match the table prefix and a 
/data/ segment.
+    String tablePrefix = format("%s/%s/", namespace, srcTableName);
 
     // build AddFiles pipeline and let it run in the background
-    PipelineResult addFilesPipeline = startAddFilesListener(filter);
+    PipelineResult addFilesPipeline =
+        startAddFilesListener(name -> name.contains(tablePrefix) && 
name.contains("/data/"));
 
     // before writing, confirm the destination table still does not exist
     assertFalse(catalog.tableExists(destTableId));
@@ -503,6 +507,11 @@ public class AddFilesIT {
   }
 
   private PipelineResult startAddFilesListener(String filter) throws 
InterruptedException {
+    return startAddFilesListener(name -> name.contains(filter));
+  }
+
+  private PipelineResult startAddFilesListener(
+      SerializableFunction<String, Boolean> objectNameFilter) throws 
InterruptedException {
     DirectOptions options = 
TestPipeline.testingPipelineOptions().as(DirectOptions.class);
     options.setBlockOnRun(false);
     Pipeline p = Pipeline.create(options);
@@ -510,7 +519,7 @@ public class AddFilesIT {
     PCollectionRowTuple tuple =
         p.apply(PubsubIO.readStrings().fromTopic(notificationsTopic))
             .apply(JsonToRow.withSchema(NOTIFICATION_SCHEMA))
-            .apply(Filter.by(row -> row.getString("name").contains(filter)))
+            .apply(Filter.by(row -> 
objectNameFilter.apply(row.getString("name"))))
             .apply(
                 MapElements.into(strings())
                     .via(