From 7ed46303d1db29ec9a5d1cd1a8f5d4a5336fbaca Mon Sep 17 00:00:00 2001 From: Gabe Fierro Date: Fri, 29 May 2026 12:37:24 -0600 Subject: [PATCH 1/2] Optimize DuckDB timeseries storage --- src/acquirium/Storage/duckdb_store.py | 9 ++---- tests/test_duckdb_store.py | 46 +++++++++++++++++++++++++++ 2 files changed, 48 insertions(+), 7 deletions(-) diff --git a/src/acquirium/Storage/duckdb_store.py b/src/acquirium/Storage/duckdb_store.py index fb54d17..07fc470 100644 --- a/src/acquirium/Storage/duckdb_store.py +++ b/src/acquirium/Storage/duckdb_store.py @@ -93,15 +93,9 @@ def ensure_table(self) -> str: ts TIMESTAMP NOT NULL, numeric_value DOUBLE, text_value VARCHAR, - CHECK (numeric_value IS NULL OR text_value IS NULL), - UNIQUE (ref_uri, ts) + CHECK (numeric_value IS NULL OR text_value IS NULL) ) """, - f"CREATE INDEX IF NOT EXISTS idx_ts_ref ON {TIMESERIES_TABLE} (ref_uri, ts)", - f"CREATE INDEX IF NOT EXISTS idx_ts_numeric_ref ON {TIMESERIES_TABLE} (ref_uri, ts, numeric_value)", - f"CREATE INDEX IF NOT EXISTS idx_ts_text_ref ON {TIMESERIES_TABLE} (ref_uri, ts, text_value)", - f"CREATE INDEX IF NOT EXISTS idx_ts_numeric_value ON {TIMESERIES_TABLE} (ref_uri, numeric_value)", - f"CREATE INDEX IF NOT EXISTS idx_ts_text_value ON {TIMESERIES_TABLE} (ref_uri, text_value)", f""" CREATE TABLE IF NOT EXISTS {STREAMS_TABLE} ( ref_uri VARCHAR PRIMARY KEY, @@ -204,6 +198,7 @@ def bulk_insert_polars(self, df: pl.DataFrame) -> int: pl.col("ts").dt.convert_time_zone("UTC").dt.replace_time_zone(None) ) df = df.unique(subset=["ref_uri", "ts"], keep="last", maintain_order=True) + df = df.sort(["ref_uri", "ts"]) deduped = len(df) logger.debug("bulk_insert_polars: %d rows after dedupe (was %d)", deduped, in_rows) with self._lock, timed_debug(logger, "bulk_insert_polars DELETE+INSERT rows=%d", deduped): diff --git a/tests/test_duckdb_store.py b/tests/test_duckdb_store.py index 402b6fd..850e48b 100644 --- a/tests/test_duckdb_store.py +++ b/tests/test_duckdb_store.py @@ -39,6 +39,19 @@ def test_protocol_conformance(tmp_path): s.close() +@pytest.mark.unit +def test_fresh_store_does_not_create_timeseries_indexes(tmp_path): + s = DuckDBStore(db_path=tmp_path / "schema.duckdb", recreate=True) + try: + result = s.sql_query( + "SELECT index_name FROM duckdb_indexes() WHERE table_name = 'timeseries'" + ) + finally: + s.close() + + assert result["rows"] == [] + + # ---- upsert_rows ---- @pytest.mark.unit @@ -130,6 +143,39 @@ def test_bulk_insert_polars_splits_numeric_and_text_values(store): ] +@pytest.mark.unit +def test_bulk_insert_polars_appends_rows_in_stream_time_order(tmp_path): + s = DuckDBStore(db_path=tmp_path / "ordered_insert.duckdb", recreate=True) + try: + df = pl.DataFrame( + { + "ref_uri": ["urn:test:duck:z", "urn:test:duck:a", "urn:test:duck:z", "urn:test:duck:a"], + "ts": [_utc(2024, 2, 2), _utc(2024, 2, 2), _utc(2024, 2, 1), _utc(2024, 2, 1)], + "value": [4.0, 2.0, 3.0, 1.0], + "value_kind": ["numeric", "numeric", "numeric", "numeric"], + } + ) + + assert s.bulk_insert_polars(df) == 4 + + stored = s.sql_query( + """ + SELECT ref_uri, ts, numeric_value + FROM timeseries + ORDER BY rowid + """ + )["rows"] + finally: + s.close() + + assert stored == [ + ["urn:test:duck:a", _utc(2024, 2, 1).replace(tzinfo=None), 1.0], + ["urn:test:duck:a", _utc(2024, 2, 2).replace(tzinfo=None), 2.0], + ["urn:test:duck:z", _utc(2024, 2, 1).replace(tzinfo=None), 3.0], + ["urn:test:duck:z", _utc(2024, 2, 2).replace(tzinfo=None), 4.0], + ] + + # ---- timeseries query ---- @pytest.mark.unit From b506fb04cb73170aa8a72976704506b517767381 Mon Sep 17 00:00:00 2001 From: Gabe Fierro Date: Fri, 29 May 2026 15:44:14 -0600 Subject: [PATCH 2/2] Add DuckDB timeseries uniqueness indexes --- src/acquirium/Storage/duckdb_store.py | 3 +++ tests/test_duckdb_store.py | 29 +++++++++++++++++++++++++-- 2 files changed, 30 insertions(+), 2 deletions(-) diff --git a/src/acquirium/Storage/duckdb_store.py b/src/acquirium/Storage/duckdb_store.py index 07fc470..3dd46fc 100644 --- a/src/acquirium/Storage/duckdb_store.py +++ b/src/acquirium/Storage/duckdb_store.py @@ -96,6 +96,9 @@ def ensure_table(self) -> str: CHECK (numeric_value IS NULL OR text_value IS NULL) ) """, + f"CREATE UNIQUE INDEX IF NOT EXISTS idx_timeseries_ref_ts_unique ON {TIMESERIES_TABLE} (ref_uri, ts)", + f"CREATE INDEX IF NOT EXISTS idx_timeseries_numeric_ref_ts ON {TIMESERIES_TABLE} (ref_uri, numeric_value, ts)", + f"CREATE INDEX IF NOT EXISTS idx_timeseries_text_ref_ts ON {TIMESERIES_TABLE} (ref_uri, text_value, ts)", f""" CREATE TABLE IF NOT EXISTS {STREAMS_TABLE} ( ref_uri VARCHAR PRIMARY KEY, diff --git a/tests/test_duckdb_store.py b/tests/test_duckdb_store.py index 850e48b..d19af14 100644 --- a/tests/test_duckdb_store.py +++ b/tests/test_duckdb_store.py @@ -40,7 +40,7 @@ def test_protocol_conformance(tmp_path): @pytest.mark.unit -def test_fresh_store_does_not_create_timeseries_indexes(tmp_path): +def test_fresh_store_creates_timeseries_indexes(tmp_path): s = DuckDBStore(db_path=tmp_path / "schema.duckdb", recreate=True) try: result = s.sql_query( @@ -49,7 +49,32 @@ def test_fresh_store_does_not_create_timeseries_indexes(tmp_path): finally: s.close() - assert result["rows"] == [] + assert sorted(row[0] for row in result["rows"]) == [ + "idx_timeseries_numeric_ref_ts", + "idx_timeseries_ref_ts_unique", + "idx_timeseries_text_ref_ts", + ] + + +@pytest.mark.unit +def test_fresh_store_enforces_ref_uri_timestamp_uniqueness(tmp_path): + s = DuckDBStore(db_path=tmp_path / "unique.duckdb", recreate=True) + try: + s.sql_query( + """ + INSERT INTO timeseries (ref_uri, ts, numeric_value, text_value) + VALUES ('urn:test:duck:unique', TIMESTAMP '2024-01-01 00:00:00', 1.0, NULL) + """ + ) + with pytest.raises(Exception, match="Duplicate key"): + s.sql_query( + """ + INSERT INTO timeseries (ref_uri, ts, numeric_value, text_value) + VALUES ('urn:test:duck:unique', TIMESTAMP '2024-01-01 00:00:00', 2.0, NULL) + """ + ) + finally: + s.close() # ---- upsert_rows ----