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/ReadTypes.java b/java/vortex-spark/common/src/main/java/dev/vortex/spark/read/ReadTypes.java index 2e60f0a6c22..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 @@ -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); } @@ -221,6 +225,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 ab23f77b3fd..24e052f3999 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; @@ -268,7 +266,29 @@ private static StructField findField(StructType schema, String name) { return null; } - private static Optional 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 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 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 2fd70f71202..9ce18b42fc6 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 @@ -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 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 fieldNames(StructType fileSchema, boolean caseSensit return names; } - private static Optional buildFilterExpression( - Filter[] filters, StructType dataSchema, List fileFields) { - if (filters == null || filters.length == 0) { + static Optional buildFilterExpression( + Filter[] filters, NativePredicate[] nativePredicates, StructType dataSchema, List 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 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 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 3a07ab2cec6..49b0b4041ae 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 @@ -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 buildColumnarReader(PartitionedFile file, io, formatOptions, pushedFilters, + nativePredicates, caseSensitive, bytesRead); } 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..9cbe19c16df --- /dev/null +++ b/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 + } + } +} 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 2a54df49300..d05e3618e0f 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 @@ -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 ) diff --git a/java/vortex-spark/common/src/test/java/dev/vortex/spark/OneFileAtATimeSource.java b/java/vortex-spark/common/src/test/java/dev/vortex/spark/OneFileAtATimeSource.java index 29e5ef6994f..1c8dcc4ac22 100644 --- a/java/vortex-spark/common/src/test/java/dev/vortex/spark/OneFileAtATimeSource.java +++ b/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 open(String path) { SCHEMA, new StructType(), new Filter[0], + new NativePredicate[0], false); return factory.buildColumnarReader(partitioned); } 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..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 @@ -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,41 @@ 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); + String plan = filtered.queryExecution().executedPlan().toString(); + assertTrue(plan.contains("VortexListFilter"), plan); + } + + 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/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 new file mode 100644 index 00000000000..8434aa11441 --- /dev/null +++ b/java/vortex-spark/common/src/test/scala/dev/vortex/spark/read/VortexListFilterTest.scala @@ -0,0 +1,191 @@ +// 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) + ) + + 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. + 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( + 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)) + ) + ) + ) + assertTrue( + pushed(filter, schema, fields) + ) + } + + @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( + pushed( + filter, + schema, + fields(Field.nullable("price", new ArrowType.Int(32, true))) + ) + ) + assertFalse( + pushed( + filter, + schema, + fields(Field.nullable("price", new ArrowType.Int(64, true))) + ) + ) + } + + @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 + ) + } +}