feat: expose Spark scan schema and pushed predicates

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.
+     *
+     * <p>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.
+     *
+     * <p>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() {

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);
+    }
+}