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 cdeb84d341 Spark 4.2: Support iterative runtime filtering with
PartitionPredicate (#18600)
cdeb84d341 is described below
commit cdeb84d341631b0c89a4defc61f38eccf5768cea
Author: Xiduo You <[email protected]>
AuthorDate: Thu Oct 22 06:50:36 2026 +0800
Spark 4.2: Support iterative runtime filtering with PartitionPredicate
(#18600)
---
.../spark/source/SparkPartitioningAwareScan.java | 12 ++
.../spark/source/SparkRuntimeFilterableScan.java | 50 +++++
.../iceberg/spark/source/TestRuntimeFiltering.java | 232 ++++++++++++++++++++-
3 files changed, 291 insertions(+), 3 deletions(-)
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 d6a04cd884..f4b1c406d6 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
@@ -306,6 +306,18 @@ abstract class SparkPartitioningAwareScan<T extends
PartitionScanTask> extends S
return evaluator.eval(task.partition());
}
+ /**
+ * Returns the planned tasks that match all partition predicates in the scan.
+ *
+ * <p>Tasks from specs where a predicate cannot be evaluated are kept.
+ */
+ protected synchronized List<T> tasksMatchingPartitionPredicates() {
+ Map<Integer, PartitionPredicateEvaluator> evaluatorsBySpecId =
Maps.newHashMap();
+ return tasks().stream()
+ .filter(task -> matchesPartitionPredicates(task, evaluatorsBySpecId))
+ .collect(Collectors.toList());
+ }
+
private static List<PartitionField> activePartitionFields(PartitionSpec
spec) {
return spec.fields().stream()
.filter(field -> !field.transform().isVoid())
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 25bca85263..44293cbb85 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
@@ -65,6 +65,7 @@ abstract class SparkRuntimeFilterableScan extends
SparkPartitioningAwareScan<Par
private static final Logger LOG =
LoggerFactory.getLogger(SparkRuntimeFilterableScan.class);
private final List<Expression> runtimeFilters;
+ private final List<Predicate> pushedPredicates;
protected SparkRuntimeFilterableScan(
SparkSession spark,
@@ -87,6 +88,7 @@ abstract class SparkRuntimeFilterableScan extends
SparkPartitioningAwareScan<Par
partitionPredicates,
scanReportSupplier);
this.runtimeFilters = Lists.newArrayList();
+ this.pushedPredicates = Lists.newArrayList();
}
@Override
@@ -116,8 +118,55 @@ abstract class SparkRuntimeFilterableScan extends
SparkPartitioningAwareScan<Par
.toArray(NamedReference[]::new);
}
+ @Override
+ public boolean supportsIterativePushdown() {
+ return true;
+ }
+
+ @Override
+ public synchronized Predicate[] pushedPredicates() {
+ return pushedPredicates.toArray(new Predicate[0]);
+ }
+
@Override
public synchronized void filter(Predicate[] predicates) {
+ List<PartitionPredicate> partitionPredicates = Lists.newArrayList();
+ List<Predicate> runtimePredicates = Lists.newArrayList();
+
+ for (Predicate predicate : predicates) {
+ if (predicate instanceof PartitionPredicate partitionPredicate) {
+ partitionPredicates.add(partitionPredicate);
+ } else {
+ runtimePredicates.add(predicate);
+ }
+ }
+
+ if (!runtimePredicates.isEmpty()) {
+ applyRuntimeFilter(runtimePredicates.toArray(new Predicate[0]));
+ }
+
+ if (!partitionPredicates.isEmpty()) {
+ applyPartitionPredicates(partitionPredicates);
+ pushedPredicates.addAll(partitionPredicates);
+ }
+ }
+
+ private void applyPartitionPredicates(List<PartitionPredicate>
additionalPredicates) {
+ partitionPredicates().addAll(additionalPredicates);
+
+ List<PartitionScanTask> filteredTasks = tasksMatchingPartitionPredicates();
+ if (filteredTasks.size() < tasks().size()) {
+ LOG.info(
+ "{} of {} task(s) for table {} matched {} opaque runtime partition
predicate(s)",
+ filteredTasks.size(),
+ tasks().size(),
+ table().name(),
+ additionalPredicates.size());
+ resetTasks(filteredTasks);
+ }
+ }
+
+ private void applyRuntimeFilter(Predicate[] predicates) {
Expression runtimeFilter = convertRuntimePredicates(predicates);
if (runtimeFilter != Expressions.alwaysTrue()) {
@@ -194,6 +243,7 @@ abstract class SparkRuntimeFilterableScan extends
SparkPartitioningAwareScan<Par
try {
Binder.bind(projection().asStruct(), expr, caseSensitive());
filter = Expressions.and(filter, expr);
+ pushedPredicates.add(predicate);
} catch (ValidationException e) {
LOG.warn("Failed to bind {} to expected schema, skipping runtime
filter", expr, e);
}
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestRuntimeFiltering.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestRuntimeFiltering.java
index e7346e270f..0623d63e64 100644
---
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestRuntimeFiltering.java
+++
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestRuntimeFiltering.java
@@ -27,12 +27,15 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.List;
+import java.util.Locale;
import java.util.Set;
import org.apache.commons.lang3.StringUtils;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.Parameter;
import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.Parameters;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.PlanningMode;
import org.apache.iceberg.Table;
import org.apache.iceberg.expressions.Expression;
@@ -46,6 +49,7 @@ import org.apache.iceberg.spark.TestBaseWithCatalog;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.types.DataTypes;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -458,6 +462,190 @@ public class TestRuntimeFiltering extends
TestBaseWithCatalog {
sql(query));
}
+ @TestTemplate
+ public void scalarSubqueryWithUnsupportedPartitionPredicate() {
+ sql("CREATE TABLE %s (id INT, part STRING) USING iceberg PARTITIONED BY
(part)", tableName);
+ configurePlanningMode(planningMode);
+ sql("INSERT INTO %s VALUES (1, 'abc'), (2, 'abx'), (3, 'def'), (4,
'xyz')", tableName);
+
+ sql("CREATE TABLE dim (pattern STRING) USING parquet");
+ sql("INSERT INTO dim VALUES ('^ab')");
+
+ String query =
+ String.format(
+ "SELECT * FROM %s WHERE part RLIKE (SELECT max(pattern) FROM dim)
ORDER BY id",
+ tableName);
+
+ deleteNotMatchingFiles(Expressions.startsWith("part", "ab"));
+
+ assertEquals(
+ "Should have expected rows",
+ sql("SELECT * FROM %s WHERE part RLIKE '^ab' ORDER BY id", tableName),
+ sql(query));
+ }
+
+ @TestTemplate
+ public void scalarSubqueryWithUdfPartitionPredicate() {
+ sql("CREATE TABLE %s (id INT, part STRING) USING iceberg PARTITIONED BY
(part)", tableName);
+ configurePlanningMode(planningMode);
+ sql("INSERT INTO %s VALUES (1, 'a'), (2, 'A'), (3, 'b'), (4, 'B')",
tableName);
+
+ sql("CREATE TABLE dim (val STRING) USING parquet");
+ sql("INSERT INTO dim VALUES ('A')");
+
+ spark
+ .udf()
+ .register(
+ "my_upper_runtime",
+ (String value) -> value == null ? null :
value.toUpperCase(Locale.ROOT),
+ DataTypes.StringType);
+
+ String query =
+ String.format(
+ "SELECT * FROM %s WHERE my_upper_runtime(part) = (SELECT max(val)
FROM dim) ORDER BY id",
+ tableName);
+
+ deleteNotMatchingFiles(
+ Expressions.or(Expressions.equal("part", "a"),
Expressions.equal("part", "A")));
+
+ assertEquals(
+ "Should have expected rows",
+ sql("SELECT * FROM %s WHERE my_upper_runtime(part) = 'A' ORDER BY id",
tableName),
+ sql(query));
+ }
+
+ @TestTemplate
+ public void scalarSubqueryOnNonFirstPartitionColumn() {
+ sql(
+ "CREATE TABLE %s (id INT, p1 INT, p2 STRING) USING iceberg PARTITIONED
BY (p1, p2)",
+ tableName);
+ configurePlanningMode(planningMode);
+ sql(
+ "INSERT INTO %s VALUES (1, 1, 'abc'), (2, 1, 'def'), (3, 2, 'abx'),
(4, 2, 'xyz')",
+ tableName);
+
+ sql("CREATE TABLE dim (pattern STRING) USING parquet");
+ sql("INSERT INTO dim VALUES ('^ab')");
+
+ String query =
+ String.format(
+ "SELECT * FROM %s WHERE p2 RLIKE (SELECT max(pattern) FROM dim)
ORDER BY id",
+ tableName);
+
+ deleteNotMatchingFiles(Expressions.startsWith("p2", "ab"));
+
+ assertEquals(
+ "Should have expected rows",
+ sql("SELECT * FROM %s WHERE p2 RLIKE '^ab' ORDER BY id", tableName),
+ sql(query));
+ }
+
+ @TestTemplate
+ public void scalarSubqueryWithSupportedPartitionPredicate() {
+ sql("CREATE TABLE %s (id INT, part STRING) USING iceberg PARTITIONED BY
(part)", tableName);
+ configurePlanningMode(planningMode);
+ sql("INSERT INTO %s VALUES (1, 'abc'), (2, 'def')", tableName);
+
+ sql("CREATE TABLE dim (val STRING) USING parquet");
+ sql("INSERT INTO dim VALUES ('abc')");
+
+ String query =
+ String.format(
+ "SELECT * FROM %s WHERE part = (SELECT max(val) FROM dim) ORDER BY
id", tableName);
+
+ deleteNotMatchingFiles(Expressions.equal("part", "abc"));
+
+ assertEquals(
+ "Should have expected rows",
+ sql("SELECT * FROM %s WHERE part = 'abc' ORDER BY id", tableName),
+ sql(query));
+ }
+
+ @TestTemplate
+ public void scalarSubqueryWithMultiplePartitionSpecs() {
+ sql("CREATE TABLE %s (id INT, part STRING) USING iceberg PARTITIONED BY
(part)", tableName);
+ configurePlanningMode(planningMode);
+ sql("INSERT INTO %s VALUES (1, 'abc'), (2, 'def')", tableName);
+
+
validationCatalog.loadTable(tableIdent).updateSpec().addField("id").commit();
+ sql("REFRESH TABLE %s", tableName);
+
+ sql("INSERT INTO %s VALUES (3, 'abx'), (4, 'xyz')", tableName);
+
+ sql("CREATE TABLE dim (pattern STRING) USING parquet");
+ sql("INSERT INTO dim VALUES ('^ab')");
+
+ String query =
+ String.format(
+ "SELECT * FROM %s WHERE part RLIKE (SELECT max(pattern) FROM dim)
ORDER BY id",
+ tableName);
+
+ deleteNotMatchingFiles(Expressions.startsWith("part", "ab"));
+
+ assertEquals(
+ "Should have expected rows",
+ sql("SELECT * FROM %s WHERE part RLIKE '^ab' ORDER BY id", tableName),
+ sql(query));
+ }
+
+ @TestTemplate
+ public void scalarSubqueryKeepsTasksWithNonEvaluablePartitionSpec() {
+ sql("CREATE TABLE %s (id INT, data STRING) USING iceberg PARTITIONED BY
(data)", tableName);
+ configurePlanningMode(planningMode);
+ sql("INSERT INTO %s VALUES (1, 'x'), (2, 'y'), (9, 'x')", tableName);
+
+ validationCatalog
+ .loadTable(tableIdent)
+ .updateSpec()
+ .addField("id")
+ .removeField("data")
+ .commit();
+ sql("REFRESH TABLE %s", tableName);
+
+ sql("INSERT INTO %s VALUES (1, 'z'), (4, 'z')", tableName);
+
+ sql("CREATE TABLE dim (val INT) USING parquet");
+ sql("INSERT INTO dim VALUES (1)");
+
+ deletePartitionFile(validationCatalog.loadTable(tableIdent), "id", 1);
+
+ String query =
+ String.format(
+ "SELECT * FROM %s WHERE coalesce(id, 0) > (SELECT max(val) FROM
dim) ORDER BY id",
+ tableName);
+
+ assertEquals(
+ "Should have expected rows",
+ sql("SELECT * FROM %s WHERE id > 1 ORDER BY id", tableName),
+ sql(query));
+ }
+
+ @TestTemplate
+ public void scalarSubqueriesWithAcceptedAndRejectedPredicates() {
+ sql(
+ "CREATE TABLE %s (id INT, p1 STRING, p2 INT) USING iceberg PARTITIONED
BY (p1, p2)",
+ tableName);
+ configurePlanningMode(planningMode);
+ sql("INSERT INTO %s VALUES (1, 'a', 1), (2, 'a', 2), (3, 'b', 1), (4, 'b',
2)", tableName);
+
+ sql("CREATE TABLE dim (v1 STRING, v2 INT) USING parquet");
+ sql("INSERT INTO dim VALUES ('a', 1)");
+
+ String query =
+ String.format(
+ "SELECT * FROM %s WHERE p1 = (SELECT max(v1) FROM dim) "
+ + "AND coalesce(p2, 0) > (SELECT max(v2) FROM dim) ORDER BY
id",
+ tableName);
+
+ deleteNotMatchingFiles(
+ Expressions.and(Expressions.equal("p1", "a"), Expressions.equal("p2",
2)));
+
+ assertEquals(
+ "Should have expected rows",
+ sql("SELECT * FROM %s WHERE p1 = 'a' AND p2 = 2 ORDER BY id",
tableName),
+ sql(query));
+ }
+
private void assertQueryContainsRuntimeFilter(String query) {
assertQueryContainsRuntimeFilters(query, 1, "Query should have 1 runtime
filter");
}
@@ -477,6 +665,18 @@ public class TestRuntimeFiltering extends
TestBaseWithCatalog {
// delete files that don't match the filter to ensure dynamic filtering
works and only required
// files are read
private void deleteNotMatchingFiles(Expression filter, int
expectedDeletedFileCount) {
+ assertThat(deleteFilesNotMatching(filter))
+ .as("Deleted unexpected number of files")
+ .hasSize(expectedDeletedFileCount);
+ }
+
+ private void deleteNotMatchingFiles(Expression filter) {
+ assertThat(deleteFilesNotMatching(filter))
+ .as("Should delete files to verify runtime filtering")
+ .isNotEmpty();
+ }
+
+ private Set<String> deleteFilesNotMatching(Expression filter) {
Table table = validationCatalog.loadTable(tableIdent);
FileIO io = table.io();
@@ -503,8 +703,34 @@ public class TestRuntimeFiltering extends
TestBaseWithCatalog {
throw new UncheckedIOException(e);
}
- assertThat(deletedFileLocations)
- .as("Deleted unexpected number of files")
- .hasSize(expectedDeletedFileCount);
+ return deletedFileLocations;
+ }
+
+ private void deletePartitionFile(Table table, String fieldName, int
partitionValue) {
+ PartitionSpec spec = table.spec();
+ int position = partitionPosition(spec, fieldName);
+ FileIO io = table.io();
+
+ try (CloseableIterable<FileScanTask> tasks = table.newScan().planFiles()) {
+ for (FileScanTask task : tasks) {
+ if (task.spec().specId() == spec.specId()
+ && partitionValue == task.partition().get(position,
Integer.class)) {
+ io.deleteFile(task.file().location());
+ }
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ private static int partitionPosition(PartitionSpec spec, String fieldName) {
+ for (int pos = 0; pos < spec.fields().size(); pos++) {
+ PartitionField field = spec.fields().get(pos);
+ if (field.name().equals(fieldName) && !field.transform().isVoid()) {
+ return pos;
+ }
+ }
+
+ throw new IllegalArgumentException("Cannot find partition field: " +
fieldName);
}
}
