Cut allocation in GeoPackage reads: one row at a time, zero-copy geometry

CHANGELOG.md

@@ -14,6 +14,7 @@
 - `FloatingPointFormatter` has new default `writeOrdinates` methods that format a coordinate's 2-4 ordinates in one call. `DefaultFloatingPointFormatter` uses them to convert two short digit groups at once (both integral parts in degrees, both fractions in metres). The default metre (0-3 decimals) and degree (0-7 decimals) settings take specialised paths, metres with the fraction read from a table. Existing implementations keep working unchanged. Callers of the `byte[]` methods must leave room for the widest value, not just the bytes written: `DefaultFloatingPointFormatter` now stores whole 8-byte words and may write up to 18 bytes from the offset (37 for an x, y pair) whatever offset it returns. All writers in this repository already reserve more than that.
 - JSON-FG: the m ordinate of a 4D `geometry` (CRS84) is now written with the degree formatter like x, y and z. When `place` was in a projected CRS, m used to get the metre formatter's decimals.
 - CSV: WKT geometries with Z or ZM coordinates now separate their coordinates with commas (`LINESTRING Z(1 2 3,4 5 6)`, previously `LINESTRING Z(1 2 34 5 6)`).
+- `hakunapi-source-gpkg` reads rows one at a time into a reused container instead of batches of 250 (`BufferedResultSet` is removed from this module; `hakunapi-source-postgis` has its own copy). Geometry blobs are read through SQLite's own buffer instead of a `byte[]` per feature, and NULL checks are asked of SQLite by storage class. Output is unchanged.
 
 ## 2.0.0
 

src/hakunapi-source-gpkg/src/main/java/fi/nls/hakunapi/source/gpkg/GpkgSimpleFeatureProducer.java

