feat(spark): push list predicates into V2 native scans

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<Expression> convert(StructType schema);
+}

java/vortex-spark/common/src/main/java/dev/vortex/spark/read/ReadTypes.java

@@ -184,6 +184,10 @@ static StructType filterable(StructType schema, List<Field> fileFields, Filter[]
         for (Filter filter : filters) {
             paths.addAll(Arrays.asList(filter.v2references()));
         }
+        return filterable(schema, fileFields, paths);
+    }
+
+    static StructType filterable(StructType schema, List<Field> fileFields, List<String[]> paths) {
         return filterable(schema, fileFields, paths, 0);
     }
 
@@ -221,6 +225,24 @@ private static StructType filterable(StructType schema, List<Field> 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<String, Field> 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;
         }

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;
@@ -268,7 +266,29 @@ private static StructField findField(StructType schema, String name) {
         return null;
     }
 
-    private static Optional<Expression> literal(Object value, DataType dataType) {
+    /**
+     * 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<Expression> literal(Object value, DataType dataType) {
+        if (!comparableType(dataType)) {
+            return Optional.empty();
+        }
         if (dataType instanceof BooleanType) {
             return value == null
                     ? Optional.of(Expression.nullLiteralBool())
@@ -288,11 +308,6 @@ private static Optional<Expression> 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));

java/vortex-spark/common/src/main/java/dev/vortex/spark/read/VortexPartitionReader.java

@@ -87,6 +87,7 @@ public VortexPartitionReader(
                 io,
                 formatOptions,
                 pushedFilters,
+                new NativePredicate[0],
                 caseSensitive,
                 ignored -> {});
     }
@@ -99,6 +100,7 @@ public VortexPartitionReader(
             VortexIo io,
             VortexOptions formatOptions,
             Filter[] pushedFilters,
+            NativePredicate[] nativePredicates,
             boolean caseSensitive,
             LongConsumer bytesRead) {
         this.file = file;
@@ -154,7 +156,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, nativePredicates, dataSchema, arrowFields)
+                    .ifPresent(options::filter);
             Optional<SplitRows> 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.
@@ -223,20 +226,32 @@ private Map<String, String> fieldNames(StructType fileSchema, boolean caseSensit
         return names;
     }
 
-    private static Optional<Expression> buildFilterExpression(
-            Filter[] filters, StructType dataSchema, List<Field> fileFields) {
-        if (filters == null || filters.length == 0) {
+    static Optional<Expression> buildFilterExpression(
+            Filter[] filters, NativePredicate[] nativePredicates, StructType dataSchema, List<Field> fileFields) {
+        filters = filters == null ? new Filter[0] : filters;
+        if (filters.length == 0 && nativePredicates.length == 0) {
             return Optional.empty();
         }
-        StructType filterable = ReadTypes.filterable(dataSchema, fileFields, filters);
-        Expression combined = null;
+        List<String[]> paths = new ArrayList<>();
         for (Filter filter : filters) {
-            Optional<Expression> 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 predicate : nativePredicates) {
+            paths.addAll(Arrays.asList(predicate.references()));
+        }
+        StructType filterable = ReadTypes.filterable(dataSchema, fileFields, paths);
+        List<Expression> 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

java/vortex-spark/common/src/main/java/dev/vortex/spark/read/VortexPartitionReaderFactory.java

@@ -40,6 +40,7 @@ public final class VortexPartitionReaderFactory extends FilePartitionReaderFacto
     private final StructType readDataSchema;
     private final StructType readPartitionSchema;
     private final Filter[] pushedFilters;
+    private final NativePredicate[] nativePredicates;
     private final boolean caseSensitive;
 
     public VortexPartitionReaderFactory(
@@ -50,6 +51,7 @@ public VortexPartitionReaderFactory(
             StructType readDataSchema,
             StructType readPartitionSchema,
             Filter[] pushedFilters,
+            NativePredicate[] nativePredicates,
             boolean caseSensitive) {
         this.fileOptions = fileOptions;
         this.io = io;
@@ -58,6 +60,7 @@ public VortexPartitionReaderFactory(
         this.readDataSchema = readDataSchema;
         this.readPartitionSchema = readPartitionSchema;
         this.pushedFilters = pushedFilters == null ? new Filter[0] : pushedFilters.clone();
+        this.nativePredicates = nativePredicates == null ? new NativePredicate[0] : nativePredicates.clone();
         this.caseSensitive = caseSensitive;
     }
 
@@ -101,6 +104,7 @@ private PartitionReader<ColumnarBatch> buildColumnarReader(PartitionedFile file,
                 io,
                 formatOptions,
                 pushedFilters,
+                nativePredicates,
                 caseSensitive,
                 bytesRead);
     }

java/vortex-spark/common/src/main/scala/dev/vortex/spark/read/VortexListFilter.scala

@@ -0,0 +1,138 @@
+// 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.{DataType, StringType, 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]] =
+    VortexListFilter.columnPaths(predicate).map(_.toArray).toArray
+
+  def convert(schema: StructType): Optional[VortexExpression] =
+    Optional.ofNullable(VortexListFilter.translate(predicate, schema).orNull)
+}
+
+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
+    }
+
+  /** 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,
+      variable: Option[NamedLambdaVariable] = None
+  ): Option[VortexExpression] = {
+    def child(e: Expression): Option[VortexExpression] =
+      translate(e, schema, variable)
+    def binary(left: Expression, right: Expression, op: BinaryOp) =
+      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 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.childSchema(field.ordinal).name, _)
+        )
+      case literal: Literal =>
+        Option(
+          SparkFilterToVortexExpression
+            .literal(literal.value, literal.dataType)
+            .orElse(null)
+        )
+      case ArrayContains(list, needle) if comparable(needle.dataType) =>
+        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
+    }
+  }
+}

java/vortex-spark/common/src/main/scala/dev/vortex/spark/read/VortexScan.scala

@@ -56,6 +56,9 @@ final class VortexScan(
     override val dataFilters: Seq[Expression]
 ) extends FileScan {
 
+  private lazy val nativePredicates: Array[NativePredicate] =
+    dataFilters.flatMap(VortexListFilter.from(_, dataSchema)).toArray
+
   private val caseSensitiveOptions = options.asCaseSensitiveMap.asScala.toMap
   private val hadoopConf =
     sparkSession.sessionState.newHadoopConfWithOptions(caseSensitiveOptions)
@@ -97,6 +100,7 @@ final class VortexScan(
           readDataSchema,
           readPartitionSchema,
           pushedFilters,
+          nativePredicates,
           caseSensitive
         )
     }
@@ -167,6 +171,7 @@ final class VortexScan(
       .getOrElse("[]")
     super.getMetaData() ++ Map(
       "PushedFilters" -> seqToString(pushedFilters.toSeq),
+      "PushedNativePredicates" -> seqToString(nativePredicates.toSeq),
       "PushedAggregation" -> aggregation,
       "PushedGroupBy" -> groupBy
     )

java/vortex-spark/common/src/test/java/dev/vortex/spark/OneFileAtATimeSource.java

@@ -4,6 +4,7 @@
 package dev.vortex.spark;
 
 import dev.vortex.spark.io.VortexIo;
+import dev.vortex.spark.read.NativePredicate;
 import dev.vortex.spark.read.VortexBytesReadMetric;
 import dev.vortex.spark.read.VortexPartitionReaderFactory;
 import java.io.IOException;
@@ -172,6 +173,7 @@ private PartitionReader<ColumnarBatch> open(String path) {
                     SCHEMA,
                     new StructType(),
                     new Filter[0],
+                    new NativePredicate[0],
                     false);
             return factory.buildColumnarReader(partitioned);
         }