Making duckdb backend more space efficient

src/acquirium/Storage/duckdb_store.py

@@ -93,15 +93,12 @@ 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 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,
@@ -204,6 +201,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):

tests/test_duckdb_store.py

@@ -39,6 +39,44 @@ def test_protocol_conformance(tmp_path):
     s.close()
 
 
+@pytest.mark.unit
+def test_fresh_store_creates_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 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 ----
 
 @pytest.mark.unit
@@ -130,6 +168,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