Avoid empty results table in main DB for split raw data storage

docs/dataset/dataset_design.rst

@@ -97,9 +97,18 @@ From a design perspective, this feature adds a thin routing layer inside the
   ``__len__``) all go through this single routing point.
 - The per-dataset SQLite file is a lightweight database containing only the
   results table and numpy type adapters -- no QCoDeS metadata schema.
+- When raw data storage is enabled, **no results table is created in the main
+  database** at all -- only the run metadata is stored there. This mirrors how
+  ``DataSetInMem`` records runs (with ``create_run_table=False``). A run is
+  identified as a split-storage dataset by the ``raw_data_db_path`` column in
+  the ``runs`` table (recorded at dataset creation), which is used both to
+  reconnect to the raw data file and to distinguish such runs from
+  ``DataSetInMem`` runs (which also have no results table). This column is an
+  internal storage detail and is kept out of the user-facing metadata.
 - Subscriber triggers (used for real-time data callbacks) are created on the
-  data connection so that they fire regardless of which database holds the
-  results table.
+  data connection. Because the results table only exists once the dataset is
+  started (in the per-dataset file), subscriptions requested before the dataset
+  is started are deferred and materialised at start time.
 
 The implementation is contained in ``qcodes.dataset._raw_data_storage`` (helper
 functions) and a handful of additions to ``qcodes.dataset.data_set`` (routing

docs/dataset/introduction.rst

@@ -92,9 +92,9 @@ This feature is controlled by two configuration options in ``qcodesrc.json``:
 
 When enabled:
 
-- The main database retains the full results table schema (column definitions) but no data rows are written to it, keeping it lightweight.
+- No results table is created in the main database; it stores only the run metadata, keeping it lightweight.
 - All ``INSERT`` and ``SELECT`` operations on results data are transparently routed to the per-dataset file.
-- The path to the per-dataset file is persisted in the run's metadata (``raw_data_db_path``), so ``load_by_id`` and related loading functions automatically reconnect to the correct file.
+- The path to the per-dataset file is recorded in the ``runs`` table, so ``load_by_id`` and related loading functions automatically reconnect to the correct file.
 - All public ``DataSet`` APIs (``get_parameter_data``, ``to_pandas_dataframe``, ``to_xarray_dataset``, ``cache``, ``export``, etc.) work identically whether split storage is enabled or not.
 
 Example runtime configuration::

src/qcodes/dataset/data_set.py

@@ -12,9 +12,9 @@
 from threading import Thread
 from typing import TYPE_CHECKING, Any, Literal
 
 import numpy
 import numpy.typing as npt
 from tqdm.auto import trange
 
 import qcodes
 from qcodes.dataset.data_set_protocol import (
@@ -62,6 +62,7 @@
     get_metadata_from_run_id,
     get_parameter_data,
     get_parent_dataset_links,
+    get_raw_data_db_path_for_run,
     get_run_description,
     get_run_timestamp_from_run_id,
     get_runid_from_guid,
@@ -70,6 +71,7 @@
     mark_run_complete,
     remove_trigger,
     run_exists,
+    set_raw_data_db_path_for_run,
     set_run_timestamp,
     update_parent_datasets,
     update_run_description,
@@ -111,8 +113,8 @@
 if TYPE_CHECKING:
     from collections.abc import Callable, Mapping, Sequence
 
     import pandas as pd
     import xarray as xr
 
     from qcodes.dataset.descriptions.param_spec import ParamSpec
     from qcodes.dataset.descriptions.versioning.rundescribertypes import Shapes
@@ -292,12 +294,18 @@
 
         self._debug = False
         self.subscribers: dict[str, _Subscriber] = {}
+        #: Subscriptions requested before the dataset was started, and thus
+        #: before the results table exists. They are materialised into real
+        #: subscribers (with their SQL triggers) once the dataset is started.
+        self._pending_subscribers: dict[str, dict[str, Any]] = {}
         self._parent_dataset_links: list[Link]
         #: In memory representation of the data in the dataset.
         self._cache: DataSetCacheWithDBBackend = DataSetCacheWithDBBackend(self)
         self._results: list[dict[str, VALUE]] = []
         self._in_memory_cache = in_memory_cache
         self._raw_data_conn: AtomicConnection | None = None
+        #: Path to the per-dataset raw data file when raw data storage is used.
+        self._raw_data_db_path: str | None = None
 
         if run_id is not None:
             if not run_exists(self.conn, run_id):
@@ -317,19 +325,23 @@
                 self.metadata.get("export_info", "")
             )
             # If this dataset was saved with raw data in a separate db,
-            # re-open that connection for reads.
-            raw_db_path = self._metadata.get("raw_data_db_path")
+            # re-open that connection for reads. The path is stored in a
+            # dedicated runs-table column, not in the user-facing metadata.
+            raw_db_path = get_raw_data_db_path_for_run(self.conn, self.run_id)
+            self._raw_data_db_path = raw_db_path
             if raw_db_path is not None:
                 if Path(raw_db_path).is_file():
                     self._raw_data_conn = connect_to_raw_data_db(
                         raw_db_path, read_only=read_only
                     )
-                else:
+                elif self._started:
                     raise FileNotFoundError(
                         f"Raw data file for dataset {self.guid} not found at "
                         f"'{raw_db_path}'. The per-dataset SQLite file may "
                         f"have been moved or deleted."
                     )
+                # else: the dataset was never started, so the raw data file has
+                # not been created yet - there is simply no data to connect to.
         else:
             # Actually perform all the side effects needed for the creation
             # of a new dataset. Note that a dataset is created (in the DB)
@@ -338,6 +350,10 @@
             if exp_id is None:
                 exp_id = get_default_experiment_id(self.conn)
             name = name or "dataset"
+            # When raw data is stored in a separate backend (e.g. a per-dataset
+            # SQLite file), no results table is created in the main database -
+            # only the run metadata is kept there. This mirrors how
+            # ``DataSetInMem`` records runs without a results table.
             _, run_id, __ = create_run(
                 self.conn,
                 exp_id,
@@ -346,6 +362,7 @@
                 parameters=None,
                 values=values,
                 metadata=metadata,
+                create_run_table=not is_raw_data_storage_enabled(),
             )
             # this is really the UUID (an ever increasing count in the db)
             self._run_id = run_id
@@ -364,6 +381,18 @@
             self._metadata = get_metadata_from_run_id(self.conn, self.run_id)
             self._parent_dataset_links = []
             self._export_info = ExportInfo({})
+
+            if is_raw_data_storage_enabled():
+                # Record the raw-data backend location up front. This marks the
+                # run as a split-storage dataset (so it can be told apart from a
+                # ``DataSetInMem`` run, which also has no results table) even
+                # before it is started and before the raw data file is created.
+                # The path is stored in a dedicated column, not in the
+                # user-facing metadata.
+                raw_path_str = str(get_raw_data_db_path(self.guid))
+                self._raw_data_db_path = raw_path_str
+                with atomic(self.conn) as aconn:
+                    set_raw_data_db_path_for_run(aconn, self.run_id, raw_path_str)
         assert self.path_to_db is not None
         if _WRITERS.get(self.path_to_db) is None:
             queue: Queue[Any] = Queue()
@@ -409,6 +438,17 @@
             return self._raw_data_conn
         return self.conn
 
+    @property
+    def _results_table_exists(self) -> bool:
+        """Whether the physical results table exists on the data connection.
+
+        When raw data storage is enabled the results table is created in the
+        per-dataset raw data file only once the dataset has been started, so
+        before that (and in the main database in general) no results table
+        exists. Callers that count rows must handle this case.
+        """
+        return _check_if_table_found(self._data_conn, self.table_name)
+
     @property
     def run_id(self) -> int:
         return self._run_id
@@ -470,6 +510,8 @@
 
     @property
     def number_of_results(self) -> int:
+        if not self._results_table_exists:
+            return 0
         sql = f'SELECT COUNT(*) FROM "{self.table_name}"'
         cursor = atomic_transaction(self._data_conn, sql)
         return one(cursor, "COUNT(*)")
@@ -740,26 +782,30 @@
                 spec,
                 conn=self.conn,
                 run_id=self.run_id,
-                insert_into_results_table=True,
+                # The results table only lives in the main database when raw
+                # data storage is disabled; with it enabled the parameter
+                # columns are created in the per-dataset raw data file below.
+                insert_into_results_table=not raw_data_enabled,
             )
 
         # When raw data split is enabled, create a per-dataset SQLite file
         # for results data with the full results table.
         if raw_data_enabled:
