GEOMESA-3582 Added connector-configurable BBOX-containment short-circuiting logic to the plugin connector, which shortcuts the more expensive WKB decode during row-filtering, provides additional performance improvement for coarsely-pruned datasets.

Author: cwdobbins

geomesa-trino/geomesa-trino-plugin/src/main/java/org/locationtech/geomesa/trino/spatial/iceberg/connector/BboxRejectingPageSource.java

@@ -0,0 +1,160 @@
+/***********************************************************************
+ * Copyright (c) 2013-2025 General Atomics Integrated Intelligence, Inc.
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Apache License, Version 2.0
+ * which accompanies this distribution and is available at
+ * https://www.apache.org/licenses/LICENSE-2.0
+ ***********************************************************************/
+
+package org.locationtech.geomesa.trino.spatial.iceberg.connector;
+
+import io.trino.spi.Page;
+import io.trino.spi.block.Block;
+import io.trino.spi.connector.ConnectorPageSource;
+import io.trino.spi.connector.SourcePage;
+import io.trino.spi.metrics.Metrics;
+import io.trino.spi.type.RealType;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.OptionalLong;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.ObjLongConsumer;
+
+/**
+ * Wraps the Iceberg page source with a sound, pre-decode <em>cheap-reject</em>: for a spatial
+ * query the connector has already injected the query envelope as REAL domains on a geometry's
+ * {@code __X_bbox__} sub-fields, and those domains ride to the worker in the table handle's
+ * unenforced predicate. This page source reads only those cheap bbox columns and drops any row
+ * whose bbox cannot overlap the envelope — a necessary condition for any intersection-implying
+ * predicate — so the engine's exact {@code ST_Intersects}/{@code ST_Within} residual never
+ * decodes those rows' geometry WKB.
+ *
+ * <p><strong>Correctness.</strong> Only rows that <em>fail</em> a necessary condition are
+ * dropped, so no matching row is ever removed; the exact predicate still runs on every survivor
+ * (the engine's residual is untouched — this is a pre-filter, not a replacement). Null bbox
+ * values are kept (cannot prove non-overlap). Bounds come from the domains'
+ * {@link io.trino.spi.predicate.SortedRangeSet#getSpan span}, so a disjunction (OR) of envelopes
+ * degrades to their bounding box — still a superset, still sound.
+ *
+ * <p><strong>Laziness.</strong> Only the bbox columns are materialized here; the drop is applied
+ * via {@link SourcePage#selectPositions} <em>before</em> the geometry column is read, so the
+ * engine decodes WKB only for surviving rows. The bbox columns this page source added to the
+ * physical read (beyond what the query projected) are stripped from the returned page.
+ */
+final class BboxRejectingPageSource implements ConnectorPageSource {
+
+    /** One injected bbox-envelope bound: the row's value in {@code channel} must lie within
+     *  {@code [low, high]} (either side may be infinite) or the row cannot overlap. */
+    record BboxBound(int channel, float low, float high) {}
+
+    private final ConnectorPageSource delegate;
+    private final List<BboxBound> bounds;
+    /** Number of channels the query actually requested; any beyond this were added to read the
+     *  bbox sub-fields and must be hidden from the engine. */
+    private final int outputChannelCount;
+    private final boolean stripAddedChannels;
+
+    BboxRejectingPageSource(ConnectorPageSource delegate, int outputChannelCount,
+                            boolean stripAddedChannels, List<BboxBound> bounds) {
+        this.delegate = delegate;
+        this.outputChannelCount = outputChannelCount;
+        this.stripAddedChannels = stripAddedChannels;
+        this.bounds = bounds;
+    }
+
+    @Override
+    public SourcePage getNextSourcePage() {
+        SourcePage page = delegate.getNextSourcePage();
+        if (page == null) {
+            return null;
+        }
+        int positions = page.getPositionCount();
+
+        // Materialize ONLY the cheap bbox columns (geometry stays lazy).
+        Block[] boundBlocks = new Block[bounds.size()];
+        for (int i = 0; i < bounds.size(); i++) {
+            boundBlocks[i] = page.getBlock(bounds.get(i).channel());
+        }
+
+        int[] retained = new int[positions];
+        int kept = 0;
+        for (int p = 0; p < positions; p++) {
+            if (overlaps(p, boundBlocks)) {
+                retained[kept++] = p;
+            }
+        }
+
+        // Apply the drop BEFORE the geometry column is read, so survivors alone decode WKB.
+        if (kept < positions) {
+            page.selectPositions(retained, 0, kept);
+        }
+        return stripAddedChannels ? new ChannelPrefixSourcePage(page, outputChannelCount) : page;
+    }
+
+    /** True iff the row's bbox could overlap the query envelope (every bound satisfied); a null
+     *  bbox value cannot prove non-overlap, so it does not reject. */
+    private boolean overlaps(int position, Block[] boundBlocks) {
+        for (int i = 0; i < bounds.size(); i++) {
+            Block b = boundBlocks[i];
+            if (b.isNull(position)) {
+                continue;
+            }
+            float v = RealType.REAL.getFloat(b, position);
+            BboxBound bound = bounds.get(i);
+            if (v < bound.low() || v > bound.high()) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    /**
+     * Pure delegation
+     */
+    @Override public long getCompletedBytes() { return delegate.getCompletedBytes(); }
+    @Override public OptionalLong getCompletedPositions() { return delegate.getCompletedPositions(); }
+    @Override public long getReadTimeNanos() { return delegate.getReadTimeNanos(); }
+    @Override public boolean isFinished() { return delegate.isFinished(); }
+    @Override public long getMemoryUsage() { return delegate.getMemoryUsage(); }
+    @Override public void close() throws IOException { delegate.close(); }
+    @Override public CompletableFuture<?> isBlocked() { return delegate.isBlocked(); }
+    @Override public Metrics getMetrics() { return delegate.getMetrics(); }
+
+    /** A {@link SourcePage} view exposing only the first {@code channelCount} channels of a
+     *  delegate, hiding the bbox sub-field columns this page source added to the physical read. */
+    private static final class ChannelPrefixSourcePage implements SourcePage {
+        private final SourcePage delegate;
+        private final int channelCount;
+
+        ChannelPrefixSourcePage(SourcePage delegate, int channelCount) {
+            this.delegate = delegate;
+            this.channelCount = channelCount;
+        }
+
+        @Override public int getPositionCount() { return delegate.getPositionCount(); }
+        @Override public long getSizeInBytes() { return delegate.getSizeInBytes(); }
+        @Override public long getRetainedSizeInBytes() { return delegate.getRetainedSizeInBytes(); }
+        @Override public void retainedBytesForEachPart(ObjLongConsumer<Object> consumer) {
+            delegate.retainedBytesForEachPart(consumer);
+        }
+        @Override public int getChannelCount() { return channelCount; }
+        @Override public Block getBlock(int channel) {
+            if (channel >= channelCount) {
+                throw new IndexOutOfBoundsException("channel " + channel + " >= " + channelCount);
+            }
+            return delegate.getBlock(channel);
+        }
+        @Override public Page getPage() {
+            int[] channels = new int[channelCount];
+            for (int i = 0; i < channelCount; i++) {
+                channels[i] = i;
+            }
+            return delegate.getColumns(channels);
+        }
+        @Override public Page getColumns(int[] channels) { return delegate.getColumns(channels); }
+        @Override public void selectPositions(int[] positions, int offset, int size) {
+            delegate.selectPositions(positions, offset, size);
+        }
+    }
+}

geomesa-trino/geomesa-trino-plugin/src/main/java/org/locationtech/geomesa/trino/spatial/iceberg/connector/SpatialConnector.java

@@ -34,14 +34,15 @@ public class SpatialConnector implements Connector {
     private final Connector delegate;
     private final GeoMesaColumnCatalog geomCatalog;
     private final ConnectorAccessControl accessControl;
+    private final boolean bboxPageFilter;
 
     /**
      * Wraps a delegate connector with no Trino-layer visibility enforcement.
      *
      * @param delegate the underlying iceberg connector
      */
     public SpatialConnector(Connector delegate) {
-        this(delegate, null, null);
+        this(delegate, null, null, true);
     }
 
     /**
@@ -55,10 +56,26 @@ public SpatialConnector(Connector delegate) {
      *                    access control delegates to Iceberg (no extra filter).
      */
     public SpatialConnector(Connector delegate, String catalogName, AuthorizationResolver resolver) {
+        this(delegate, catalogName, resolver, true);
+    }
+
+    /**
+     * Wraps a delegate connector, optionally installing Trino-layer row-visibility enforcement
+     * and the page-source bbox cheap-reject.
+     *
+     * @param delegate       the underlying iceberg connector
+     * @param catalogName    the Trino catalog name; may be null when no resolver.
+     * @param resolver       identity→auths resolver; null disables Trino-layer enforcement.
+     * @param bboxPageFilter when true, wrap the page source with the bbox cheap-reject
+     *                       ({@link SpatialPageSourceProvider}).
+     */
+    public SpatialConnector(Connector delegate, String catalogName, AuthorizationResolver resolver,
+                            boolean bboxPageFilter) {
         this.delegate = delegate;
         this.geomCatalog = new GeoMesaColumnCatalog();
         this.accessControl = resolver == null ? null
             : new VisibilityAccessControl(catalogName, geomCatalog, resolver);
+        this.bboxPageFilter = bboxPageFilter;
     }
 
     /**
@@ -125,7 +142,8 @@ public ConnectorSplitManager getSplitManager() {
      */
     @Override
     public ConnectorPageSourceProvider getPageSourceProvider() {
-        return delegate.getPageSourceProvider();
+        ConnectorPageSourceProvider p = delegate.getPageSourceProvider();
+        return bboxPageFilter ? new SpatialPageSourceProvider(p) : p;
     }
 
     /**
@@ -135,7 +153,10 @@ public ConnectorPageSourceProvider getPageSourceProvider() {
      */
     @Override
     public ConnectorPageSourceProviderFactory getPageSourceProviderFactory() {
-        return delegate.getPageSourceProviderFactory();
+        ConnectorPageSourceProviderFactory factory = delegate.getPageSourceProviderFactory();
+        return bboxPageFilter
+            ? () -> new SpatialPageSourceProvider(factory.createPageSourceProvider())
+            : factory;
     }
 
     /**

geomesa-trino/geomesa-trino-plugin/src/main/java/org/locationtech/geomesa/trino/spatial/iceberg/connector/SpatialConnectorFactory.java

@@ -40,6 +40,17 @@ public class SpatialConnectorFactory implements ConnectorFactory {
     private static final String AUTH_RESOLVER   = SECURITY_PREFIX + "auth-resolver";
     private static final String AUTH_MAPPING    = SECURITY_PREFIX + "auth-mapping-file";
 
+
+    /** Enables page-source bbox cheap-reject (see {@link BboxRejectingPageSource}): drops rows
+     *  whose bbox can't overlap the query envelope before their geometry WKB is decoded. Sound and
+     *  safety-net-preserving. OFF by default: its value is inversely proportional to spatial
+     *  partition quality — on a well-Z2/XZ2-partitioned table file/row-group pruning already
+     *  removes most non-matching rows, so the extra bbox-column read can outweigh the saved
+     *  decodes. Enable ({@code geomesa.spatial.bbox-page-filter=true}) for coarsely-partitioned
+     *  tables (few, large files) where row-level rejection catches what file pruning misses. */
+    private static final String SPATIAL_PREFIX  = "geomesa.spatial.";
+    private static final String BBOX_PAGE_FILTER = SPATIAL_PREFIX + "bbox-page-filter";
+
     /** Default constructor; instances are created by {@link SpatialIcebergPlugin}. */
     public SpatialConnectorFactory() {}
 
@@ -65,16 +76,19 @@ public String getName() {
     public Connector create(String catalogName, Map<String, String> config,
                             ConnectorContext context) {
         AuthorizationResolver resolver = buildResolver(config);
+        boolean bboxPageFilter = Boolean.parseBoolean(config.getOrDefault(BBOX_PAGE_FILTER, "false"));
 
         // Iceberg uses strict config validation; strip our keys so it doesn't
         // reject them as unused.
         Map<String, String> icebergConfig = new LinkedHashMap<>();
-        config.forEach((k, v) -> { if (!k.startsWith(SECURITY_PREFIX)) icebergConfig.put(k, v); });
+        config.forEach((k, v) -> {
+            if (!k.startsWith(SECURITY_PREFIX) && !k.startsWith(SPATIAL_PREFIX)) icebergConfig.put(k, v);
+        });
 
         ConnectorFactory icebergFactory =
             new IcebergPlugin().getConnectorFactories().iterator().next();
         Connector icebergConnector = icebergFactory.create(catalogName, icebergConfig, context);
-        return new SpatialConnector(icebergConnector, catalogName, resolver);
+        return new SpatialConnector(icebergConnector, catalogName, resolver, bboxPageFilter);
     }
 
     /** Builds the identity→auths resolver from catalog config, or null when no