From 0a7c6f1abcbbb90f3acc5d17ef3cde022f5da5ad Mon Sep 17 00:00:00 2001 From: Christopher Wirt Date: Wed, 11 Nov 2026 15:49:25 +0000 Subject: [PATCH 1/2] feat(spark): push list predicates into V2 native scans Signed-off-by: Christopher Wirt --- .../vortex/spark/read/NativePredicate.java | 18 ++ .../spark/read/NestedSchemaPruning.java | 3 +- .../java/dev/vortex/spark/read/ReadTypes.java | 26 ++- .../read/SparkFilterToVortexExpression.java | 2 +- .../spark/read/VortexPartitionReader.java | 40 ++++- .../read/VortexPartitionReaderFactory.java | 13 +- .../vortex/spark/read/VortexListFilter.scala | 121 +++++++++++++ .../dev/vortex/spark/read/VortexScan.scala | 5 + .../spark/VortexFilterPushdownTest.java | 38 ++++ .../spark/read/VortexListFilterTest.scala | 167 ++++++++++++++++++ 10 files changed, 422 insertions(+), 11 deletions(-) create mode 100644 java/vortex-spark/common/src/main/java/dev/vortex/spark/read/NativePredicate.java create mode 100644 java/vortex-spark/common/src/main/scala/dev/vortex/spark/read/VortexListFilter.scala create mode 100644 java/vortex-spark/common/src/test/scala/dev/vortex/spark/read/VortexListFilterTest.scala diff --git a/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/NativePredicate.java b/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/NativePredicate.java new file mode 100644 index 00000000000..1b58cd0a293 --- /dev/null +++ b/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/NativePredicate.java @@ -0,0 +1,18 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +package dev.vortex.spark.read; + +import dev.vortex.api.Expression; +import java.io.Serializable; +import java.util.Optional; +import org.apache.spark.sql.types.StructType; + +/** A serializable predicate whose native handles are built separately on each executor. */ +public interface NativePredicate extends Serializable { + /** Field paths needed to validate the predicate against a file's physical schema. */ + String[][] references(); + + /** Returns empty if this file cannot safely evaluate the predicate. */ + Optional convert(StructType schema); +} diff --git a/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/NestedSchemaPruning.java b/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/NestedSchemaPruning.java index 80660ff2bc1..e8c21cf96e9 100644 --- a/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/NestedSchemaPruning.java +++ b/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/NestedSchemaPruning.java @@ -155,8 +155,7 @@ private static Node struct( boolean filled = false; for (int i = 0; i < fields.length; i++) { String fieldPath = path.isEmpty() ? fields[i].name() : path + "." + fields[i].name(); - int index = lookup.indexOf( - fields[i].name(), resolution.fileName(), path.isEmpty() ? "" : path + "."); + int index = lookup.indexOf(fields[i].name(), resolution.fileName(), path.isEmpty() ? "" : path + "."); if (index < 0) { if (!fields[i].nullable()) { throw new IllegalArgumentException(String.format( diff --git a/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/ReadTypes.java b/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/ReadTypes.java index 2e60f0a6c22..054165d4815 100644 --- a/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/ReadTypes.java +++ b/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/ReadTypes.java @@ -184,6 +184,10 @@ static StructType filterable(StructType schema, List fileFields, Filter[] for (Filter filter : filters) { paths.addAll(Arrays.asList(filter.v2references())); } + return filterable(schema, fileFields, paths); + } + + static StructType filterable(StructType schema, List fileFields, List paths) { return filterable(schema, fileFields, paths, 0); } @@ -210,7 +214,9 @@ private static StructType filterable(StructType schema, List fileFields, if (!(fileField.getType() instanceof ArrowType.Struct)) { continue; } - type = filterable(struct, fileField.getChildren(), entry.getValue(), depth + 1); + type = entry.getValue().stream().anyMatch(path -> path.length == depth + 1) + ? (literalType(fileField, struct) ? struct : DataTypes.NullType) + : filterable(struct, fileField.getChildren(), entry.getValue(), depth + 1); } else { type = literalType(fileField, field.dataType()) ? field.dataType() : DataTypes.NullType; } @@ -221,6 +227,24 @@ private static StructType filterable(StructType schema, List fileFields, /** Whether the file stores {@code read} values as the native scan compares filter literals of that type. */ private static boolean literalType(Field file, DataType read) { + if (read instanceof ArrayType array) { + return (file.getType() instanceof ArrowType.List || file.getType() instanceof ArrowType.ListView) + && file.getChildren().size() == 1 + && literalType(file.getChildren().get(0), array.elementType()); + } + if (read instanceof StructType struct) { + if (!(file.getType() instanceof ArrowType.Struct)) { + return false; + } + Function fields = lookup(file.getChildren(), true); + for (StructField child : struct.fields()) { + Field stored = fields.apply(child.name()); + if (stored == null || !literalType(stored, child.dataType())) { + return false; + } + } + return true; + } if (!read.equals(leafType(file))) { return false; } diff --git a/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/SparkFilterToVortexExpression.java b/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/SparkFilterToVortexExpression.java index b4ca1b3be8d..ee2cc82ec63 100644 --- a/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/SparkFilterToVortexExpression.java +++ b/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/SparkFilterToVortexExpression.java @@ -332,7 +332,7 @@ private static StructField findField(StructType schema, String name) { return null; } - private static Optional literal(Object value, DataType dataType) { + static Optional literal(Object value, DataType dataType) { if (dataType instanceof BooleanType) { return value == null ? Optional.of(Expression.nullLiteralBool()) diff --git a/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/VortexPartitionReader.java b/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/VortexPartitionReader.java index 2fa75f9f965..8376a161ab8 100644 --- a/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/VortexPartitionReader.java +++ b/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/VortexPartitionReader.java @@ -78,6 +78,28 @@ public VortexPartitionReader( VortexOptions formatOptions, Filter[] pushedFilters, boolean caseSensitive) { + this( + file, + dataSchema, + readDataSchema, + readPartitionSchema, + io, + formatOptions, + pushedFilters, + new NativePredicate[0], + caseSensitive); + } + + public VortexPartitionReader( + PartitionedFile file, + StructType dataSchema, + StructType readDataSchema, + StructType readPartitionSchema, + VortexIo io, + VortexOptions formatOptions, + Filter[] pushedFilters, + NativePredicate[] listFilters, + boolean caseSensitive) { this.file = file; this.readDataSchema = readDataSchema; this.readPartitionSchema = readPartitionSchema; @@ -131,7 +153,8 @@ public VortexPartitionReader( // Filters are converted against this file's own columns. Spark evaluates every data filter above the // scan anyway, so one that reads a column this file lacks, or compares one it stores in another type, is // dropped rather than pushed. - buildFilterExpression(pushedFilters, dataSchema, arrowFields).ifPresent(options::filter); + buildFilterExpression(pushedFilters, listFilters, dataSchema, arrowFields) + .ifPresent(options::filter); Optional rows = splitRows(file, dataSource.rowCount()); if (rows.isPresent() && rows.get().count() == 0) { // The native scan reads a range of 0 to 0 as the whole file. @@ -200,11 +223,9 @@ private Map fieldNames(StructType fileSchema, boolean caseSensit return names; } - private static Optional buildFilterExpression( - Filter[] filters, StructType dataSchema, List fileFields) { - if (filters == null || filters.length == 0) { - return Optional.empty(); - } + static Optional buildFilterExpression( + Filter[] filters, NativePredicate[] listFilters, StructType dataSchema, List fileFields) { + filters = filters == null ? new Filter[0] : filters; StructType filterable = ReadTypes.filterable(dataSchema, fileFields, filters); Expression combined = null; for (Filter filter : filters) { @@ -213,6 +234,13 @@ private static Optional buildFilterExpression( combined = combined == null ? converted.get() : Expression.and(combined, converted.get()); } } + for (NativePredicate filter : listFilters) { + StructType listSchema = ReadTypes.filterable(dataSchema, fileFields, Arrays.asList(filter.references())); + Optional converted = filter.convert(listSchema); + if (converted.isPresent()) { + combined = combined == null ? converted.get() : Expression.and(combined, converted.get()); + } + } return Optional.ofNullable(combined); } diff --git a/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/VortexPartitionReaderFactory.java b/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/VortexPartitionReaderFactory.java index 13f8ee06922..8836129e74a 100644 --- a/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/VortexPartitionReaderFactory.java +++ b/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/VortexPartitionReaderFactory.java @@ -39,6 +39,7 @@ public final class VortexPartitionReaderFactory extends FilePartitionReaderFacto private final StructType readDataSchema; private final StructType readPartitionSchema; private final Filter[] pushedFilters; + private final NativePredicate[] listFilters; private final boolean caseSensitive; public VortexPartitionReaderFactory( @@ -49,6 +50,7 @@ public VortexPartitionReaderFactory( StructType readDataSchema, StructType readPartitionSchema, Filter[] pushedFilters, + NativePredicate[] listFilters, boolean caseSensitive) { this.fileOptions = fileOptions; this.io = io; @@ -57,6 +59,7 @@ public VortexPartitionReaderFactory( this.readDataSchema = readDataSchema; this.readPartitionSchema = readPartitionSchema; this.pushedFilters = pushedFilters == null ? new Filter[0] : pushedFilters.clone(); + this.listFilters = listFilters.clone(); this.caseSensitive = caseSensitive; } @@ -77,7 +80,15 @@ public PartitionReader buildColumnarReader(PartitionedFile file) } return new VortexPartitionReader( - file, dataSchema, readDataSchema, readPartitionSchema, io, formatOptions, pushedFilters, caseSensitive); + file, + dataSchema, + readDataSchema, + readPartitionSchema, + io, + formatOptions, + pushedFilters, + listFilters, + caseSensitive); } @Override diff --git a/java/vortex-spark/common/src/main/scala/dev/vortex/spark/read/VortexListFilter.scala b/java/vortex-spark/common/src/main/scala/dev/vortex/spark/read/VortexListFilter.scala new file mode 100644 index 00000000000..dcf6e1d40f5 --- /dev/null +++ b/java/vortex-spark/common/src/main/scala/dev/vortex/spark/read/VortexListFilter.scala @@ -0,0 +1,121 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +package dev.vortex.spark.read + +import java.util.Optional + +import org.apache.spark.sql.catalyst.expressions._ +import org.apache.spark.sql.types.StructType + +import dev.vortex.api.{Expression => VortexExpression} +import dev.vortex.api.Expression.BinaryOp + +/** Carries a Catalyst predicate to executors without serializing native + * handles. Spark retains the original predicate as a residual filter. + */ +final case class VortexListFilter(predicate: Expression) + extends NativePredicate { + override def references(): Array[Array[String]] = + predicate.references.toSeq.map(attribute => Array(attribute.name)).toArray + + def convert(schema: StructType): Optional[VortexExpression] = + VortexListFilter.translate(predicate, schema) match { + case Some(expression) => Optional.of(expression) + case None => Optional.empty() + } +} + +object VortexListFilter { + def from(expression: Expression, schema: StructType): Seq[NativePredicate] = + expression match { + case And(left, right) => from(left, schema) ++ from(right, schema) + case _: ArrayContains | _: ArrayExists => + if (translate(expression, schema).isDefined) + Seq(VortexListFilter(expression)) + else Seq.empty + case _ => Seq.empty + } + + private def translate( + expression: Expression, + schema: StructType, + variable: Option[NamedLambdaVariable] = None + ): Option[VortexExpression] = { + def child(e: Expression): Option[VortexExpression] = + translate(e, schema, variable) + def comparable(e: Expression): Boolean = + SparkFilterToVortexExpression.literal(null, e.dataType).isPresent && + (!e.dataType.isInstanceOf[org.apache.spark.sql.types.StringType] || + e.dataType == org.apache.spark.sql.types.DataTypes.StringType) + def binary(left: Expression, right: Expression, op: BinaryOp) = + for { + _ <- if (comparable(left) && comparable(right)) Some(()) else None + l <- child(left) + r <- child(right) + } yield VortexExpression.binary(op, l, r) + + expression match { + case attribute: AttributeReference + if variable.isEmpty && schema.fields.exists(field => + field.name == attribute.name && field.dataType == attribute.dataType + ) => + Some(VortexExpression.column(attribute.name)) + case parameter: NamedLambdaVariable + if variable.exists(_.exprId == parameter.exprId) => + Some(VortexExpression.variable(parameter.exprId.toString)) + case field: GetStructField => + child(field.child).map( + VortexExpression.getItem( + field.child.dataType + .asInstanceOf[StructType] + .fields(field.ordinal) + .name, + _ + ) + ) + case literal: Literal => + val converted = + SparkFilterToVortexExpression.literal(literal.value, literal.dataType) + if (converted.isPresent) Some(converted.get()) else None + case ArrayContains(list, needle) if comparable(needle) => + for { + l <- child(list) + n <- child(needle) + } yield VortexExpression.listContains(l, n, true) + case exists: ArrayExists if variable.isEmpty => + exists.function match { + case LambdaFunction(body, Seq(parameter: NamedLambdaVariable), _) => + for { + list <- child(exists.argument) + predicate <- translate(body, schema, Some(parameter)) + } yield VortexExpression.listContains( + VortexExpression + .listTransform(list, parameter.exprId.toString, predicate), + VortexExpression.literal(true), + exists.followThreeValuedLogic + ) + case _ => None + } + case EqualTo(left, right) => binary(left, right, BinaryOp.EQ) + case GreaterThan(left, right) => binary(left, right, BinaryOp.GT) + case GreaterThanOrEqual(left, right) => binary(left, right, BinaryOp.GTE) + case LessThan(left, right) => binary(left, right, BinaryOp.LT) + case LessThanOrEqual(left, right) => binary(left, right, BinaryOp.LTE) + case And(left, right) => + for { l <- child(left); r <- child(right) } yield VortexExpression.and( + l, + r + ) + case Or(left, right) => + for { l <- child(left); r <- child(right) } yield VortexExpression.or( + l, + r + ) + case Not(value) => child(value).map(VortexExpression.not) + case IsNull(value) => child(value).map(VortexExpression.isNull) + case IsNotNull(value) => child(value).map(VortexExpression.isNotNull) + case _ => None + } + } +} diff --git a/java/vortex-spark/common/src/main/scala/dev/vortex/spark/read/VortexScan.scala b/java/vortex-spark/common/src/main/scala/dev/vortex/spark/read/VortexScan.scala index 71e35da5df2..97755a2aedd 100644 --- a/java/vortex-spark/common/src/main/scala/dev/vortex/spark/read/VortexScan.scala +++ b/java/vortex-spark/common/src/main/scala/dev/vortex/spark/read/VortexScan.scala @@ -55,6 +55,9 @@ final class VortexScan( override val dataFilters: Seq[Expression] ) extends FileScan { + private lazy val listFilters: Array[NativePredicate] = + dataFilters.flatMap(VortexListFilter.from(_, dataSchema)).toArray + private val caseSensitiveOptions = options.asCaseSensitiveMap.asScala.toMap private val hadoopConf = sparkSession.sessionState.newHadoopConfWithOptions(caseSensitiveOptions) @@ -96,6 +99,7 @@ final class VortexScan( readDataSchema, readPartitionSchema, pushedFilters, + listFilters, caseSensitive ) } @@ -163,6 +167,7 @@ final class VortexScan( .getOrElse("[]") super.getMetaData() ++ Map( "PushedFilters" -> seqToString(pushedFilters.toSeq), + "PushedListFilters" -> seqToString(listFilters.toSeq), "PushedAggregation" -> aggregation, "PushedGroupBy" -> groupBy ) diff --git a/java/vortex-spark/common/src/test/java/dev/vortex/spark/VortexFilterPushdownTest.java b/java/vortex-spark/common/src/test/java/dev/vortex/spark/VortexFilterPushdownTest.java index b3d6095a3e7..fb27bda69f8 100644 --- a/java/vortex-spark/common/src/test/java/dev/vortex/spark/VortexFilterPushdownTest.java +++ b/java/vortex-spark/common/src/test/java/dev/vortex/spark/VortexFilterPushdownTest.java @@ -35,6 +35,8 @@ import org.junit.jupiter.api.TestInstance; import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; import org.junit.jupiter.params.provider.ValueSource; /** @@ -97,6 +99,42 @@ void nullSafeEqualityPreservesNullsUnderNegation(boolean v1) { } } + @ParameterizedTest + @MethodSource("listPredicateCases") + void listPredicatesArePushedIntoV2Scans(boolean threeValuedLogic, String predicate) { + SparkSession session = spark.newSession(); + SparkSession.setActiveSession(session); + session.conf().set("spark.sql.sources.useV1SourceList", ""); + session.conf().set("spark.sql.legacy.followThreeValuedLogicInArrayExists", threeValuedLogic); + Dataset source = session.sql("SELECT 0 AS id, array('sports', CAST(NULL AS STRING)) AS tags, " + + "array(named_struct('price', 101), named_struct('price', CAST(NULL AS INT))) AS items " + + "UNION ALL SELECT 1, array('news', NULL), array(named_struct('price', 10), named_struct('price', NULL)) " + + "UNION ALL SELECT 2, array(), array() " + + "UNION ALL SELECT 3, NULL, NULL " + + "UNION ALL SELECT 4, array(NULL), array(CAST(NULL AS STRUCT))"); + Path output = tempDir.resolve("list_predicates_" + threeValuedLogic); + source.coalesce(1).write().format("vortex").save(output.toString()); + Dataset data = session.read().format("vortex").load(output.toString()); + Dataset filtered = data.where(predicate); + assertEquals( + source.where(predicate).orderBy("id").collectAsList(), + filtered.orderBy("id").collectAsList(), + predicate); + assertTrue( + filtered.queryExecution().executedPlan().toString().contains("VortexListFilter"), + filtered.queryExecution().executedPlan().toString()); + } + + static Stream listPredicateCases() { + return Stream.of(false, true).flatMap(threeValuedLogic -> Stream.of( + "array_contains(tags, 'sports')", + "exists(items, x -> x.price > 100)", + "exists(items, x -> x.price IS NULL)", + "exists(items, x -> x.price > 100 OR x.price < 20)", + "exists(items, x -> NOT (x.price > 100))") + .map(predicate -> Arguments.of(threeValuedLogic, predicate))); + } + @Test @DisplayName("Equality, comparison, IS NULL, IN, AND/OR/NOT all return correct rows after pushdown") public void testFilterPushdownCorrectness() throws IOException { diff --git a/java/vortex-spark/common/src/test/scala/dev/vortex/spark/read/VortexListFilterTest.scala b/java/vortex-spark/common/src/test/scala/dev/vortex/spark/read/VortexListFilterTest.scala new file mode 100644 index 00000000000..16c6840442f --- /dev/null +++ b/java/vortex-spark/common/src/test/scala/dev/vortex/spark/read/VortexListFilterTest.scala @@ -0,0 +1,167 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +package dev.vortex.spark.read + +import java.util.Arrays + +import org.apache.spark.sql.catalyst.expressions._ +import org.apache.spark.sql.sources.Filter +import org.apache.spark.sql.types._ +import org.junit.jupiter.api.Assertions._ +import org.junit.jupiter.api.Test + +import dev.vortex.relocated.org.apache.arrow.vector.types.TimeUnit +import dev.vortex.relocated.org.apache.arrow.vector.types.pojo.{ + ArrowType, + Field, + FieldType +} + +final class VortexListFilterTest { + private def arrayField(name: String, element: Field): Field = + new Field( + name, + FieldType.nullable(ArrowType.List.INSTANCE), + Arrays.asList(element) + ) + + @Test + def executorBuildsListPredicate(): Unit = { + // A dot here is part of the field name, not a nested path. + val arrayType = ArrayType(StringType, true) + val schema = new StructType().add("tag.values", arrayType) + val attribute = AttributeReference("tag.values", arrayType)() + val filter = VortexListFilter(ArrayContains(attribute, Literal("sports"))) + val fields = Arrays.asList( + arrayField( + "tag.values", + Field.nullable("element", ArrowType.Utf8.INSTANCE) + ) + ) + assertTrue( + VortexPartitionReader + .buildFilterExpression( + Array.empty[Filter], + Array[NativePredicate](filter), + schema, + fields + ) + .isPresent + ) + } + + @Test + def executorChecksNestedPhysicalTypes(): Unit = { + val element = new StructType().add("price", IntegerType) + val arrayType = ArrayType(element, true) + val schema = new StructType().add("items", arrayType) + val attribute = AttributeReference("items", arrayType)() + val parameter = NamedLambdaVariable("x", element, true) + val predicate = ArrayExists( + attribute, + LambdaFunction( + GreaterThan(GetStructField(parameter, 0), Literal(100)), + Seq(parameter) + ) + ) + val filter = VortexListFilter(predicate) + def fields(price: Field) = Arrays.asList( + arrayField( + "items", + new Field( + "element", + FieldType.nullable(ArrowType.Struct.INSTANCE), + Arrays.asList(price) + ) + ) + ) + assertTrue( + VortexPartitionReader + .buildFilterExpression( + Array.empty[Filter], + Array[NativePredicate](filter), + schema, + fields(Field.nullable("price", new ArrowType.Int(32, true))) + ) + .isPresent + ) + assertFalse( + VortexPartitionReader + .buildFilterExpression( + Array.empty[Filter], + Array[NativePredicate](filter), + schema, + fields(Field.nullable("price", new ArrowType.Int(64, true))) + ) + .isPresent + ) + } + + @Test + def timestampUnitsRemainCheckedInsideLists(): Unit = { + val arrayType = ArrayType(TimestampType, true) + val schema = new StructType().add("times", arrayType) + val fields = Arrays.asList( + arrayField( + "times", + Field.nullable( + "element", + new ArrowType.Timestamp(TimeUnit.MILLISECOND, "UTC") + ) + ) + ) + assertEquals( + NullType, + ReadTypes + .filterable(schema, fields, Arrays.asList(Array("times"))) + .apply("times") + .dataType + ) + } + + @Test + def floatingPointComparisonsStayInSpark(): Unit = { + val arrayType = ArrayType(DoubleType, true) + val schema = + new StructType().add("values", arrayType).add("needle", DoubleType) + val values = AttributeReference("values", arrayType)() + val needle = AttributeReference("needle", DoubleType)() + assertTrue( + VortexListFilter.from(ArrayContains(values, needle), schema).isEmpty + ) + val parameter = NamedLambdaVariable("x", DoubleType, true) + assertTrue( + VortexListFilter + .from( + ArrayExists( + values, + LambdaFunction(EqualTo(parameter, parameter), Seq(parameter)) + ), + schema + ) + .isEmpty + ) + } + + @Test + def correlatedLambdasStayInSpark(): Unit = { + val arrayType = ArrayType(IntegerType, true) + val schema = + new StructType().add("values", arrayType).add("threshold", IntegerType) + val values = AttributeReference("values", arrayType)() + val threshold = AttributeReference("threshold", IntegerType)() + val parameter = NamedLambdaVariable("x", IntegerType, true) + assertTrue( + VortexListFilter + .from( + ArrayExists( + values, + LambdaFunction(GreaterThan(parameter, threshold), Seq(parameter)) + ), + schema + ) + .isEmpty + ) + } +} From 0f59c77f9e97414e70ab141d6f71db57ba372727 Mon Sep 17 00:00:00 2001 From: Christopher Wirt Date: Wed, 11 Nov 2026 16:47:52 +0000 Subject: [PATCH 2/2] fix(spark): keep nested V1 pushdown and simplify list filter plumbing - Emit nested field paths from list predicates instead of special-casing whole structs in the shared ReadTypes.filterable, which had stopped `s.a` filters being pushed alongside Spark's inferred `isnotnull(s)`. - Share the comparable-type rule between V1 literals and list filters, without allocating native expressions to check a type. - Resolve filterable columns once per file and AND all converted filters in one native call. - Rename the plumbing to nativePredicates and tidy tests. Co-Authored-By: Claude Opus 5.5 --- .../java/dev/vortex/spark/read/ReadTypes.java | 4 +- .../read/SparkFilterToVortexExpression.java | 29 ++++-- .../spark/read/VortexPartitionReader.java | 39 ++++---- .../read/VortexPartitionReaderFactory.java | 8 +- .../vortex/spark/read/VortexListFilter.scala | 95 +++++++++++-------- .../dev/vortex/spark/read/VortexScan.scala | 6 +- .../spark/VortexFilterPushdownTest.java | 5 +- .../dev/vortex/spark/read/ReadTypesTest.java | 13 +++ .../spark/read/VortexListFilterTest.scala | 70 +++++++++----- 9 files changed, 171 insertions(+), 98 deletions(-) diff --git a/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/ReadTypes.java b/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/ReadTypes.java index 054165d4815..cf09a366802 100644 --- a/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/ReadTypes.java +++ b/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/ReadTypes.java @@ -214,9 +214,7 @@ private static StructType filterable(StructType schema, List fileFields, if (!(fileField.getType() instanceof ArrowType.Struct)) { continue; } - type = entry.getValue().stream().anyMatch(path -> path.length == depth + 1) - ? (literalType(fileField, struct) ? struct : DataTypes.NullType) - : filterable(struct, fileField.getChildren(), entry.getValue(), depth + 1); + type = filterable(struct, fileField.getChildren(), entry.getValue(), depth + 1); } else { type = literalType(fileField, field.dataType()) ? field.dataType() : DataTypes.NullType; } diff --git a/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/SparkFilterToVortexExpression.java b/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/SparkFilterToVortexExpression.java index ee2cc82ec63..4be939023a5 100644 --- a/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/SparkFilterToVortexExpression.java +++ b/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/SparkFilterToVortexExpression.java @@ -44,8 +44,6 @@ import org.apache.spark.sql.types.DateType; import org.apache.spark.sql.types.Decimal; import org.apache.spark.sql.types.DecimalType; -import org.apache.spark.sql.types.DoubleType; -import org.apache.spark.sql.types.FloatType; import org.apache.spark.sql.types.IntegerType; import org.apache.spark.sql.types.LongType; import org.apache.spark.sql.types.ShortType; @@ -332,7 +330,29 @@ private static StructField findField(StructType schema, String name) { return null; } + /** + * Whether the native scan compares values of {@code dataType} as Spark does. Vortex compares floats in total order, + * so -0.0 and 0.0 differ, while Spark treats them as equal. Spark cannot restore rows that Vortex drops, so these + * comparisons stay in Spark (see #10064). + */ + static boolean comparableType(DataType dataType) { + return dataType instanceof BooleanType + || dataType instanceof ByteType + || dataType instanceof ShortType + || dataType instanceof IntegerType + || dataType instanceof LongType + || dataType instanceof StringType + || dataType instanceof BinaryType + || dataType instanceof DateType + || dataType instanceof TimestampType + || dataType instanceof TimestampNTZType + || dataType instanceof DecimalType; + } + static Optional literal(Object value, DataType dataType) { + if (!comparableType(dataType)) { + return Optional.empty(); + } if (dataType instanceof BooleanType) { return value == null ? Optional.of(Expression.nullLiteralBool()) @@ -352,11 +372,6 @@ static Optional literal(Object value, DataType dataType) { if (dataType instanceof LongType) { return numericLiteral(value, Expression.DType.I64, number -> Expression.literal(number.longValue())); } - if (dataType instanceof FloatType || dataType instanceof DoubleType) { - // Vortex compares floats in total order, so -0.0 and 0.0 differ, while Spark treats them as equal. - // Spark cannot restore rows that Vortex drops, so keep these comparisons in Spark (see #10064). - return Optional.empty(); - } if (dataType instanceof StringType) { if (value == null) { return Optional.of(Expression.nullLiteral(Expression.DType.UTF8)); diff --git a/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/VortexPartitionReader.java b/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/VortexPartitionReader.java index 8376a161ab8..af16afea191 100644 --- a/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/VortexPartitionReader.java +++ b/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/VortexPartitionReader.java @@ -98,7 +98,7 @@ public VortexPartitionReader( VortexIo io, VortexOptions formatOptions, Filter[] pushedFilters, - NativePredicate[] listFilters, + NativePredicate[] nativePredicates, boolean caseSensitive) { this.file = file; this.readDataSchema = readDataSchema; @@ -153,7 +153,7 @@ public VortexPartitionReader( // Filters are converted against this file's own columns. Spark evaluates every data filter above the // scan anyway, so one that reads a column this file lacks, or compares one it stores in another type, is // dropped rather than pushed. - buildFilterExpression(pushedFilters, listFilters, dataSchema, arrowFields) + buildFilterExpression(pushedFilters, nativePredicates, dataSchema, arrowFields) .ifPresent(options::filter); Optional rows = splitRows(file, dataSource.rowCount()); if (rows.isPresent() && rows.get().count() == 0) { @@ -224,24 +224,31 @@ private Map fieldNames(StructType fileSchema, boolean caseSensit } static Optional buildFilterExpression( - Filter[] filters, NativePredicate[] listFilters, StructType dataSchema, List fileFields) { + Filter[] filters, NativePredicate[] nativePredicates, StructType dataSchema, List fileFields) { filters = filters == null ? new Filter[0] : filters; - StructType filterable = ReadTypes.filterable(dataSchema, fileFields, filters); - Expression combined = null; + if (filters.length == 0 && nativePredicates.length == 0) { + return Optional.empty(); + } + List paths = new ArrayList<>(); for (Filter filter : filters) { - Optional converted = SparkFilterToVortexExpression.convert(filter, filterable); - if (converted.isPresent()) { - combined = combined == null ? converted.get() : Expression.and(combined, converted.get()); - } + paths.addAll(Arrays.asList(filter.v2references())); } - for (NativePredicate filter : listFilters) { - StructType listSchema = ReadTypes.filterable(dataSchema, fileFields, Arrays.asList(filter.references())); - Optional converted = filter.convert(listSchema); - if (converted.isPresent()) { - combined = combined == null ? converted.get() : Expression.and(combined, converted.get()); - } + for (NativePredicate predicate : nativePredicates) { + paths.addAll(Arrays.asList(predicate.references())); + } + StructType filterable = ReadTypes.filterable(dataSchema, fileFields, paths); + List converted = new ArrayList<>(); + for (Filter filter : filters) { + SparkFilterToVortexExpression.convert(filter, filterable).ifPresent(converted::add); + } + for (NativePredicate predicate : nativePredicates) { + predicate.convert(filterable).ifPresent(converted::add); } - return Optional.ofNullable(combined); + return switch (converted.size()) { + case 0 -> Optional.empty(); + case 1 -> Optional.of(converted.get(0)); + default -> Optional.of(Expression.and(converted.toArray(new Expression[0]))); + }; } @Override diff --git a/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/VortexPartitionReaderFactory.java b/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/VortexPartitionReaderFactory.java index 8836129e74a..58c2f6abc94 100644 --- a/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/VortexPartitionReaderFactory.java +++ b/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/VortexPartitionReaderFactory.java @@ -39,7 +39,7 @@ public final class VortexPartitionReaderFactory extends FilePartitionReaderFacto private final StructType readDataSchema; private final StructType readPartitionSchema; private final Filter[] pushedFilters; - private final NativePredicate[] listFilters; + private final NativePredicate[] nativePredicates; private final boolean caseSensitive; public VortexPartitionReaderFactory( @@ -50,7 +50,7 @@ public VortexPartitionReaderFactory( StructType readDataSchema, StructType readPartitionSchema, Filter[] pushedFilters, - NativePredicate[] listFilters, + NativePredicate[] nativePredicates, boolean caseSensitive) { this.fileOptions = fileOptions; this.io = io; @@ -59,7 +59,7 @@ public VortexPartitionReaderFactory( this.readDataSchema = readDataSchema; this.readPartitionSchema = readPartitionSchema; this.pushedFilters = pushedFilters == null ? new Filter[0] : pushedFilters.clone(); - this.listFilters = listFilters.clone(); + this.nativePredicates = nativePredicates == null ? new NativePredicate[0] : nativePredicates.clone(); this.caseSensitive = caseSensitive; } @@ -87,7 +87,7 @@ public PartitionReader buildColumnarReader(PartitionedFile file) io, formatOptions, pushedFilters, - listFilters, + nativePredicates, caseSensitive); } diff --git a/java/vortex-spark/common/src/main/scala/dev/vortex/spark/read/VortexListFilter.scala b/java/vortex-spark/common/src/main/scala/dev/vortex/spark/read/VortexListFilter.scala index dcf6e1d40f5..9cbe19c16df 100644 --- a/java/vortex-spark/common/src/main/scala/dev/vortex/spark/read/VortexListFilter.scala +++ b/java/vortex-spark/common/src/main/scala/dev/vortex/spark/read/VortexListFilter.scala @@ -6,7 +6,7 @@ package dev.vortex.spark.read import java.util.Optional import org.apache.spark.sql.catalyst.expressions._ -import org.apache.spark.sql.types.StructType +import org.apache.spark.sql.types.{DataType, StringType, StructType} import dev.vortex.api.{Expression => VortexExpression} import dev.vortex.api.Expression.BinaryOp @@ -17,13 +17,10 @@ import dev.vortex.api.Expression.BinaryOp final case class VortexListFilter(predicate: Expression) extends NativePredicate { override def references(): Array[Array[String]] = - predicate.references.toSeq.map(attribute => Array(attribute.name)).toArray + VortexListFilter.columnPaths(predicate).map(_.toArray).toArray def convert(schema: StructType): Optional[VortexExpression] = - VortexListFilter.translate(predicate, schema) match { - case Some(expression) => Optional.of(expression) - case None => Optional.empty() - } + Optional.ofNullable(VortexListFilter.translate(predicate, schema).orNull) } object VortexListFilter { @@ -37,6 +34,39 @@ object VortexListFilter { case _ => Seq.empty } + /** Nested field names of a column read, e.g. `s.tags` as `Seq("s", "tags")`. + */ + private def columnPath(expression: Expression): Option[Seq[String]] = + expression match { + case attribute: AttributeReference => Some(Seq(attribute.name)) + case field: GetStructField => + columnPath(field.child).map(_ :+ field.childSchema(field.ordinal).name) + case _ => None + } + + private def columnPaths(expression: Expression): Seq[Seq[String]] = + columnPath(expression) match { + case Some(path) => Seq(path) + case None => expression.children.flatMap(columnPaths) + } + + private object ColumnPath { + def unapply(expression: Expression): Option[Seq[String]] = + columnPath(expression) + } + + private def typeAt(schema: StructType, path: Seq[String]): Option[DataType] = + path.foldLeft(Option[DataType](schema)) { + case (Some(struct: StructType), name) => + struct.fields.find(_.name == name).map(_.dataType) + case _ => None + } + + /** Non-binary string collations compare differently in the native scan. */ + private def comparable(dataType: DataType): Boolean = + SparkFilterToVortexExpression.comparableType(dataType) && + (!dataType.isInstanceOf[StringType] || dataType == StringType) + private def translate( expression: Expression, schema: StructType, @@ -44,41 +74,32 @@ object VortexListFilter { ): Option[VortexExpression] = { def child(e: Expression): Option[VortexExpression] = translate(e, schema, variable) - def comparable(e: Expression): Boolean = - SparkFilterToVortexExpression.literal(null, e.dataType).isPresent && - (!e.dataType.isInstanceOf[org.apache.spark.sql.types.StringType] || - e.dataType == org.apache.spark.sql.types.DataTypes.StringType) def binary(left: Expression, right: Expression, op: BinaryOp) = - for { - _ <- if (comparable(left) && comparable(right)) Some(()) else None - l <- child(left) - r <- child(right) - } yield VortexExpression.binary(op, l, r) + if (!comparable(left.dataType) || !comparable(right.dataType)) None + else + for { l <- child(left); r <- child(right) } yield VortexExpression + .binary(op, l, r) expression match { - case attribute: AttributeReference - if variable.isEmpty && schema.fields.exists(field => - field.name == attribute.name && field.dataType == attribute.dataType - ) => - Some(VortexExpression.column(attribute.name)) + case column @ ColumnPath(path) => + // Lambdas may read only their own parameter. + if (variable.isEmpty && typeAt(schema, path).contains(column.dataType)) + Some(VortexExpression.column(path.toArray)) + else None case parameter: NamedLambdaVariable if variable.exists(_.exprId == parameter.exprId) => Some(VortexExpression.variable(parameter.exprId.toString)) case field: GetStructField => child(field.child).map( - VortexExpression.getItem( - field.child.dataType - .asInstanceOf[StructType] - .fields(field.ordinal) - .name, - _ - ) + VortexExpression.getItem(field.childSchema(field.ordinal).name, _) ) case literal: Literal => - val converted = - SparkFilterToVortexExpression.literal(literal.value, literal.dataType) - if (converted.isPresent) Some(converted.get()) else None - case ArrayContains(list, needle) if comparable(needle) => + Option( + SparkFilterToVortexExpression + .literal(literal.value, literal.dataType) + .orElse(null) + ) + case ArrayContains(list, needle) if comparable(needle.dataType) => for { l <- child(list) n <- child(needle) @@ -103,15 +124,11 @@ object VortexListFilter { case LessThan(left, right) => binary(left, right, BinaryOp.LT) case LessThanOrEqual(left, right) => binary(left, right, BinaryOp.LTE) case And(left, right) => - for { l <- child(left); r <- child(right) } yield VortexExpression.and( - l, - r - ) + for { l <- child(left); r <- child(right) } yield VortexExpression + .and(l, r) case Or(left, right) => - for { l <- child(left); r <- child(right) } yield VortexExpression.or( - l, - r - ) + for { l <- child(left); r <- child(right) } yield VortexExpression + .or(l, r) case Not(value) => child(value).map(VortexExpression.not) case IsNull(value) => child(value).map(VortexExpression.isNull) case IsNotNull(value) => child(value).map(VortexExpression.isNotNull) diff --git a/java/vortex-spark/common/src/main/scala/dev/vortex/spark/read/VortexScan.scala b/java/vortex-spark/common/src/main/scala/dev/vortex/spark/read/VortexScan.scala index 97755a2aedd..231c224c587 100644 --- a/java/vortex-spark/common/src/main/scala/dev/vortex/spark/read/VortexScan.scala +++ b/java/vortex-spark/common/src/main/scala/dev/vortex/spark/read/VortexScan.scala @@ -55,7 +55,7 @@ final class VortexScan( override val dataFilters: Seq[Expression] ) extends FileScan { - private lazy val listFilters: Array[NativePredicate] = + private lazy val nativePredicates: Array[NativePredicate] = dataFilters.flatMap(VortexListFilter.from(_, dataSchema)).toArray private val caseSensitiveOptions = options.asCaseSensitiveMap.asScala.toMap @@ -99,7 +99,7 @@ final class VortexScan( readDataSchema, readPartitionSchema, pushedFilters, - listFilters, + nativePredicates, caseSensitive ) } @@ -167,7 +167,7 @@ final class VortexScan( .getOrElse("[]") super.getMetaData() ++ Map( "PushedFilters" -> seqToString(pushedFilters.toSeq), - "PushedListFilters" -> seqToString(listFilters.toSeq), + "PushedNativePredicates" -> seqToString(nativePredicates.toSeq), "PushedAggregation" -> aggregation, "PushedGroupBy" -> groupBy ) diff --git a/java/vortex-spark/common/src/test/java/dev/vortex/spark/VortexFilterPushdownTest.java b/java/vortex-spark/common/src/test/java/dev/vortex/spark/VortexFilterPushdownTest.java index fb27bda69f8..629a1e9f0dd 100644 --- a/java/vortex-spark/common/src/test/java/dev/vortex/spark/VortexFilterPushdownTest.java +++ b/java/vortex-spark/common/src/test/java/dev/vortex/spark/VortexFilterPushdownTest.java @@ -120,9 +120,8 @@ void listPredicatesArePushedIntoV2Scans(boolean threeValuedLogic, String predica source.where(predicate).orderBy("id").collectAsList(), filtered.orderBy("id").collectAsList(), predicate); - assertTrue( - filtered.queryExecution().executedPlan().toString().contains("VortexListFilter"), - filtered.queryExecution().executedPlan().toString()); + String plan = filtered.queryExecution().executedPlan().toString(); + assertTrue(plan.contains("VortexListFilter"), plan); } static Stream listPredicateCases() { diff --git a/java/vortex-spark/common/src/test/java/dev/vortex/spark/read/ReadTypesTest.java b/java/vortex-spark/common/src/test/java/dev/vortex/spark/read/ReadTypesTest.java index 695abb48b30..2554caa929e 100644 --- a/java/vortex-spark/common/src/test/java/dev/vortex/spark/read/ReadTypesTest.java +++ b/java/vortex-spark/common/src/test/java/dev/vortex/spark/read/ReadTypesTest.java @@ -52,6 +52,19 @@ void filtersSeeOnlyTheColumnsTheyReference() { "struct<>", ReadTypes.filterable(SCHEMA, FILE, new Filter[0]).catalogString()); } + @Test + void structNullCheckKeepsNestedFieldFilters() { + // A file missing st.b must still push down st.a alongside Spark's inferred isnotnull(st). + List file = List.of(new Field( + "st", + FieldType.nullable(ArrowType.Struct.INSTANCE), + List.of(Field.nullable("a", new ArrowType.Int(64, true))))); + Filter[] filters = {new IsNotNull("st"), new GreaterThan("st.a", 1L)}; + StructType filterable = ReadTypes.filterable(SCHEMA, file, filters); + assertTrue(SparkFilterToVortexExpression.isPushable(filters[0], filterable)); + assertTrue(SparkFilterToVortexExpression.isPushable(filters[1], filterable)); + } + @Test void nullChecksArePushedOnATimestampInAnotherUnit() { Filter[] filters = { diff --git a/java/vortex-spark/common/src/test/scala/dev/vortex/spark/read/VortexListFilterTest.scala b/java/vortex-spark/common/src/test/scala/dev/vortex/spark/read/VortexListFilterTest.scala index 16c6840442f..8434aa11441 100644 --- a/java/vortex-spark/common/src/test/scala/dev/vortex/spark/read/VortexListFilterTest.scala +++ b/java/vortex-spark/common/src/test/scala/dev/vortex/spark/read/VortexListFilterTest.scala @@ -26,6 +26,15 @@ final class VortexListFilterTest { Arrays.asList(element) ) + private def pushed( + filter: NativePredicate, + schema: StructType, + fields: java.util.List[Field] + ): Boolean = + VortexPartitionReader + .buildFilterExpression(Array.empty[Filter], Array(filter), schema, fields) + .isPresent + @Test def executorBuildsListPredicate(): Unit = { // A dot here is part of the field name, not a nested path. @@ -40,14 +49,35 @@ final class VortexListFilterTest { ) ) assertTrue( - VortexPartitionReader - .buildFilterExpression( - Array.empty[Filter], - Array[NativePredicate](filter), - schema, - fields + pushed(filter, schema, fields) + ) + } + + @Test + def listInsideStructIsCheckedByItsOwnPath(): Unit = { + val tags = ArrayType(StringType, true) + val struct = new StructType().add("tags", tags).add("other", IntegerType) + val schema = new StructType().add("s", struct) + val attribute = AttributeReference("s", struct)() + val filter = VortexListFilter( + ArrayContains(GetStructField(attribute, 0), Literal("sports")) + ) + assertArrayEquals( + Array[AnyRef](Array("s", "tags")), + filter.references().asInstanceOf[Array[AnyRef]] + ) + // The file lacks s.other, which the predicate never reads. + val fields = Arrays.asList( + new Field( + "s", + FieldType.nullable(ArrowType.Struct.INSTANCE), + Arrays.asList( + arrayField("tags", Field.nullable("element", ArrowType.Utf8.INSTANCE)) ) - .isPresent + ) + ) + assertTrue( + pushed(filter, schema, fields) ) } @@ -77,24 +107,18 @@ final class VortexListFilterTest { ) ) assertTrue( - VortexPartitionReader - .buildFilterExpression( - Array.empty[Filter], - Array[NativePredicate](filter), - schema, - fields(Field.nullable("price", new ArrowType.Int(32, true))) - ) - .isPresent + pushed( + filter, + schema, + fields(Field.nullable("price", new ArrowType.Int(32, true))) + ) ) assertFalse( - VortexPartitionReader - .buildFilterExpression( - Array.empty[Filter], - Array[NativePredicate](filter), - schema, - fields(Field.nullable("price", new ArrowType.Int(64, true))) - ) - .isPresent + pushed( + filter, + schema, + fields(Field.nullable("price", new ArrowType.Int(64, true))) + ) ) }