-            raw_db_path = get_raw_data_db_path(self.guid)
+            # The raw-data path was already recorded at dataset creation time;
+            # reuse it so both locations stay in sync.
+            raw_path_str = self._raw_data_db_path or str(
+                get_raw_data_db_path(self.guid)
+            )
+            raw_db_path = Path(raw_path_str)
             self._raw_data_conn = create_raw_data_db(
                 raw_db_path,
                 self.table_name,
                 self._rundescriber.interdeps.paramspecs,
             )
-            # Persist the raw data path in metadata so we can find it when
-            # loading the dataset later.
-            raw_path_str = str(raw_db_path)
-            self._metadata["raw_data_db_path"] = raw_path_str
-            with atomic(self.conn) as aconn:
-                add_data_to_dynamic_columns(
-                    aconn, self.run_id, {"raw_data_db_path": raw_path_str}
-                )
+            if self._raw_data_db_path != raw_path_str:
+                self._raw_data_db_path = raw_path_str
+                with atomic(self.conn) as aconn:
+                    set_raw_data_db_path_for_run(aconn, self.run_id, raw_path_str)
 
         desc_str = serial.to_json_for_storage(self.description)
 
@@ -796,6 +842,11 @@
         writer_status.active_datasets.add(self.run_id)
         self.cache.prepare()
 
