Skip to content

perf(gpkg): connection reuse + prepared bulk-insert + keyset paging (MAPCO-11320) #255

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
shimoncohen wants to merge 3 commits into
base: master
Choose a base branch
from
Open

perf(gpkg): connection reuse + prepared bulk-insert + keyset paging (MAPCO-11320) #255

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
781e604
perf(gpkg): reuse connection, prepared bulk-insert, keyset paging (MA…
shimoncohen Aug 4, 2026
79dc642
docs: tighten GetBatch keyset-pagination comment
shimoncohen Aug 5, 2026
7551a1b
fix(gpkg): dispose partially-built sources; rename GetConnection
shimoncohen Aug 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions MergerCli/Program.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,10 @@ private static void Main(string[] args)
{
_logger.LogError("minimum of 2 sources is required");
PrintHelp(programName);
foreach (IData source in sources)
{
source.Dispose();
}
return;
}

Expand Down Expand Up @@ -136,6 +140,13 @@ private static void Main(string[] args)
_logger.LogError(ex, ex.Message);
return;
}
finally
{
foreach (IData source in sources)
Comment thread
asafmas-rnd marked this conversation as resolved.
{
source.Dispose();
}
}

totalTimeStopwatch.Stop();
// Get the elapsed time as a TimeSpan value.
Expand Down
131 changes: 83 additions & 48 deletions MergerLogic/Clients/GpkgClient.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,13 @@ public class GpkgClient : DataUtils, IGpkgClient
private readonly ILogger _logger;
private readonly IFileSystem _fileSystem;

// Single reusable connection for the hot path (GetTile/TileExists/GetLastTile/InsertTiles/GetBatch).
// Opened lazily and guarded by _connectionLock so the client is safe under the concurrent access
// that Data.GetLastExistingTile (Parallel.ForEachAsync) and the service-side parallel merge perform.
private readonly object _connectionLock = new object();
private SQLiteConnection? _connection;
private bool _disposed;

