diff --git a/java/vortex-spark/src/test/java/dev/vortex/spark/read/VortexScanBuilderTest.java b/java/vortex-spark/src/test/java/dev/vortex/spark/read/VortexScanBuilderTest.java new file mode 100644 index 00000000000..9cc841b631b --- /dev/null +++ b/java/vortex-spark/src/test/java/dev/vortex/spark/read/VortexScanBuilderTest.java @@ -0,0 +1,176 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +package dev.vortex.spark.read; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.util.Map; +import org.apache.spark.sql.connector.catalog.Column; +import org.apache.spark.sql.connector.expressions.Expression; +import org.apache.spark.sql.connector.expressions.Expressions; +import org.apache.spark.sql.connector.expressions.LiteralValue; +import org.apache.spark.sql.connector.expressions.NamedReference; +import org.apache.spark.sql.connector.expressions.Transform; +import org.apache.spark.sql.connector.expressions.filter.Predicate; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.unsafe.types.UTF8String; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +/** + * Unit tests for {@link VortexScanBuilder}, focusing on how {@link VortexScanBuilder#pushPredicates(Predicate[])} + * splits predicates between Vortex pushdown and Spark post-scan evaluation. + * + *

Characterizes the partition-column exclusion rule (partition columns live in directory paths, not in the Vortex + * files, so predicates referencing them must stay with Spark), the handling of unsupported operators, and the defensive + * copy returned by {@link VortexScanBuilder#pushedPredicates()}. + */ +final class VortexScanBuilderTest { + + private static VortexScanBuilder builderWithColumns(Transform[] partitionTransforms, Column... columns) { + VortexScanBuilder builder = new VortexScanBuilder(Map.of(), partitionTransforms); + builder.addPath("/tmp/data.vortex"); + for (Column column : columns) { + builder.addColumn(column); + } + return builder; + } + + private static Predicate equality(String column, int value) { + return new Predicate("=", new Expression[] {ref(column), new LiteralValue<>(value, DataTypes.IntegerType)}); + } + + private static NamedReference ref(String name) { + return Expressions.column(name); + } + + // --- pushPredicates: data columns --- + + @Test + @DisplayName("Predicate on a data column is pushed and removed from post-scan predicates") + void dataColumnPredicateIsPushed() { + VortexScanBuilder builder = builderWithColumns( + new Transform[0], + Column.create("id", DataTypes.IntegerType), + Column.create("name", DataTypes.StringType)); + + Predicate predicate = equality("id", 42); + Predicate[] postScan = builder.pushPredicates(new Predicate[] {predicate}); + + assertEquals(0, postScan.length); + assertArrayEquals(new Predicate[] {predicate}, builder.pushedPredicates()); + } + + @Test + @DisplayName("Predicate on a column missing from the read schema is left to Spark") + void unknownColumnPredicateIsNotPushed() { + VortexScanBuilder builder = builderWithColumns(new Transform[0], Column.create("id", DataTypes.IntegerType)); + + Predicate predicate = equality("missing", 1); + Predicate[] postScan = builder.pushPredicates(new Predicate[] {predicate}); + + assertArrayEquals(new Predicate[] {predicate}, postScan); + assertEquals(0, builder.pushedPredicates().length); + } + + @Test + @DisplayName("Predicate with an unsupported operator is left to Spark") + void unsupportedOperatorIsNotPushed() { + VortexScanBuilder builder = builderWithColumns(new Transform[0], Column.create("name", DataTypes.StringType)); + + // LIKE with arbitrary user pattern is not a translatable V2 predicate name. + Predicate predicate = new Predicate( + "LIKE", + new Expression[] {ref("name"), new LiteralValue<>(UTF8String.fromString("%a%"), DataTypes.StringType)}); + Predicate[] postScan = builder.pushPredicates(new Predicate[] {predicate}); + + assertArrayEquals(new Predicate[] {predicate}, postScan); + assertEquals(0, builder.pushedPredicates().length); + } + + // --- pushPredicates: partition columns --- + + @Test + @DisplayName("Predicate on a partition column is left to Spark") + void partitionColumnPredicateIsNotPushed() { + Transform[] transforms = new Transform[] {Expressions.identity("year")}; + VortexScanBuilder builder = builderWithColumns( + transforms, Column.create("id", DataTypes.IntegerType), Column.create("year", DataTypes.IntegerType)); + + Predicate predicate = equality("year", 2024); + Predicate[] postScan = builder.pushPredicates(new Predicate[] {predicate}); + + assertArrayEquals(new Predicate[] {predicate}, postScan); + assertEquals(0, builder.pushedPredicates().length); + } + + @Test + @DisplayName("Mixed predicates split into pushed (data) and post-scan (partition)") + void mixedPredicatesAreSplit() { + Transform[] transforms = new Transform[] {Expressions.identity("year")}; + VortexScanBuilder builder = builderWithColumns( + transforms, Column.create("id", DataTypes.IntegerType), Column.create("year", DataTypes.IntegerType)); + + Predicate onData = equality("id", 7); + Predicate onPartition = equality("year", 2024); + Predicate[] postScan = builder.pushPredicates(new Predicate[] {onData, onPartition}); + + assertArrayEquals(new Predicate[] {onPartition}, postScan); + assertArrayEquals(new Predicate[] {onData}, builder.pushedPredicates()); + } + + @Test + @DisplayName("AND spanning a data column and a partition column is left to Spark as a whole") + void conjunctionSpanningPartitionColumnIsNotPushed() { + Transform[] transforms = new Transform[] {Expressions.identity("year")}; + VortexScanBuilder builder = builderWithColumns( + transforms, Column.create("id", DataTypes.IntegerType), Column.create("year", DataTypes.IntegerType)); + + Predicate and = + new org.apache.spark.sql.connector.expressions.filter.And(equality("id", 7), equality("year", 2024)); + Predicate[] postScan = builder.pushPredicates(new Predicate[] {and}); + + assertArrayEquals(new Predicate[] {and}, postScan); + assertEquals(0, builder.pushedPredicates().length); + } + + // --- pushedPredicates --- + + @Test + @DisplayName("pushedPredicates is empty before any pushPredicates call") + void pushedPredicatesEmptyByDefault() { + VortexScanBuilder builder = builderWithColumns(new Transform[0], Column.create("id", DataTypes.IntegerType)); + assertEquals(0, builder.pushedPredicates().length); + } + + @Test + @DisplayName("pushedPredicates returns a defensive copy") + void pushedPredicatesReturnsCopy() { + VortexScanBuilder builder = builderWithColumns(new Transform[0], Column.create("id", DataTypes.IntegerType)); + builder.pushPredicates(new Predicate[] {equality("id", 1)}); + + Predicate[] first = builder.pushedPredicates(); + Predicate[] second = builder.pushedPredicates(); + assertNotSame(first, second); + + // Mutating one returned array must not leak into subsequent calls. + first[0] = null; + assertEquals(1, builder.pushedPredicates().length); + assertNotNull(builder.pushedPredicates()[0]); + } + + // --- build --- + + @Test + @DisplayName("build without paths fails fast") + void buildWithoutPathsThrows() { + VortexScanBuilder builder = new VortexScanBuilder(Map.of()); + builder.addColumn(Column.create("id", DataTypes.IntegerType)); + assertThrows(IllegalStateException.class, builder::build); + } +}