+        # Now that the dataset is started (and, for split raw data storage, the
+        # results table exists in the per-dataset file), create any subscribers
+        # that were requested while the dataset was still pristine.
+        self._start_pending_subscribers()
+
     def mark_completed(self) -> None:
         """
         Mark :class:`.DataSet` as complete and thus read-only and notify the subscribers
@@ -1247,13 +1298,44 @@
         callback_kwargs: Mapping[str, Any] | None = None,
     ) -> str:
         subscriber_id = uuid.uuid4().hex
+        if not self._results_table_exists:
+            # The results table does not exist yet - this happens with split
+            # raw data storage, where the table lives in the per-dataset file
+            # created only when the dataset is started. A subscriber has
+            # nothing to observe before the dataset is started, so defer
+            # creating it (and its SQL trigger) until then.
+            self._pending_subscribers[subscriber_id] = {
+                "callback": callback,
+                "min_wait": min_wait,
+                "min_count": min_count,
+                "state": state,
+                "callback_kwargs": callback_kwargs,
+            }
+            return subscriber_id
         subscriber = _Subscriber(
             self, subscriber_id, callback, state, min_wait, min_count, callback_kwargs
         )
         self.subscribers[subscriber_id] = subscriber
         subscriber.start()
         return subscriber_id
 
+    def _start_pending_subscribers(self) -> None:
+        """Materialise subscriptions that were requested before the dataset was
+        started (and thus before the results table existed)."""
+        for subscriber_id, kwargs in self._pending_subscribers.items():
+            subscriber = _Subscriber(
+                self,
+                subscriber_id,
+                kwargs["callback"],
+                kwargs["state"],
+                kwargs["min_wait"],
+                kwargs["min_count"],
+                kwargs["callback_kwargs"],
+            )
+            self.subscribers[subscriber_id] = subscriber
+            subscriber.start()
+        self._pending_subscribers.clear()
+
     def subscribe_from_config(self, name: str) -> str:
         """
         Subscribe a subscriber defined in the `qcodesrc.json` config file to
@@ -1292,6 +1374,10 @@
         """
         Remove subscriber with the provided uuid
         """
+        if uuid in self._pending_subscribers:
+            # Not yet materialised into a real subscriber/trigger.
+            del self._pending_subscribers[uuid]
+            return
         with atomic(self._data_conn) as conn:
             sub = self.subscribers[uuid]
             remove_trigger(conn, sub.trigger_id)
@@ -1322,6 +1408,8 @@
         return get_data_by_tag_and_table_name(self.conn, tag, self.table_name)
 
     def __len__(self) -> int:
+        if not self._results_table_exists:
+            return 0
         return length(self._data_conn, self.table_name)
 
     def __repr__(self) -> str:
