Fix spatial filter WKB rendering

vgi/src/main/java/farm/query/vgi/pushdown/PushdownFilters.java

@@ -229,7 +229,7 @@ private static String expressionSql(FilterExpression expression) {
             case FilterExpression.ColumnRef value -> quoteIdentifier(value.columnName());
             case FilterExpression.FieldRef value -> expressionSql(value.expression()) + "."
                     + quoteIdentifier(value.fieldName());
-            case FilterExpression.Literal value -> sqlLiteral(value.value());
+            case FilterExpression.Literal value -> literalSql(value);
             case FilterExpression.Comparison value -> "(" + expressionSql(value.left()) + " "
                     + value.op().symbol() + " " + expressionSql(value.right()) + ")";
             case FilterExpression.BooleanExpression value -> "(" + value.children().stream()
@@ -271,10 +271,24 @@ private static String functionName(Object function) {
             return standard.name().toLowerCase();
         }
         FilterIdentity identity = (FilterIdentity) function;
-        if (identity.equals(new FilterIdentity("duckdb.spatial", "intersects_extent", 1))) return "&&";
+        if (identity.equals(new FilterIdentity("duckdb.spatial", "intersects_extent", 1))) {
+            return "st_intersects_extent";
+        }
         return identity.namespace() + "." + identity.name();
     }
 
+    private static String literalSql(FilterExpression.Literal literal) {
+        if (isWkb(literal.field()) && literal.value() instanceof byte[] bytes) {
+            return "ST_GeomFromHEXWKB('" + java.util.HexFormat.of().formatHex(bytes) + "')";
+        }
+        return sqlLiteral(literal.value());
+    }
+
+    private static boolean isWkb(org.apache.arrow.vector.types.pojo.Field field) {
+        return field != null && field.getMetadata() != null
+                && "geoarrow.wkb".equals(field.getMetadata().get("ARROW:extension:name"));
+    }
+
     /**
      * Format the filters as a human-readable {@code AND}-joined SQL-like
      * string with values inlined. Used by diagnostic fixtures like

vgi/src/test/java/farm/query/vgi/pushdown/PushdownFiltersV2Test.java

@@ -253,6 +253,31 @@ void identitiesAreStrictEvenForAdvisoryPredicates() {
                 () -> new FilterIdentity("example.filters", "custom", 0));
     }
 
+    @Test
+    void spatialExpressionsRenderWkbAsGeometry() {
+        Field geometry = new Field("geom", new FieldType(true, new ArrowType.Binary(), null,
+                Map.of("ARROW:extension:name", "geoarrow.wkb",
+                        "ARROW:extension:metadata", "{}")), null);
+        Field literal = new Field("value_0", new FieldType(true, new ArrowType.Binary(), null,
+                Map.of("ARROW:extension:name", "geoarrow.wkb",
+                        "ARROW:extension:metadata", "{}")), null);
+        String expression = "{\"node\":\"call\",\"function\":{"
+                + "\"namespace\":\"duckdb.spatial\",\"name\":\"intersects_extent\",\"version\":1},"
+                + "\"arguments\":[" + column("geom")
+                + ",{\"node\":\"literal\",\"value_ref\":0}]}";
+        PushdownFiltersDecoder.Capabilities capabilities = new PushdownFiltersDecoder.Capabilities(
+                java.util.Set.of(new FilterIdentity("duckdb.spatial", "intersects_extent", 1)),
+                java.util.Set.of(), Map.of());
+        PushdownFilters filters = PushdownFiltersDecoder.decode(
+                batch(snapshot(predicate("spatial", "required", expression)),
+                        List.of(literal), List.of(new byte[] {1, 2, (byte) 0xff})),
+                new Schema(List.of(geometry)), List.of(), capabilities);
+
+        assertEquals(List.of("st_intersects_extent(\"geom\", "
+                        + "ST_GeomFromHEXWKB('0102ff'))"),
+                filters.expressionPredicates());
+    }
+
     @Test
     void configuredLimitsAndFullUint64RevisionsAreEnforced() {
         String longId = "x".repeat(129);