This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new bccd21e040 Spark: Fix concurrent runtime filtering on shared scan for
COW UPDATE with subquery (#18005)
bccd21e040 is described below
commit bccd21e040a8caf0f084d167dd68a481f85c4423
Author: hemanthboyina <[email protected]>
AuthorDate: Thu Oct 8 08:18:12 2026 +0530
Spark: Fix concurrent runtime filtering on shared scan for COW UPDATE with
subquery (#18005)
---
.../iceberg/spark/source/SparkBatchQueryScan.java | 2 +-
.../iceberg/spark/source/SparkCopyOnWriteScan.java | 2 +-
.../spark/source/SparkPartitioningAwareScan.java | 2 +-
.../TestSparkCopyOnWriteScanConcurrency.java | 194 +++++++++++++++++++
.../iceberg/spark/source/SparkBatchQueryScan.java | 2 +-
.../iceberg/spark/source/SparkCopyOnWriteScan.java | 2 +-
.../spark/source/SparkPartitioningAwareScan.java | 2 +-
.../TestSparkCopyOnWriteScanConcurrency.java | 194 +++++++++++++++++++
.../iceberg/spark/source/SparkCopyOnWriteScan.java | 2 +-
.../spark/source/SparkPartitioningAwareScan.java | 2 +-
.../spark/source/SparkRuntimeFilterableScan.java | 2 +-
.../TestSparkCopyOnWriteScanConcurrency.java | 205 ++++++++++++++++++++
.../iceberg/spark/source/SparkCopyOnWriteScan.java | 2 +-
.../spark/source/SparkPartitioningAwareScan.java | 2 +-
.../spark/source/SparkRuntimeFilterableScan.java | 2 +-
.../TestSparkCopyOnWriteScanConcurrency.java | 206 +++++++++++++++++++++
16 files changed, 811 insertions(+), 12 deletions(-)
diff --git
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchQueryScan.java
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchQueryScan.java
index a361a7f1ba..69fc4e2425 100644
---
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchQueryScan.java
+++
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchQueryScan.java
@@ -124,7 +124,7 @@ class SparkBatchQueryScan extends
SparkPartitioningAwareScan<PartitionScanTask>
}
@Override
- public void filter(Predicate[] predicates) {
+ public synchronized void filter(Predicate[] predicates) {
Expression runtimeFilterExpr = convertRuntimeFilters(predicates);
if (runtimeFilterExpr != Expressions.alwaysTrue()) {
diff --git
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkCopyOnWriteScan.java
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkCopyOnWriteScan.java
index 9674a7333f..72b427ad53 100644
---
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkCopyOnWriteScan.java
+++
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkCopyOnWriteScan.java
@@ -118,7 +118,7 @@ class SparkCopyOnWriteScan extends
SparkPartitioningAwareScan<FileScanTask>
}
@Override
- public void filter(Predicate[] predicates) {
+ public synchronized void filter(Predicate[] predicates) {
Preconditions.checkState(
Objects.equals(snapshotId(), currentSnapshotId()),
"Runtime file filtering is not possible: the table has been
concurrently modified. "
diff --git
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java
index 4d9fb7556b..7dad233895 100644
---
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java
+++
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java
@@ -243,7 +243,7 @@ abstract class SparkPartitioningAwareScan<T extends
PartitionScanTask> extends S
// only task groups can be reset while resetting tasks
// the set of scanned specs and grouping key type must never change
- protected void resetTasks(List<T> filteredTasks) {
+ protected synchronized void resetTasks(List<T> filteredTasks) {
this.taskGroups = null;
this.tasks = filteredTasks;
}
diff --git
a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkCopyOnWriteScanConcurrency.java
b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkCopyOnWriteScanConcurrency.java
new file mode 100644
index 0000000000..0182cb3c01
--- /dev/null
+++
b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkCopyOnWriteScanConcurrency.java
@@ -0,0 +1,194 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.source;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import org.apache.iceberg.BatchScan;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.MetadataColumns;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.metrics.ScanReport;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.spark.SparkReadConf;
+import org.apache.iceberg.spark.TestBaseWithCatalog;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.connector.expressions.FieldReference;
+import org.apache.spark.sql.connector.expressions.LiteralValue;
+import org.apache.spark.sql.connector.expressions.NamedReference;
+import org.apache.spark.sql.connector.expressions.filter.Predicate;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.Timeout;
+
+public class TestSparkCopyOnWriteScanConcurrency extends TestBaseWithCatalog {
+
+ private static final long TIMEOUT_SECONDS = 10;
+
+ @AfterEach
+ public void removeTables() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ @TestTemplate
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ public void concurrentRuntimeFilteringNarrowsSharedScan() throws Exception {
+ sql("CREATE TABLE %s (id BIGINT, data STRING) USING iceberg", tableName);
+ sql("INSERT INTO %s VALUES (1, 'a')", tableName);
+ sql("INSERT INTO %s VALUES (2, 'b')", tableName);
+
+ Table table = validationCatalog.loadTable(tableIdent);
+ CountDownLatch enteredResetTasks = new CountDownLatch(1);
+ CountDownLatch releaseFilter = new CountDownLatch(1);
+ InstrumentedScan scan = newScan(table, enteredResetTasks, releaseFilter);
+
+ List<String> locations =
+ scan.tasks().stream().map(task ->
task.file().location()).collect(Collectors.toList());
+ assertThat(locations).hasSize(2);
+ assertThat(
+ scan.taskGroups().stream()
+ .flatMap(group -> group.tasks().stream())
+ .map(task -> task.file().location())
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrderElementsOf(locations);
+ String location = locations.get(0);
+ NamedReference fileRef =
FieldReference.apply(MetadataColumns.FILE_PATH.name());
+ LiteralValue<UTF8String> literal =
+ new LiteralValue<>(UTF8String.fromString(location),
DataTypes.StringType);
+ Predicate[] predicates = {
+ new Predicate(
+ "IN", new org.apache.spark.sql.connector.expressions.Expression[]
{fileRef, literal})
+ };
+
+ FutureTask<Void> winnerTask =
+ new FutureTask<>(
+ () -> {
+ scan.filter(predicates);
+ return null;
+ });
+ FutureTask<Void> loserTask =
+ new FutureTask<>(
+ () -> {
+ scan.filter(predicates);
+ assertThat(scan.tasks())
+ .extracting(task -> task.file().location())
+ .containsExactly(location);
+ assertThat(
+ scan.taskGroups().stream()
+ .flatMap(group -> group.tasks().stream())
+ .map(task -> task.file().location())
+ .collect(Collectors.toList()))
+ .containsExactly(location);
+ return null;
+ });
+ Thread winner = new Thread(winnerTask);
+ Thread loser = new Thread(loserTask);
+
+ winner.start();
+ assertThat(enteredResetTasks.await(TIMEOUT_SECONDS,
TimeUnit.SECONDS)).isTrue();
+ loser.start();
+ try {
+ Awaitility.await()
+ .atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS)
+ .until(() -> loser.getState() == Thread.State.BLOCKED ||
loserTask.isDone());
+ } finally {
+ releaseFilter.countDown();
+ }
+
+ winner.join(TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS));
+ loser.join(TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS));
+ assertThat(winner.isAlive()).isFalse();
+ assertThat(loser.isAlive()).isFalse();
+ assertThat(winnerTask.get()).isNull();
+ assertThat(loserTask.get()).isNull();
+ }
+
+ private InstrumentedScan newScan(
+ Table table, CountDownLatch enteredResetTasks, CountDownLatch
releaseFilter) {
+ SparkReadConf readConf = new SparkReadConf(spark, table,
ImmutableMap.of());
+ Schema schema = table.schema();
+ Snapshot snapshot = table.currentSnapshot();
+ BatchScan batchScan =
+ table
+ .newBatchScan()
+ .useSnapshot(snapshot.snapshotId())
+ .ignoreResiduals()
+ .caseSensitive(readConf.caseSensitive())
+ .filter(Expressions.alwaysTrue())
+ .project(schema);
+ return new InstrumentedScan(
+ spark,
+ table,
+ batchScan,
+ snapshot,
+ readConf,
+ schema,
+ Collections.emptyList(),
+ () -> null,
+ enteredResetTasks,
+ releaseFilter);
+ }
+
+ private static class InstrumentedScan extends SparkCopyOnWriteScan {
+ private final CountDownLatch enteredResetTasks;
+ private final CountDownLatch releaseFilter;
+
+ InstrumentedScan(
+ SparkSession spark,
+ Table table,
+ BatchScan scan,
+ Snapshot snapshot,
+ SparkReadConf readConf,
+ Schema schema,
+ List<Expression> filters,
+ Supplier<ScanReport> scanReportSupplier,
+ CountDownLatch enteredResetTasks,
+ CountDownLatch releaseFilter) {
+ super(spark, table, scan, snapshot, readConf, schema, filters,
scanReportSupplier);
+ this.enteredResetTasks = enteredResetTasks;
+ this.releaseFilter = releaseFilter;
+ }
+
+ @Override
+ protected void resetTasks(List<FileScanTask> filteredTasks) {
+ enteredResetTasks.countDown();
+ try {
+ assertThat(releaseFilter.await(TIMEOUT_SECONDS,
TimeUnit.SECONDS)).isTrue();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(e);
+ }
+ super.resetTasks(filteredTasks);
+ }
+ }
+}
diff --git
a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchQueryScan.java
b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchQueryScan.java
index a361a7f1ba..69fc4e2425 100644
---
a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchQueryScan.java
+++
b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchQueryScan.java
@@ -124,7 +124,7 @@ class SparkBatchQueryScan extends
SparkPartitioningAwareScan<PartitionScanTask>
}
@Override
- public void filter(Predicate[] predicates) {
+ public synchronized void filter(Predicate[] predicates) {
Expression runtimeFilterExpr = convertRuntimeFilters(predicates);
if (runtimeFilterExpr != Expressions.alwaysTrue()) {
diff --git
a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkCopyOnWriteScan.java
b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkCopyOnWriteScan.java
index f957b97d60..7930068189 100644
---
a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkCopyOnWriteScan.java
+++
b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkCopyOnWriteScan.java
@@ -103,7 +103,7 @@ class SparkCopyOnWriteScan extends
SparkPartitioningAwareScan<FileScanTask>
}
@Override
- public void filter(Predicate[] predicates) {
+ public synchronized void filter(Predicate[] predicates) {
Preconditions.checkState(
Objects.equals(snapshotId(), currentSnapshotId()),
"Runtime file filtering is not possible: the table has been
concurrently modified. "
diff --git
a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java
b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java
index 4d9fb7556b..7dad233895 100644
---
a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java
+++
b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java
@@ -243,7 +243,7 @@ abstract class SparkPartitioningAwareScan<T extends
PartitionScanTask> extends S
// only task groups can be reset while resetting tasks
// the set of scanned specs and grouping key type must never change
- protected void resetTasks(List<T> filteredTasks) {
+ protected synchronized void resetTasks(List<T> filteredTasks) {
this.taskGroups = null;
this.tasks = filteredTasks;
}
diff --git
a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkCopyOnWriteScanConcurrency.java
b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkCopyOnWriteScanConcurrency.java
new file mode 100644
index 0000000000..0182cb3c01
--- /dev/null
+++
b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkCopyOnWriteScanConcurrency.java
@@ -0,0 +1,194 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.source;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import org.apache.iceberg.BatchScan;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.MetadataColumns;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.metrics.ScanReport;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.spark.SparkReadConf;
+import org.apache.iceberg.spark.TestBaseWithCatalog;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.connector.expressions.FieldReference;
+import org.apache.spark.sql.connector.expressions.LiteralValue;
+import org.apache.spark.sql.connector.expressions.NamedReference;
+import org.apache.spark.sql.connector.expressions.filter.Predicate;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.Timeout;
+
+public class TestSparkCopyOnWriteScanConcurrency extends TestBaseWithCatalog {
+
+ private static final long TIMEOUT_SECONDS = 10;
+
+ @AfterEach
+ public void removeTables() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ @TestTemplate
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ public void concurrentRuntimeFilteringNarrowsSharedScan() throws Exception {
+ sql("CREATE TABLE %s (id BIGINT, data STRING) USING iceberg", tableName);
+ sql("INSERT INTO %s VALUES (1, 'a')", tableName);
+ sql("INSERT INTO %s VALUES (2, 'b')", tableName);
+
+ Table table = validationCatalog.loadTable(tableIdent);
+ CountDownLatch enteredResetTasks = new CountDownLatch(1);
+ CountDownLatch releaseFilter = new CountDownLatch(1);
+ InstrumentedScan scan = newScan(table, enteredResetTasks, releaseFilter);
+
+ List<String> locations =
+ scan.tasks().stream().map(task ->
task.file().location()).collect(Collectors.toList());
+ assertThat(locations).hasSize(2);
+ assertThat(
+ scan.taskGroups().stream()
+ .flatMap(group -> group.tasks().stream())
+ .map(task -> task.file().location())
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrderElementsOf(locations);
+ String location = locations.get(0);
+ NamedReference fileRef =
FieldReference.apply(MetadataColumns.FILE_PATH.name());
+ LiteralValue<UTF8String> literal =
+ new LiteralValue<>(UTF8String.fromString(location),
DataTypes.StringType);
+ Predicate[] predicates = {
+ new Predicate(
+ "IN", new org.apache.spark.sql.connector.expressions.Expression[]
{fileRef, literal})
+ };
+
+ FutureTask<Void> winnerTask =
+ new FutureTask<>(
+ () -> {
+ scan.filter(predicates);
+ return null;
+ });
+ FutureTask<Void> loserTask =
+ new FutureTask<>(
+ () -> {
+ scan.filter(predicates);
+ assertThat(scan.tasks())
+ .extracting(task -> task.file().location())
+ .containsExactly(location);
+ assertThat(
+ scan.taskGroups().stream()
+ .flatMap(group -> group.tasks().stream())
+ .map(task -> task.file().location())
+ .collect(Collectors.toList()))
+ .containsExactly(location);
+ return null;
+ });
+ Thread winner = new Thread(winnerTask);
+ Thread loser = new Thread(loserTask);
+
+ winner.start();
+ assertThat(enteredResetTasks.await(TIMEOUT_SECONDS,
TimeUnit.SECONDS)).isTrue();
+ loser.start();
+ try {
+ Awaitility.await()
+ .atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS)
+ .until(() -> loser.getState() == Thread.State.BLOCKED ||
loserTask.isDone());
+ } finally {
+ releaseFilter.countDown();
+ }
+
+ winner.join(TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS));
+ loser.join(TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS));
+ assertThat(winner.isAlive()).isFalse();
+ assertThat(loser.isAlive()).isFalse();
+ assertThat(winnerTask.get()).isNull();
+ assertThat(loserTask.get()).isNull();
+ }
+
+ private InstrumentedScan newScan(
+ Table table, CountDownLatch enteredResetTasks, CountDownLatch
releaseFilter) {
+ SparkReadConf readConf = new SparkReadConf(spark, table,
ImmutableMap.of());
+ Schema schema = table.schema();
+ Snapshot snapshot = table.currentSnapshot();
+ BatchScan batchScan =
+ table
+ .newBatchScan()
+ .useSnapshot(snapshot.snapshotId())
+ .ignoreResiduals()
+ .caseSensitive(readConf.caseSensitive())
+ .filter(Expressions.alwaysTrue())
+ .project(schema);
+ return new InstrumentedScan(
+ spark,
+ table,
+ batchScan,
+ snapshot,
+ readConf,
+ schema,
+ Collections.emptyList(),
+ () -> null,
+ enteredResetTasks,
+ releaseFilter);
+ }
+
+ private static class InstrumentedScan extends SparkCopyOnWriteScan {
+ private final CountDownLatch enteredResetTasks;
+ private final CountDownLatch releaseFilter;
+
+ InstrumentedScan(
+ SparkSession spark,
+ Table table,
+ BatchScan scan,
+ Snapshot snapshot,
+ SparkReadConf readConf,
+ Schema schema,
+ List<Expression> filters,
+ Supplier<ScanReport> scanReportSupplier,
+ CountDownLatch enteredResetTasks,
+ CountDownLatch releaseFilter) {
+ super(spark, table, scan, snapshot, readConf, schema, filters,
scanReportSupplier);
+ this.enteredResetTasks = enteredResetTasks;
+ this.releaseFilter = releaseFilter;
+ }
+
+ @Override
+ protected void resetTasks(List<FileScanTask> filteredTasks) {
+ enteredResetTasks.countDown();
+ try {
+ assertThat(releaseFilter.await(TIMEOUT_SECONDS,
TimeUnit.SECONDS)).isTrue();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(e);
+ }
+ super.resetTasks(filteredTasks);
+ }
+ }
+}
diff --git
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkCopyOnWriteScan.java
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkCopyOnWriteScan.java
index b8b3a0edea..5ce5fabc48 100644
---
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkCopyOnWriteScan.java
+++
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkCopyOnWriteScan.java
@@ -92,7 +92,7 @@ class SparkCopyOnWriteScan extends
SparkPartitioningAwareScan<FileScanTask>
}
@Override
- public void filter(Predicate[] predicates) {
+ public synchronized void filter(Predicate[] predicates) {
for (Predicate predicate : predicates) {
// Spark can only pass IN predicates at the moment
if (isFilePathInPredicate(predicate)) {
diff --git
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java
index fe5eeee8fb..646680be6d 100644
---
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java
+++
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java
@@ -244,7 +244,7 @@ abstract class SparkPartitioningAwareScan<T extends
PartitionScanTask> extends S
// only task groups can be reset while resetting tasks
// the set of scanned specs and grouping key type must never change
- protected void resetTasks(List<T> filteredTasks) {
+ protected synchronized void resetTasks(List<T> filteredTasks) {
this.taskGroups = null;
this.tasks = filteredTasks;
}
diff --git
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkRuntimeFilterableScan.java
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkRuntimeFilterableScan.java
index dfd38a1548..a1f19a4bdc 100644
---
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkRuntimeFilterableScan.java
+++
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkRuntimeFilterableScan.java
@@ -106,7 +106,7 @@ abstract class SparkRuntimeFilterableScan extends
SparkPartitioningAwareScan<Par
}
@Override
- public void filter(Predicate[] predicates) {
+ public synchronized void filter(Predicate[] predicates) {
Expression runtimeFilter = convertRuntimePredicates(predicates);
if (runtimeFilter != Expressions.alwaysTrue()) {
diff --git
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkCopyOnWriteScanConcurrency.java
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkCopyOnWriteScanConcurrency.java
new file mode 100644
index 0000000000..380461d295
--- /dev/null
+++
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkCopyOnWriteScanConcurrency.java
@@ -0,0 +1,205 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.source;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import org.apache.iceberg.BatchScan;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.MetadataColumns;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.metrics.ScanReport;
+import org.apache.iceberg.spark.SparkReadConf;
+import org.apache.iceberg.spark.TestBaseWithCatalog;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.connector.expressions.FieldReference;
+import org.apache.spark.sql.connector.expressions.LiteralValue;
+import org.apache.spark.sql.connector.expressions.NamedReference;
+import org.apache.spark.sql.connector.expressions.filter.Predicate;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.Timeout;
+
+public class TestSparkCopyOnWriteScanConcurrency extends TestBaseWithCatalog {
+
+ private static final long TIMEOUT_SECONDS = 10;
+
+ @AfterEach
+ public void removeTables() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ @TestTemplate
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ public void concurrentRuntimeFilteringNarrowsSharedScan() throws Exception {
+ sql("CREATE TABLE %s (id BIGINT, data STRING) USING iceberg", tableName);
+ sql("INSERT INTO %s VALUES (1, 'a')", tableName);
+ sql("INSERT INTO %s VALUES (2, 'b')", tableName);
+
+ Table table = validationCatalog.loadTable(tableIdent);
+ CountDownLatch enteredResetTasks = new CountDownLatch(1);
+ CountDownLatch releaseFilter = new CountDownLatch(1);
+ InstrumentedScan scan = newScan(table, enteredResetTasks, releaseFilter);
+
+ List<String> locations =
+ scan.tasks().stream().map(task ->
task.file().location()).collect(Collectors.toList());
+ assertThat(locations).hasSize(2);
+ assertThat(
+ scan.taskGroups().stream()
+ .flatMap(group -> group.tasks().stream())
+ .map(task -> task.file().location())
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrderElementsOf(locations);
+ String location = locations.get(0);
+ NamedReference fileRef =
FieldReference.apply(MetadataColumns.FILE_PATH.name());
+ LiteralValue<UTF8String> literal =
+ new LiteralValue<>(UTF8String.fromString(location),
DataTypes.StringType);
+ Predicate[] predicates = {
+ new Predicate(
+ "IN", new org.apache.spark.sql.connector.expressions.Expression[]
{fileRef, literal})
+ };
+
+ FutureTask<Void> winnerTask =
+ new FutureTask<>(
+ () -> {
+ scan.filter(predicates);
+ return null;
+ });
+ FutureTask<Void> loserTask =
+ new FutureTask<>(
+ () -> {
+ scan.filter(predicates);
+ assertThat(scan.tasks())
+ .extracting(task -> task.file().location())
+ .containsExactly(location);
+ assertThat(
+ scan.taskGroups().stream()
+ .flatMap(group -> group.tasks().stream())
+ .map(task -> task.file().location())
+ .collect(Collectors.toList()))
+ .containsExactly(location);
+ return null;
+ });
+ Thread winner = new Thread(winnerTask);
+ Thread loser = new Thread(loserTask);
+
+ winner.start();
+ assertThat(enteredResetTasks.await(TIMEOUT_SECONDS,
TimeUnit.SECONDS)).isTrue();
+ loser.start();
+ try {
+ Awaitility.await()
+ .atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS)
+ .until(() -> loser.getState() == Thread.State.BLOCKED ||
loserTask.isDone());
+ } finally {
+ releaseFilter.countDown();
+ }
+
+ winner.join(TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS));
+ loser.join(TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS));
+ assertThat(winner.isAlive()).isFalse();
+ assertThat(loser.isAlive()).isFalse();
+ assertThat(winnerTask.get()).isNull();
+ assertThat(loserTask.get()).isNull();
+ }
+
+ private InstrumentedScan newScan(
+ Table table, CountDownLatch enteredResetTasks, CountDownLatch
releaseFilter) {
+ SparkReadConf readConf = new SparkReadConf(spark, table);
+ Schema schema = table.schema();
+ Snapshot snapshot = table.currentSnapshot();
+ BatchScan batchScan =
+ table
+ .newBatchScan()
+ .useSnapshot(snapshot.snapshotId())
+ .ignoreResiduals()
+ .caseSensitive(readConf.caseSensitive())
+ .filter(Expressions.alwaysTrue())
+ .project(schema);
+ return new InstrumentedScan(
+ spark,
+ table,
+ schema,
+ snapshot,
+ batchScan,
+ readConf,
+ schema,
+ Collections.emptyList(),
+ () -> null,
+ enteredResetTasks,
+ releaseFilter);
+ }
+
+ private static class InstrumentedScan extends SparkCopyOnWriteScan {
+ private final CountDownLatch enteredResetTasks;
+ private final CountDownLatch releaseFilter;
+
+ InstrumentedScan(
+ SparkSession spark,
+ Table table,
+ Schema schema,
+ Snapshot snapshot,
+ BatchScan scan,
+ SparkReadConf readConf,
+ Schema projection,
+ List<Expression> filters,
+ Supplier<ScanReport> scanReportSupplier,
+ CountDownLatch enteredResetTasks,
+ CountDownLatch releaseFilter) {
+ super(
+ spark,
+ table,
+ schema,
+ snapshot,
+ null,
+ scan,
+ readConf,
+ projection,
+ filters,
+ scanReportSupplier);
+ this.enteredResetTasks = enteredResetTasks;
+ this.releaseFilter = releaseFilter;
+ }
+
+ @Override
+ protected void resetTasks(List<FileScanTask> filteredTasks) {
+ enteredResetTasks.countDown();
+ try {
+ assertThat(releaseFilter.await(TIMEOUT_SECONDS,
TimeUnit.SECONDS)).isTrue();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(e);
+ }
+ super.resetTasks(filteredTasks);
+ }
+ }
+}
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkCopyOnWriteScan.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkCopyOnWriteScan.java
index 528409f96b..7ebc013431 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkCopyOnWriteScan.java
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkCopyOnWriteScan.java
@@ -103,7 +103,7 @@ class SparkCopyOnWriteScan extends
SparkPartitioningAwareScan<FileScanTask>
}
@Override
- public void filter(Predicate[] predicates) {
+ public synchronized void filter(Predicate[] predicates) {
for (Predicate predicate : predicates) {
// Spark can only pass IN predicates at the moment
if (isFilePathInPredicate(predicate)) {
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java
index 124f00f0b9..d6a04cd884 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java
@@ -273,7 +273,7 @@ abstract class SparkPartitioningAwareScan<T extends
PartitionScanTask> extends S
// only task groups can be reset while resetting tasks
// the set of scanned specs and grouping key type must never change
- protected void resetTasks(List<T> filteredTasks) {
+ protected synchronized void resetTasks(List<T> filteredTasks) {
this.taskGroups = null;
this.tasks = filteredTasks;
}
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkRuntimeFilterableScan.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkRuntimeFilterableScan.java
index 130e453196..25bca85263 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkRuntimeFilterableScan.java
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkRuntimeFilterableScan.java
@@ -117,7 +117,7 @@ abstract class SparkRuntimeFilterableScan extends
SparkPartitioningAwareScan<Par
}
@Override
- public void filter(Predicate[] predicates) {
+ public synchronized void filter(Predicate[] predicates) {
Expression runtimeFilter = convertRuntimePredicates(predicates);
if (runtimeFilter != Expressions.alwaysTrue()) {
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkCopyOnWriteScanConcurrency.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkCopyOnWriteScanConcurrency.java
new file mode 100644
index 0000000000..120985f154
--- /dev/null
+++
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkCopyOnWriteScanConcurrency.java
@@ -0,0 +1,206 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.source;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import org.apache.iceberg.BatchScan;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.MetadataColumns;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.metrics.ScanReport;
+import org.apache.iceberg.spark.SparkReadConf;
+import org.apache.iceberg.spark.TestBaseWithCatalog;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.connector.expressions.FieldReference;
+import org.apache.spark.sql.connector.expressions.LiteralValue;
+import org.apache.spark.sql.connector.expressions.NamedReference;
+import org.apache.spark.sql.connector.expressions.filter.Predicate;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.Timeout;
+
+public class TestSparkCopyOnWriteScanConcurrency extends TestBaseWithCatalog {
+
+ private static final long TIMEOUT_SECONDS = 10;
+
+ @AfterEach
+ public void removeTables() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ @TestTemplate
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ public void concurrentRuntimeFilteringNarrowsSharedScan() throws Exception {
+ sql("CREATE TABLE %s (id BIGINT, data STRING) USING iceberg", tableName);
+ sql("INSERT INTO %s VALUES (1, 'a')", tableName);
+ sql("INSERT INTO %s VALUES (2, 'b')", tableName);
+
+ Table table = validationCatalog.loadTable(tableIdent);
+ CountDownLatch enteredResetTasks = new CountDownLatch(1);
+ CountDownLatch releaseFilter = new CountDownLatch(1);
+ InstrumentedScan scan = newScan(table, enteredResetTasks, releaseFilter);
+
+ List<String> locations =
+ scan.tasks().stream().map(task ->
task.file().location()).collect(Collectors.toList());
+ assertThat(locations).hasSize(2);
+ assertThat(
+ scan.taskGroups().stream()
+ .flatMap(group -> group.tasks().stream())
+ .map(task -> task.file().location())
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrderElementsOf(locations);
+ String location = locations.get(0);
+ NamedReference fileRef =
FieldReference.apply(MetadataColumns.FILE_PATH.name());
+ LiteralValue<UTF8String> literal =
+ new LiteralValue<>(UTF8String.fromString(location),
DataTypes.StringType);
+ Predicate[] predicates = {
+ new Predicate(
+ "IN", new org.apache.spark.sql.connector.expressions.Expression[]
{fileRef, literal})
+ };
+
+ FutureTask<Void> winnerTask =
+ new FutureTask<>(
+ () -> {
+ scan.filter(predicates);
+ return null;
+ });
+ FutureTask<Void> loserTask =
+ new FutureTask<>(
+ () -> {
+ scan.filter(predicates);
+ assertThat(scan.tasks())
+ .extracting(task -> task.file().location())
+ .containsExactly(location);
+ assertThat(
+ scan.taskGroups().stream()
+ .flatMap(group -> group.tasks().stream())
+ .map(task -> task.file().location())
+ .collect(Collectors.toList()))
+ .containsExactly(location);
+ return null;
+ });
+ Thread winner = new Thread(winnerTask);
+ Thread loser = new Thread(loserTask);
+
+ winner.start();
+ assertThat(enteredResetTasks.await(TIMEOUT_SECONDS,
TimeUnit.SECONDS)).isTrue();
+ loser.start();
+ try {
+ Awaitility.await()
+ .atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS)
+ .until(() -> loser.getState() == Thread.State.BLOCKED ||
loserTask.isDone());
+ } finally {
+ releaseFilter.countDown();
+ }
+
+ winner.join(TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS));
+ loser.join(TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS));
+ assertThat(winner.isAlive()).isFalse();
+ assertThat(loser.isAlive()).isFalse();
+ assertThat(winnerTask.get()).isNull();
+ assertThat(loserTask.get()).isNull();
+ }
+
+ private InstrumentedScan newScan(
+ Table table, CountDownLatch enteredResetTasks, CountDownLatch
releaseFilter) {
+ SparkReadConf readConf = new SparkReadConf(spark, table);
+ Schema schema = table.schema();
+ Snapshot snapshot = table.currentSnapshot();
+ BatchScan batchScan =
+ table
+ .newBatchScan()
+ .useSnapshot(snapshot.snapshotId())
+ .ignoreResiduals()
+ .caseSensitive(readConf.caseSensitive())
+ .filter(Expressions.alwaysTrue())
+ .project(schema);
+ return new InstrumentedScan(
+ spark,
+ table,
+ schema,
+ snapshot,
+ batchScan,
+ readConf,
+ schema,
+ Collections.emptyList(),
+ () -> null,
+ enteredResetTasks,
+ releaseFilter);
+ }
+
+ private static class InstrumentedScan extends SparkCopyOnWriteScan {
+ private final CountDownLatch enteredResetTasks;
+ private final CountDownLatch releaseFilter;
+
+ InstrumentedScan(
+ SparkSession spark,
+ Table table,
+ Schema schema,
+ Snapshot snapshot,
+ BatchScan scan,
+ SparkReadConf readConf,
+ Schema projection,
+ List<Expression> filters,
+ Supplier<ScanReport> scanReportSupplier,
+ CountDownLatch enteredResetTasks,
+ CountDownLatch releaseFilter) {
+ super(
+ spark,
+ table,
+ schema,
+ snapshot,
+ null,
+ scan,
+ readConf,
+ projection,
+ filters,
+ Collections.emptyList(),
+ scanReportSupplier);
+ this.enteredResetTasks = enteredResetTasks;
+ this.releaseFilter = releaseFilter;
+ }
+
+ @Override
+ protected void resetTasks(List<FileScanTask> filteredTasks) {
+ enteredResetTasks.countDown();
+ try {
+ assertThat(releaseFilter.await(TIMEOUT_SECONDS,
TimeUnit.SECONDS)).isTrue();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(e);
+ }
+ super.resetTasks(filteredTasks);
+ }
+ }
+}