@@ -23,12 +23,10 @@
 
 /**
  * Streaming implementation
- * Reads a batch of BATCH_SIZE rows at a time
  */
 public class GpkgSimpleFeatureProducer implements FeatureProducer {
 
     private static final Logger LOG = LoggerFactory.getLogger(GpkgSimpleFeatureProducer.class);
-    private static final int BATCH_SIZE = 250;
 
     @Override
     public FeatureStream getFeatures(GetFeatureRequest request, GetFeatureCollection col) throws Exception {
@@ -64,12 +62,13 @@ public FeatureStream getFeatures(GetFeatureRequest request, GetFeatureCollection
             c.setAutoCommit(false);
             ps = c.prepareStatement(query);
             GpkgQueryUtil.bind(c, ps, filters);
-            LOG.info("{}", ps.toString());
-            int bufSize = BATCH_SIZE;
-            ps.setFetchSize(bufSize);
+            // guarded because toString renders every bound value
+            if (LOG.isInfoEnabled()) {
+                LOG.info("{}", ps.toString());
+            }
             rs = ps.executeQuery();
             int numColsRs = rs.getMetaData().getColumnCount();
-            return new BufferedResultSet(c, ps, rs, numColsRs, mappers, bufSize);
+            return new ResultSetFeatureStream(c, ps, rs, numColsRs, mappers);
         } catch (Exception e) {
             U.closeSilent(rs);
             U.closeSilent(ps);
@@ -98,7 +97,9 @@ public int getNumberMatched(GetFeatureRequest request, GetFeatureCollection col)
         try (Connection c = ft.getDatabase().getConnection();
                 PreparedStatement ps = c.prepareStatement(query)) {
             GpkgQueryUtil.bind(c, ps, filters);
-            LOG.debug(ps.toString());
+            if (LOG.isDebugEnabled()) {
+                LOG.debug(ps.toString());
+            }
             try (ResultSet rs = ps.executeQuery()) {
                 if (!rs.next()) {
                     return -1;

src/hakunapi-source-gpkg/src/main/java/fi/nls/hakunapi/source/gpkg/BufferedResultSet.java → src/hakunapi-source-gpkg/src/main/java/fi/nls/hakunapi/source/gpkg/ResultSetFeatureStream.java

@@ -15,56 +15,49 @@
 import fi.nls.hakunapi.core.ValueProvider;
 import fi.nls.hakunapi.core.util.U;
 
-public class BufferedResultSet implements FeatureStream {
+/** Streams rows one at a time; the returned ValueProvider is valid until the next hasNext(). */
+public class ResultSetFeatureStream implements FeatureStream {
 
-    private static final Logger LOG = LoggerFactory.getLogger(BufferedResultSet.class);
+    private static final Logger LOG = LoggerFactory.getLogger(ResultSetFeatureStream.class);
 
     private final Connection c;
     private final PreparedStatement ps;
     private final ResultSet rs;
     private final ResultSetValueProvider valueProvider;
     private final List<ValueMapper> mappers;
-    private final ValueContainer[] buffer;
+    private final ValueContainer container;
 
-    private int i;
-    private int j;
+    private boolean ready;
     private boolean closed;
 
-    public BufferedResultSet(Connection c, PreparedStatement ps, ResultSet rs, int numColsRs, List<ValueMapper> mappers, int bufSize) {
+    public ResultSetFeatureStream(Connection c, PreparedStatement ps, ResultSet rs, int numColsRs,
+            List<ValueMapper> mappers) {
         this.c = c;
         this.ps = ps;
         this.rs = rs;
-        this.valueProvider = new ResultSetValueProvider(rs, numColsRs);
+        this.valueProvider = new ResultSetValueProvider(rs, numColsRs, SqliteBlobReader.of(ps));
         this.mappers = mappers;
-        this.buffer = new ValueContainer[bufSize];
+        this.container = new ObjectArrayValueContainer(mappers.size());
     }
 
     @Override
     public boolean hasNext() {
-        if (i < j) {
+        if (ready) {
             return true;
         }
         if (closed) {
             return false;
         }
-
-        i = 0;
-        j = 0;
         try {
-            for (; j < buffer.length; j++) {
-                if (!rs.next()) {
-                    close();
-                    break;
-                }
-                ValueContainer container = buffer[j];
-                if (container == null) {
-                    buffer[j] = container = new ObjectArrayValueContainer(mappers.size());
-                }
-                for (ValueMapper mapper : mappers) {
-                    mapper.accept(valueProvider, container);
-                }
+            if (!rs.next()) {
+                close();
+                return false;
+            }
+            for (ValueMapper mapper : mappers) {
+                mapper.accept(valueProvider, container);
             }
-            return j > 0;
+            ready = true;
+            return true;
         } catch (Exception e) {
             LOG.error("Failed to retrieve more features", e);
             close();
@@ -74,7 +67,8 @@ public boolean hasNext() {
 
     @Override
     public ValueProvider next() {
-        return buffer[i++];
+        ready = false;
+        return container;
     }
 
     @Override

src/hakunapi-source-gpkg/src/main/java/fi/nls/hakunapi/source/gpkg/ResultSetValueProvider.java

@@ -1,5 +1,6 @@
 package fi.nls.hakunapi.source.gpkg;
 
+import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.sql.Array;
 import java.sql.ResultSet;
@@ -19,15 +20,27 @@ public class ResultSetValueProvider implements ValueProvider {
 
     private final ResultSet rs;
     private final int numCols;
+    private final SqliteBlobReader blobReader;
+    private final SqliteColumnType columnType;
 
     public ResultSetValueProvider(ResultSet rs, int numCols) {
+        this(rs, numCols, null);
+    }
+
+    public ResultSetValueProvider(ResultSet rs, int numCols, SqliteBlobReader blobReader) {
         this.rs = rs;
         this.numCols = numCols;
+        this.blobReader = blobReader;
+        this.columnType = SqliteColumnType.of(rs);
     }
 
+    // column_type is zero-based, JDBC one-based
     @Override
     public boolean isNull(int i) {
         try {
+            if (columnType != null) {
+                return columnType.isNull(i);
+            }
             return rs.getObject(i + 1) == null;
         } catch (SQLException e) {
             throw new RuntimeException(e);
@@ -207,6 +220,10 @@ private static int fourDigits(byte[] utf8, int i) {
     @Override
     public HakunaGeometry getHakunaGeometry(int i) {
         try {
+            if (blobReader != null) {
+                ByteBuffer bb = blobReader.read(i);
+                return bb == null ? null : new HakunaGeometryGPKG(bb);
+            }
             byte[] blob = rs.getBytes(i + 1);
             if (blob == null) {
                 return null;

src/hakunapi-source-gpkg/src/main/java/fi/nls/hakunapi/source/gpkg/SqliteBlobReader.java

@@ -0,0 +1,79 @@
+package fi.nls.hakunapi.source.gpkg;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.nio.ByteBuffer;
+import java.sql.PreparedStatement;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.sqlite.core.CoreStatement;
+import org.sqlite.core.DB;
+import org.sqlite.core.SafeStmtPtr;
+
+public final class SqliteBlobReader {
+
+    private static final Logger LOG = LoggerFactory.getLogger(SqliteBlobReader.class);
+
+    private static final Field POINTER_FIELD;
+    private static final Method COLUMN_TEXT_UTF8;
+
+    static {
+        Field pointer = null;
+        Method columnText = null;
+        try {
+            pointer = CoreStatement.class.getDeclaredField("pointer");
+            pointer.setAccessible(true);
+            columnText = Class.forName("org.sqlite.core.NativeDB")
+                    .getDeclaredMethod("column_text_utf8", long.class, int.class);
+            columnText.setAccessible(true);
+        } catch (Exception e) {
+            LOG.info("Zero-copy blob reads unavailable, falling back to ResultSet#getBytes: {}",
+                    e.toString());
+            pointer = null;
+            columnText = null;
+        }
+        POINTER_FIELD = pointer;
+        COLUMN_TEXT_UTF8 = columnText;
+    }
+
+    private final SafeStmtPtr pointer;
+
+    private SqliteBlobReader(SafeStmtPtr pointer) {
+        this.pointer = pointer;
+    }
+
+    public static SqliteBlobReader of(PreparedStatement ps) {
+        if (POINTER_FIELD == null) {
+            return null;
+        }
+        try {
+            PreparedStatement raw = ps.isWrapperFor(PreparedStatement.class)
+                    ? ps.unwrap(PreparedStatement.class) : ps;
+            if (!(raw instanceof CoreStatement)) {
+                return null;
+            }
+            CoreStatement cs = (CoreStatement) raw;
+            SafeStmtPtr pointer = (SafeStmtPtr) POINTER_FIELD.get(cs);
+            DB db = cs.getDatabase();
+            if (pointer == null || !COLUMN_TEXT_UTF8.getDeclaringClass().isInstance(db)) {
+                return null;
+            }
+            return new SqliteBlobReader(pointer);
+        } catch (Exception e) {
+            LOG.info("Zero-copy blob reads unavailable for this statement: {}", e.toString());
+            return null;
+        }
+    }
+
+    /** The buffer is valid only until the cursor advances. */
+    public ByteBuffer read(int i) {
+        try {
+            return pointer.safeRun((d, ptr) -> (ByteBuffer) COLUMN_TEXT_UTF8.invoke(d, ptr, i));
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+}

src/hakunapi-source-gpkg/src/main/java/fi/nls/hakunapi/source/gpkg/SqliteColumnType.java

@@ -0,0 +1,42 @@
+package fi.nls.hakunapi.source.gpkg;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+import org.sqlite.core.CoreStatement;
+import org.sqlite.core.DB;
+
+public final class SqliteColumnType {
+
+    private static final int SQLITE_NULL = 5;
+
+    private final CoreStatement stmt;
+
+    private SqliteColumnType(CoreStatement stmt) {
+        this.stmt = stmt;
+    }
+
+    public static SqliteColumnType of(ResultSet rs) {
+        try {
+            // unwrap because the pool hands out a proxy
+            Statement s = rs.getStatement();
+            if (!s.isWrapperFor(CoreStatement.class)) {
+                return null;
+            }
+            return new SqliteColumnType(s.unwrap(CoreStatement.class));
+        } catch (SQLException e) {
+            return null;
+        }
+    }
+
+    /** @param i zero-based, as SQLite counts columns (JDBC is one-based) */
+    public boolean isNull(int i) throws SQLException {
+        return type(i) == SQLITE_NULL;
+    }
+
+    public int type(int i) throws SQLException {
+        return stmt.pointer.safeRunInt((DB db, long ptr) -> db.column_type(ptr, i));
+    }
+
+}

src/hakunapi-source-gpkg/src/test/java/fi/nls/hakunapi/source/gpkg/SqliteBlobReaderTest.java

@@ -0,0 +1,66 @@
+package fi.nls.hakunapi.source.gpkg;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.nio.ByteBuffer;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Statement;
+
+import org.junit.Test;
+
+public class SqliteBlobReaderTest {
+
+    private static final byte[] BLOB = {0x47, 0x50, 0x00, 0x03, 0x11, 0x22, 0x33, 0x44};
+
+    @Test
+    public void readsTheSameBytesAsGetBytes() throws Exception {
+        try (Connection c = memoryDb()) {
+            try (PreparedStatement ps = c.prepareStatement("SELECT g FROM t ORDER BY id")) {
+                SqliteBlobReader reader = SqliteBlobReader.of(ps);
+                assertNotNull("driver should expose the zero-copy route", reader);
+
+                ResultSet rs = ps.executeQuery();
+
+                assertTrue(rs.next());
+                assertNull("NULL must read as null", reader.read(0));
+                assertNull(rs.getBytes(1));
+
+                assertTrue(rs.next());
+                ByteBuffer empty = reader.read(0);
+                assertNotNull("an empty blob is not NULL", empty);
+                assertEquals(0, empty.remaining());
+                assertEquals(0, rs.getBytes(1).length);
+
+                assertTrue(rs.next());
+                ByteBuffer bb = reader.read(0);
+                byte[] read = new byte[bb.remaining()];
+                bb.get(read);
+                assertArrayEquals(BLOB, read);
+            }
+        }
+    }
+
+    @Test
+    public void unknownStatementYieldsNoReader() throws Exception {
+        assertNull(SqliteBlobReader.of(null));
+    }
+
+    private Connection memoryDb() throws Exception {
+        Connection c = DriverManager.getConnection("jdbc:sqlite::memory:");
+        try (Statement s = c.createStatement()) {
+            s.execute("CREATE TABLE t (id INTEGER, g BLOB)");
+            s.execute("INSERT INTO t VALUES (1, NULL)");
+            s.execute("INSERT INTO t VALUES (2, x'')");
+            s.execute("INSERT INTO t VALUES (3, x'4750000311223344')");
+        }
+        return c;
+    }
+
+}