Add FilterPushdown support to spark data source

java/build.gradle.kts

@@ -81,5 +81,10 @@ allprojects {
         }
     }
 
-    tasks.register("format").get().dependsOn("spotlessApply")
+    if (project.name == "vortex-spark_2.12") {
+        // vortex-spark_2.12 and vortex-spark_2.13 share a projectDir; format from the 2.13 variant only.
+        tasks.register("format") { enabled = false }
+    } else {
+        tasks.register("format").get().dependsOn("spotlessApply")
+    }
 }

java/vortex-jni/src/main/java/dev/vortex/api/Expression.java

@@ -73,6 +73,10 @@ public static Expression isNull(Expression child) {
         return new Expression(NativeExpression.isNull(child.nativePointer()));
     }
 
+    public static Expression isNotNull(Expression child) {
+        return new Expression(NativeExpression.isNotNull(child.nativePointer()));
+    }
+
     public static Expression literal(boolean value) {
         return new Expression(NativeExpression.literalBool(value, false));
     }

java/vortex-jni/src/main/java/dev/vortex/jni/NativeExpression.java

@@ -27,6 +27,8 @@ private NativeExpression() {}
 
     public static native long isNull(long childPointer);
 
+    public static native long isNotNull(long childPointer);
+
     public static native long literalBool(boolean value, boolean isNull);
 
     public static native long literalI8(byte value, boolean isNull);

java/vortex-spark/src/main/java/dev/vortex/spark/VortexDataSourceV2.java

@@ -20,6 +20,7 @@
 import org.apache.spark.sql.SparkSession;
 import org.apache.spark.sql.connector.catalog.Table;
 import org.apache.spark.sql.connector.catalog.TableProvider;
+import org.apache.spark.sql.connector.expressions.Expressions;
 import org.apache.spark.sql.connector.expressions.Transform;
 import org.apache.spark.sql.sources.DataSourceRegister;
 import org.apache.spark.sql.types.DataType;
@@ -118,6 +119,38 @@ public StructType inferSchema(CaseInsensitiveStringMap options) {
         return dataSchema;
     }
 
+    /**
+     * Infers partition transforms by inspecting Hive-style {@code key=value} segments in the first listed file path.
+     *
+     * <p>Spark calls this before {@link #getTable(StructType, Transform[], Map)} when the caller did not provide
+     * explicit partitioning. Returning identity transforms here lets downstream components (notably
+     * {@link dev.vortex.spark.read.VortexScanBuilder}) tell which schema columns are encoded in the directory layout
+     * rather than stored inside the Vortex files, which matters for predicate pushdown.
+     */
+    @Override
+    public Transform[] inferPartitioning(CaseInsensitiveStringMap options) {
+        var paths = getPaths(options);
+        if (paths.isEmpty()) {
+            return new Transform[0];
+        }
+        var formatOptions = buildDataSourceOptions(options.asCaseSensitiveMap());
+        String pathToInfer = Objects.requireNonNull(Iterables.getLast(paths));
+        if (!pathToInfer.endsWith(".vortex")) {
+            Optional<String> firstFile =
+                    NativeFiles.listFiles(VortexSparkSession.get(formatOptions), pathToInfer, formatOptions).stream()
+                            .findFirst();
+            if (firstFile.isEmpty()) {
+                return new Transform[0];
+            }
+            pathToInfer = firstFile.get();
+        }
+        Map<String, String> partitionValues = PartitionPathUtils.parsePartitionValues(pathToInfer);
+        if (partitionValues.isEmpty()) {
+            return new Transform[0];
+        }
+        return partitionValues.keySet().stream().map(Expressions::identity).toArray(Transform[]::new);
+    }
+
     /**
      * Creates a Vortex table instance with the given schema and properties.
      *

java/vortex-spark/src/main/java/dev/vortex/spark/VortexTable.java

@@ -58,7 +58,7 @@ public ScanBuilder newScanBuilder(CaseInsensitiveStringMap options) {
         Map<String, String> opts = Maps.newHashMap();
         opts.putAll(formatOptions);
         opts.putAll(options);
-        return new VortexScanBuilder(opts)
+        return new VortexScanBuilder(opts, partitionTransforms)
                 .addAllPaths(paths)
                 .addAllColumns(Arrays.asList(CatalogV2Util.structTypeToV2Columns(schema)));
     }