diff --git a/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexScan.java b/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexScan.java index 02a7563f925..6da2306ab69 100644 --- a/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexScan.java +++ b/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexScan.java @@ -76,6 +76,30 @@ public StructType readSchema() { return CatalogV2Util.v2ColumnsToStructType(readColumns.toArray(new Column[0])); } + /** + * Returns the table schema before projection pushdown. + * + *

An alternative execution engine must bind {@link #pushedPredicates()} against this schema. Columns used only + * by those predicates may be absent from {@link #readSchema()}. + * + * @return the full table schema + */ + public StructType tableSchema() { + return CatalogV2Util.v2ColumnsToStructType(tableColumns.toArray(new Column[0])); + } + + /** + * Returns the predicates this scan has promised to evaluate completely. + * + *

Spark may remove its own filter for these predicates. An alternative execution engine must enforce every + * returned predicate or retain this scan's original reader. + * + * @return a copy of the pushed predicate array + */ + public Predicate[] pushedPredicates() { + return pushedPredicates.clone(); + } + /** Logging-friendly readable description of the scan source. */ @Override public String description() { diff --git a/java/vortex-spark/src/test/java/dev/vortex/spark/read/VortexScanTest.java b/java/vortex-spark/src/test/java/dev/vortex/spark/read/VortexScanTest.java new file mode 100644 index 00000000000..a4eb6899708 --- /dev/null +++ b/java/vortex-spark/src/test/java/dev/vortex/spark/read/VortexScanTest.java @@ -0,0 +1,49 @@ +// 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 java.util.List; +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.filter.Predicate; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.StructType; +import org.junit.jupiter.api.Test; + +final class VortexScanTest { + @Test + void exposesPredicateColumnsEvenWhenProjectionIsEmpty() { + Column id = Column.create("id", DataTypes.IntegerType); + Predicate predicate = new Predicate("IS_NOT_NULL", new Expression[] {Expressions.column("id")}); + VortexScanBuilder builder = + new VortexScanBuilder(Map.of()).addPath("data.vortex").addColumn(id); + + assertEquals(0, builder.pushPredicates(new Predicate[] {predicate}).length); + builder.pruneColumns(new StructType()); + VortexScan scan = (VortexScan) builder.build(); + + assertEquals(0, scan.readSchema().size()); + assertArrayEquals(new String[] {"id"}, scan.tableSchema().fieldNames()); + assertEquals(DataTypes.IntegerType, scan.tableSchema().apply("id").dataType()); + assertArrayEquals(new Predicate[] {predicate}, scan.pushedPredicates()); + } + + @Test + void predicateArraysCannotReplaceTheScansPredicates() { + Predicate predicate = new Predicate("IS_NULL", new Expression[] {Expressions.column("id")}); + Predicate[] predicates = {predicate}; + VortexScan scan = new VortexScan(List.of(), List.of(), List.of(), predicates, Map.of()); + + predicates[0] = null; + scan.pushedPredicates()[0] = null; + + assertArrayEquals(new Predicate[] {predicate}, scan.pushedPredicates()); + assertEquals(0, new VortexScan(List.of(), List.of(), List.of(), null, Map.of()).pushedPredicates().length); + } +}