@@ -1578,7 +1666,7 @@
 
     def _export_as_netcdf(self, path: Path, file_name: str) -> Path:
         """Export data as netcdf to a given path with file prefix"""
         import xarray as xr
 
         file_path = path / file_name
         if (
@@ -1974,14 +2062,14 @@
     result_table_name = _get_result_table_name_by_guid(conn, guid)
     if _check_if_table_found(conn, result_table_name):
         d = DataSet(conn=conn, run_id=run_id)
+    # The results table is absent from the main DB when raw data is stored
+    # in a separate per-dataset SQLite file. Such runs are marked with a
+    # raw_data_db_path; anything else without a results table is an
+    # in-memory (netcdf-backed) dataset.
+    elif get_raw_data_db_path_for_run(conn, run_id) is not None:
+        d = DataSet(conn=conn, run_id=run_id)
     else:
-        # The results table may be absent from the main DB when raw data
-        # is stored in a separate per-dataset SQLite file.
-        metadata = get_metadata_from_run_id(conn, run_id)
-        if metadata.get("raw_data_db_path") is not None:
-            d = DataSet(conn=conn, run_id=run_id)
-        else:
-            d = DataSetInMem._load_from_db(conn=conn, guid=guid)
+        d = DataSetInMem._load_from_db(conn=conn, guid=guid)
 
     return d
 
@@ -2055,7 +2143,7 @@
     Returns: ASCII art table of information about the supplied guids.
 
     """
     from tabulate import tabulate
 
     headers = (
         "captured_run_id",

src/qcodes/dataset/sqlite/queries.py

@@ -89,6 +89,12 @@
     "captured_counter",
 )
 
+#: Name of the ``runs``-table column used to record the location of a dataset's
+#: raw data when it is stored outside the main database (e.g. in a per-dataset
+#: SQLite file). It is an internal storage detail and is deliberately excluded
+#: from the user-facing dataset metadata.
+RAW_DATA_DB_PATH_COLUMN = "raw_data_db_path"
+
 
 def is_run_id_in_database(conn: AtomicConnection, *run_ids: int) -> dict[int, bool]:
     """
@@ -1838,7 +1844,7 @@ def get_metadata_from_run_id(conn: AtomicConnection, run_id: int) -> dict[str, A
     """
     Get all metadata associated with the specified run
     """
-    non_metadata = RUNS_TABLE_COLUMNS
+    non_metadata = (*RUNS_TABLE_COLUMNS, RAW_DATA_DB_PATH_COLUMN)
 
     metadata = {}
     possible_tags = []
@@ -2309,11 +2315,42 @@ def _get_result_table_name_by_guid(conn: AtomicConnection, guid: str) -> str:
     return formatted_name
 
 
+def get_raw_data_db_path_for_run(conn: AtomicConnection, run_id: int) -> str | None:
+    """Return the stored raw-data file path for a run.
+
+    The path is read directly from the ``raw_data_db_path`` column of the
+    ``runs`` table (which is kept out of the user-facing metadata). Returns
+    ``None`` if the column does not exist or is not set for the run.
+    """
+    if not is_column_in_table(conn, "runs", RAW_DATA_DB_PATH_COLUMN):
+        return None
+    cursor = conn.execute(
+        f'SELECT "{RAW_DATA_DB_PATH_COLUMN}" FROM runs WHERE run_id = ?', (run_id,)
+    )
+    row = cursor.fetchone()
+    if row is None:
+        return None
+    return row[0]
+
+
+def set_raw_data_db_path_for_run(
+    conn: AtomicConnection, run_id: int, raw_data_db_path: str
+) -> None:
+    """Record the raw-data file path for a run in the ``runs`` table."""
+    add_data_to_dynamic_columns(
+        conn, run_id, {RAW_DATA_DB_PATH_COLUMN: raw_data_db_path}
+    )
+
+
 def get_datasets_with_raw_data_path(
     conn: AtomicConnection,
 ) -> list[tuple[int, str, str, str, float | None, float | None, str, str]]:
     """Get all datasets that have a raw_data_db_path metadata column set.
 
+    Only datasets that have been started (``run_timestamp`` is set) are
+    returned, since an unstarted run records the intended raw-data path but
+    never creates the corresponding file.
+
     Returns:
         A list of tuples:
         ``(run_id, guid, experiment_name, sample_name, run_timestamp,
@@ -2331,6 +2368,7 @@ def get_datasets_with_raw_data_path(
     FROM runs r
     JOIN experiments e ON r.exp_id = e.exp_id
     WHERE r.raw_data_db_path IS NOT NULL
+      AND r.run_timestamp IS NOT NULL
     """
     cursor = atomic_transaction(conn, sql)
     return cursor.fetchall()