public GpkgClient(string path, ITimeUtils timeUtils, ILogger<GpkgClient> logger, IFileSystem fileSystem,
IGeoUtils geoUtils) : base(path, geoUtils)
{
Expand All @@ -27,6 +34,41 @@ public GpkgClient(string path, ITimeUtils timeUtils, ILogger<GpkgClient> logger,
this._tileCache = this.InternalGetTileCache();
}

// Returns the shared connection, opening it (with WAL) on first use. Caller must hold _connectionLock.
private SQLiteConnection GetOrCreateConnection()
{
if (this._connection == null)
Comment thread
asafmas-rnd marked this conversation as resolved.
{
var connection = new SQLiteConnection($"Data Source={this.path}");
connection.Open();
using (var pragma = connection.CreateCommand())
{
// WAL + NORMAL sync: concurrent readers alongside a writer and far fewer fsyncs on the hot path.
pragma.CommandText = "PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL;";
pragma.ExecuteNonQuery();
}

this._connection = connection;
}

return this._connection;
}

public void Dispose()
{
if (this._disposed)
{
return;
}

lock (this._connectionLock)
{
this._connection?.Dispose();
this._connection = null;
this._disposed = true;
}
}

private string InternalGetTileCache()
{
if (!this.Exist())
Expand Down Expand Up @@ -126,20 +168,19 @@ public void UpdateExtent(Extent extent)

public override Tile? GetTile(int z, int x, int y)
{
byte[]? blob = null;
byte[]? blob;

using (var connection = new SQLiteConnection($"Data Source={this.path}"))
lock (this._connectionLock)
{
connection.Open();

var connection = this.GetOrCreateConnection();
using (var command = connection.CreateCommand())
{
command.CommandText =
$"SELECT tile_data FROM \"{this._tileCache}\" WHERE zoom_level=$z AND tile_column=$x AND tile_row=$y LIMIT 1";
command.Parameters.AddWithValue("$z", z);
command.Parameters.AddWithValue("$x", x);
command.Parameters.AddWithValue("$y", y);
blob = (byte[])command.ExecuteScalar();
blob = command.ExecuteScalar() as byte[];
}
}

Expand All @@ -148,10 +189,9 @@ public void UpdateExtent(Extent extent)

public override bool TileExists(int z, int x, int y)
{
using (var connection = new SQLiteConnection($"Data Source={this.path}"))
lock (this._connectionLock)
{
connection.Open();

var connection = this.GetOrCreateConnection();
using (var command = connection.CreateCommand())
{
command.CommandText =
Expand All @@ -162,73 +202,69 @@ public override bool TileExists(int z, int x, int y)

using (var reader = command.ExecuteReader(System.Data.CommandBehavior.SingleRow))
{
// Check if a row was returned
if (reader.HasRows)
{
return true;
}
return reader.HasRows;
}
}
}

return false;
}

public void InsertTiles(IEnumerable<Tile> tiles)
{
using (var connection = new SQLiteConnection($"Data Source={this.path}"))
lock (this._connectionLock)
{
connection.Open();

var connection = this.GetOrCreateConnection();
using (var transaction = connection.BeginTransaction())
using (var command = connection.CreateCommand())
{
command.CommandText =
$"REPLACE INTO \"{this._tileCache}\" (zoom_level, tile_column, tile_row, tile_data) VALUES ($z, $x, $y, $blob)";

using (var transaction = connection.BeginTransaction())
{
foreach (Tile tile in tiles)
{
byte[] tileBytes = tile.GetImageBytes();
SQLiteParameter blobParameter =
new SQLiteParameter("$blob", System.Data.DbType.Binary, tileBytes.Length);
blobParameter.Value = tileBytes;

command.Parameters.AddWithValue("$z", tile.Z);
command.Parameters.AddWithValue("$x", tile.X);
command.Parameters.AddWithValue("$y", tile.Y);
command.Parameters.Add(blobParameter);
command.ExecuteNonQuery();
}
// Bind parameters once and reuse the prepared statement for every tile in the batch.
var zParameter = command.Parameters.Add("$z", System.Data.DbType.Int32);
var xParameter = command.Parameters.Add("$x", System.Data.DbType.Int32);
var yParameter = command.Parameters.Add("$y", System.Data.DbType.Int32);
var blobParameter = command.Parameters.Add("$blob", System.Data.DbType.Binary);
command.Prepare();

transaction.Commit();
foreach (Tile tile in tiles)
{
zParameter.Value = tile.Z;
xParameter.Value = tile.X;
yParameter.Value = tile.Y;
blobParameter.Value = tile.GetImageBytes();
command.ExecuteNonQuery();
}

transaction.Commit();
}
}
}

public List<Tile> GetBatch(int batchSize, long offset)
// Keyset pagination over rowid: seeks past lastId instead of OFFSET-scanning, so page cost stays
// constant regardless of depth. lastId is the cursor (0 to start); returns the greatest rowid read
// (or lastId unchanged when the page is empty).
public (List<Tile> Tiles, long LastId) GetBatch(int batchSize, long lastId)
{
List<Tile> tiles = new List<Tile>();
List<Tile> tiles = new List<Tile>(batchSize);

using (var connection = new SQLiteConnection($"Data Source={this.path}"))
lock (this._connectionLock)
{
connection.Open();

var connection = this.GetOrCreateConnection();
using (var command = connection.CreateCommand())
{
command.CommandText =
$"SELECT zoom_level, tile_column, tile_row, tile_data FROM \"{this._tileCache}\" ORDER BY zoom_level ASC limit $limit offset $offset";
$"SELECT rowid, zoom_level, tile_column, tile_row, tile_data FROM \"{this._tileCache}\" WHERE rowid > $lastId ORDER BY rowid ASC LIMIT $limit";
command.Parameters.AddWithValue("$lastId", lastId);
command.Parameters.AddWithValue("$limit", batchSize);
command.Parameters.AddWithValue("$offset", offset);

using (var reader = command.ExecuteReader())
{
while (reader.Read())
{
var z = reader.GetInt32(0);
var x = reader.GetInt32(1);
var y = reader.GetInt32(2);
lastId = reader.GetInt64(0);
var z = reader.GetInt32(1);
var x = reader.GetInt32(2);
var y = reader.GetInt32(3);
var blob = (byte[])reader["tile_data"];

Tile tile = this.CreateTile(z, x, y, blob)!;
Expand All @@ -238,7 +274,7 @@ public List<Tile> GetBatch(int batchSize, long offset)
}
}

return tiles;
return (tiles, lastId);
}

public Tile? GetLastTile(int[] coords, int currentTileZoom)
Expand All @@ -249,10 +285,9 @@ public List<Tile> GetBatch(int batchSize, long offset)
}

Tile? lastTile = null;
using (var connection = new SQLiteConnection($"Data Source={this.path}"))
lock (this._connectionLock)
{
connection.Open();

var connection = this.GetOrCreateConnection();
using (var command = connection.CreateCommand())
{
// Build command
Expand Down
4 changes: 2 additions & 2 deletions MergerLogic/Clients/IGpkgClient.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,9 +4,9 @@

namespace MergerLogic.Clients
{
public interface IGpkgClient : IDataUtils
public interface IGpkgClient : IDataUtils, IDisposable
{
List<Tile> GetBatch(int batchSize, long offset);
(List<Tile> Tiles, long LastId) GetBatch(int batchSize, long lastId);
Comment thread
asafmas-rnd marked this conversation as resolved.
Extent GetExtent();
Tile? GetLastTile(int[] coords, int currentTileZoom);
long GetTileCount();
Expand Down
6 changes: 6 additions & 0 deletions MergerLogic/DataTypes/Data.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -336,5 +336,11 @@ public virtual void Wrapup()
public abstract long TileCount();

public abstract void setBatchIdentifier(string batchIdentifier);

public virtual void Dispose()
{
(this.Utils as IDisposable)?.Dispose();
GC.SuppressFinalize(this);
}
}
}
44 changes: 18 additions & 26 deletions MergerLogic/DataTypes/Gpkg.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,8 @@ namespace MergerLogic.DataTypes
{
public class Gpkg : Data<IGpkgClient>
{
private long _offset;
// Keyset cursor: the greatest tile id handed out so far. Persisted as the batch identifier for resume.
private long _batchCursorId;
private Extent _extent;
private readonly IConfigurationManager _configManager;
static readonly object _locker = new object();
Expand All @@ -18,7 +19,7 @@ public Gpkg(IConfigurationManager configuration, IServiceProvider container,
: base(container, DataType.GPKG, path, batchSize, grid, origin, isBase, extent)
{
this._logger.LogDebug($"[{MethodBase.GetCurrentMethod()?.Name}] Ctor started");
this._offset = 0;
this._batchCursorId = 0;
this._configManager = configuration;

if (isBase)
Expand Down Expand Up @@ -82,7 +83,7 @@ public override void Reset()
{
lock (_locker)
{
this._offset = 0;
this._batchCursorId = 0;
}
}

Expand All @@ -91,28 +92,19 @@ public override List<Tile> GetNextBatch(out string currentBatchIdentifier, out s
lock (_locker)
{
this._logger.LogDebug($"[{MethodBase.GetCurrentMethod()?.Name}] started");
currentBatchIdentifier = this._offset.ToString();
List<Tile> tiles = new List<Tile>();
if (this._offset != totalTilesCount)
{
//TODO: optimize after IOC refactoring
int counter = 0;

tiles = this.Utils.GetBatch(this.BatchSize, this._offset)
.Select(t =>
{
Tile tile = this.ConvertOriginTile(t);
tile = this.ToCurrentGrid(tile);
counter++;
return tile;
}).Where(t => t != null).ToList();

Interlocked.Add(ref this._offset, counter);
nextBatchIdentifier = this._offset.ToString();

return tiles;
}
nextBatchIdentifier = this._offset.ToString();
currentBatchIdentifier = this._batchCursorId.ToString();

var (rawTiles, cursor) = this.Utils.GetBatch(this.BatchSize, this._batchCursorId);
List<Tile> tiles = rawTiles
.Select(t =>
{
Tile tile = this.ConvertOriginTile(t);
tile = this.ToCurrentGrid(tile);
return tile;
}).Where(t => t != null).ToList();

this._batchCursorId = cursor;
nextBatchIdentifier = this._batchCursorId.ToString();
this._logger.LogDebug($"[{MethodBase.GetCurrentMethod()?.Name}] ended");
return tiles;
}
Expand All @@ -122,7 +114,7 @@ public override void setBatchIdentifier(string batchIdentifier)
{
lock (_locker)
{
this._offset = long.Parse(batchIdentifier);
this._batchCursorId = long.Parse(batchIdentifier);
}
}

Expand Down
2 changes: 1 addition & 1 deletion MergerLogic/DataTypes/IData.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

namespace MergerLogic.DataTypes
{
public interface IData
public interface IData : IDisposable
{
public DataType Type { get; }
public string Path { get; }
Expand Down
7 changes: 4 additions & 3 deletions MergerLogicUnitTests/DataTypes/GpkgTest.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -671,7 +671,7 @@ public void SetBatchIdentifier(bool isOneXOne, bool isBase, GridOrigin origin, i

this.SetupRequiredBaseMocks(isBase, isOneXOne, extent);
this._gpkgUtilsMock.Setup(utils => utils.GetBatch(10, It.IsAny<long>()))
.Returns(new List<Tile>());
.Returns<int, long>((_, lastId) => (new List<Tile>(), lastId));

var gpkg = new Gpkg(this._configurationManagerMock.Object,
this._serviceProviderMock.Object, "test.gpkg", 10, grid, origin,
Expand Down Expand Up @@ -709,7 +709,7 @@ public void Reset(bool isOneXOne, bool isBase, GridOrigin origin, int batchSize)

this.SetupRequiredBaseMocks(isBase, isOneXOne, extent);
this._gpkgUtilsMock.Setup(utils => utils.GetBatch(batchSize, It.IsAny<long>()))
.Returns(new List<Tile> { new Tile(0, 0, 0, this._jpegImageData) });
.Returns<int, long>((_, lastId) => (new List<Tile> { new Tile(0, 0, 0, this._jpegImageData) }, lastId + 1));
if (origin == GridOrigin.UPPER_LEFT)
{
this._geoUtilsMock.Setup(converter => converter.FlipY(It.IsAny<Tile>()))
Expand Down Expand Up @@ -769,10 +769,11 @@ public void GetNextBatch(bool isOneXOne, bool isBase, GridOrigin origin, int bat
var seq = new MockSequence();
for (var i = 0; i < tileBatches.Count; i++)
{
var batch = tileBatches[i].ToList();
this._gpkgUtilsMock
.InSequence(seq)
.Setup(utils => utils.GetBatch(batchSize, It.IsAny<long>()))
.Returns(tileBatches[i].ToList());
.Returns<int, long>((_, lastId) => (batch, lastId + batch.Count));
for (var j = 0; j < tileBatches[i].Length; j++)
{
if (origin == GridOrigin.UPPER_LEFT)
Expand Down
Loading
Loading