diff --git a/CHANGELOG.md b/CHANGELOG.md index f8177def..0bad837c 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/hakunapi-source-gpkg/src/main/java/fi/nls/hakunapi/source/gpkg/GpkgSimpleFeatureProducer.java b/src/hakunapi-source-gpkg/src/main/java/fi/nls/hakunapi/source/gpkg/GpkgSimpleFeatureProducer.java index d2610112..326b4d13 100644 --- a/src/hakunapi-source-gpkg/src/main/java/fi/nls/hakunapi/source/gpkg/GpkgSimpleFeatureProducer.java +++ b/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; diff --git a/src/hakunapi-source-gpkg/src/main/java/fi/nls/hakunapi/source/gpkg/BufferedResultSet.java b/src/hakunapi-source-gpkg/src/main/java/fi/nls/hakunapi/source/gpkg/ResultSetFeatureStream.java similarity index 58% rename from src/hakunapi-source-gpkg/src/main/java/fi/nls/hakunapi/source/gpkg/BufferedResultSet.java rename to src/hakunapi-source-gpkg/src/main/java/fi/nls/hakunapi/source/gpkg/ResultSetFeatureStream.java index 988568c0..5c14d120 100644 --- a/src/hakunapi-source-gpkg/src/main/java/fi/nls/hakunapi/source/gpkg/BufferedResultSet.java +++ b/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 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 mappers, int bufSize) { + public ResultSetFeatureStream(Connection c, PreparedStatement ps, ResultSet rs, int numColsRs, + List 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 diff --git a/src/hakunapi-source-gpkg/src/main/java/fi/nls/hakunapi/source/gpkg/ResultSetValueProvider.java b/src/hakunapi-source-gpkg/src/main/java/fi/nls/hakunapi/source/gpkg/ResultSetValueProvider.java index dd4fbf31..49b3e294 100644 --- a/src/hakunapi-source-gpkg/src/main/java/fi/nls/hakunapi/source/gpkg/ResultSetValueProvider.java +++ b/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; diff --git a/src/hakunapi-source-gpkg/src/main/java/fi/nls/hakunapi/source/gpkg/SqliteBlobReader.java b/src/hakunapi-source-gpkg/src/main/java/fi/nls/hakunapi/source/gpkg/SqliteBlobReader.java new file mode 100644 index 00000000..b40b56a1 --- /dev/null +++ b/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); + } + } + +} diff --git a/src/hakunapi-source-gpkg/src/main/java/fi/nls/hakunapi/source/gpkg/SqliteColumnType.java b/src/hakunapi-source-gpkg/src/main/java/fi/nls/hakunapi/source/gpkg/SqliteColumnType.java new file mode 100644 index 00000000..c66b872b --- /dev/null +++ b/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)); + } + +} diff --git a/src/hakunapi-source-gpkg/src/test/java/fi/nls/hakunapi/source/gpkg/SqliteBlobReaderTest.java b/src/hakunapi-source-gpkg/src/test/java/fi/nls/hakunapi/source/gpkg/SqliteBlobReaderTest.java new file mode 100644 index 00000000..7d141110 --- /dev/null +++ b/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; + } + +} diff --git a/src/hakunapi-source-gpkg/src/test/java/fi/nls/hakunapi/source/gpkg/SqliteColumnTypeTest.java b/src/hakunapi-source-gpkg/src/test/java/fi/nls/hakunapi/source/gpkg/SqliteColumnTypeTest.java new file mode 100644 index 00000000..4742724d --- /dev/null +++ b/src/hakunapi-source-gpkg/src/test/java/fi/nls/hakunapi/source/gpkg/SqliteColumnTypeTest.java @@ -0,0 +1,70 @@ +package fi.nls.hakunapi.source.gpkg; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.Statement; + +import org.junit.Test; + +import com.zaxxer.hikari.HikariConfig; +import com.zaxxer.hikari.HikariDataSource; + +public class SqliteColumnTypeTest { + + private static final String QUERY = "select NULL, 1, 'x', NULL"; + + @Test + public void readsColumnTypeOnPlainAndPooledConnections() throws Exception { + try (Connection c = DriverManager.getConnection("jdbc:sqlite::memory:"); + Statement s = c.createStatement(); + ResultSet rs = s.executeQuery(QUERY)) { + assertTrue(rs.next()); + SqliteColumnType type = SqliteColumnType.of(rs); + assertNotNull(type); + assertNullFlags(type); + } + + HikariConfig config = new HikariConfig(); + config.setJdbcUrl("jdbc:sqlite::memory:"); + config.setMaximumPoolSize(1); + HikariDataSource ds = new HikariDataSource(config); + try { + try (Connection c = ds.getConnection(); + Statement s = c.createStatement(); + ResultSet rs = s.executeQuery(QUERY)) { + assertTrue(rs.next()); + SqliteColumnType type = SqliteColumnType.of(rs); + assertNotNull(type); + assertNullFlags(type); + } + + try (Connection c = ds.getConnection(); + Statement s = c.createStatement(); + ResultSet rs = s.executeQuery(QUERY)) { + assertTrue(rs.next()); + assertNullFlags(new ResultSetValueProvider(rs, 4)); + } + } finally { + ds.close(); + } + } + + private static void assertNullFlags(SqliteColumnType type) throws Exception { + assertTrue(type.isNull(0)); + assertFalse(type.isNull(1)); + assertFalse(type.isNull(2)); + assertTrue(type.isNull(3)); + } + + private static void assertNullFlags(ResultSetValueProvider values) { + assertTrue(values.isNull(0)); + assertFalse(values.isNull(1)); + assertFalse(values.isNull(2)); + assertTrue(values.isNull(3)); + } +}