diff --git a/docs/dataset/dataset_design.rst b/docs/dataset/dataset_design.rst index b17727eed96..e78ee1b42bd 100644 --- a/docs/dataset/dataset_design.rst +++ b/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 diff --git a/docs/dataset/introduction.rst b/docs/dataset/introduction.rst index aeaa7cb1f70..2e28ee4311b 100644 --- a/docs/dataset/introduction.rst +++ b/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:: diff --git a/src/qcodes/dataset/data_set.py b/src/qcodes/dataset/data_set.py index ef9d475111b..49a3a2e0a76 100644 --- a/src/qcodes/dataset/data_set.py +++ b/src/qcodes/dataset/data_set.py @@ -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, @@ -292,12 +294,18 @@ def __init__( 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 @@ def __init__( 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 @@ def __init__( 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 @@ def __init__( 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 @@ def __init__( 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 @@ def _data_conn(self) -> AtomicConnection: 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 @@ def snapshot_raw(self) -> str | None: @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 @@ def _perform_start_actions(self, start_bg_writer: bool) -> None: 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 @@ def _perform_start_actions(self, start_bg_writer: bool) -> None: 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,6 +1298,20 @@ def subscribe( 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 ) @@ -1254,6 +1319,23 @@ def subscribe( 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 @@ def unsubscribe(self, uuid: str) -> None: """ 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 @@ def get_metadata(self, tag: str) -> VALUE | None: 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: @@ -1974,14 +2062,14 @@ def _get_datasetprotocol_from_guid( 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 diff --git a/src/qcodes/dataset/sqlite/queries.py b/src/qcodes/dataset/sqlite/queries.py index 8a9f247c2d6..dc547b75aa9 100644 --- a/src/qcodes/dataset/sqlite/queries.py +++ b/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() diff --git a/tests/dataset/test_raw_data_storage.py b/tests/dataset/test_raw_data_storage.py index 010b6de66cd..f7d4c339658 100644 --- a/tests/dataset/test_raw_data_storage.py +++ b/tests/dataset/test_raw_data_storage.py @@ -37,12 +37,21 @@ ) from qcodes.dataset.descriptions.dependencies import InterDependencies_ from qcodes.dataset.sqlite.database import connect, initialise_database +from qcodes.dataset.sqlite.queries import get_raw_data_db_path_for_run +from qcodes.dataset.sqlite.query_helpers import select_one_where from qcodes.parameters import ParamSpecBase if TYPE_CHECKING: from collections.abc import Generator +def _raw_file(ds: DataSet) -> Path: + """Return the per-dataset raw data file path, asserting it is set.""" + raw_path = ds._raw_data_db_path + assert raw_path is not None + return Path(raw_path) + + # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @@ -191,27 +200,37 @@ def test_raw_data_file_created(self, tmp_path: Path) -> None: assert raw_file.is_file() self._close_ds(ds) - def test_raw_data_db_path_in_metadata(self) -> None: - """The path to the raw data file should be stored in metadata.""" + def test_raw_data_db_path_recorded(self) -> None: + """The path to the raw data file should be recorded in the runs table + (in a dedicated column) but kept out of the user-facing metadata.""" ds, _ = self._make_dataset_with_data() - assert "raw_data_db_path" in ds.metadata - assert Path(ds.metadata["raw_data_db_path"]).is_file() + # Not exposed as user metadata ... + assert "raw_data_db_path" not in ds.metadata + # ... but recorded internally and pointing at the real file. + assert ds._raw_data_db_path is not None + assert _raw_file(ds).is_file() + assert ds._raw_data_db_path == get_raw_data_db_path_for_run(ds.conn, ds.run_id) self._close_ds(ds) def test_data_is_in_raw_db_not_main(self) -> None: - """Data should be in the raw data file, not the main DB.""" + """Data should be in the raw data file, and the main DB should not + contain a results table at all.""" ds, _ = self._make_dataset_with_data(n_rows=5) table_name = ds.table_name main_conn = ds.conn - # Main DB should have the results table (schema) but no data rows + # Main DB should NOT contain the results table - only metadata is kept + # there when raw data storage is enabled. cursor = main_conn.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table_name,), ) - assert cursor.fetchone() is not None - cursor = main_conn.execute(f'SELECT COUNT(*) FROM "{table_name}"') - assert cursor.fetchone()[0] == 0 + assert cursor.fetchone() is None + + # The result_table_name identifier is still recorded in the runs table. + assert table_name == select_one_where( + main_conn, "runs", "result_table_name", "run_id", ds.run_id + ) # Raw data DB should have the actual data raw_conn = ds._raw_data_conn @@ -227,7 +246,58 @@ def test_number_of_results(self) -> None: assert ds.number_of_results == 7 self._close_ds(ds) - def test_get_parameter_data(self) -> None: + def test_no_results_table_in_main_db_before_started(self) -> None: + """A pristine split dataset has no results table anywhere, and + counting results returns 0 without raising.""" + ds = new_data_set("test-split") + x = ParamSpecBase("x", "numeric") + y = ParamSpecBase("y", "numeric") + idps = InterDependencies_(dependencies={y: (x,)}) + ds.set_interdependencies(idps) + + # Not started yet: no raw data file, and no table in the main DB. + assert ds._raw_data_conn is None + cursor = ds.conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name=?", + (ds.table_name,), + ) + assert cursor.fetchone() is None + + # Counting must not raise even though the table does not exist. + assert ds.number_of_results == 0 + assert len(ds) == 0 + self._close_ds(ds) + + def test_subscribe_before_start_defers_and_fires(self) -> None: + """Subscribing before the dataset is started (before the raw data + table exists) should be deferred and then fire once data is added.""" + ds = new_data_set("test-split") + x = ParamSpecBase("x", "numeric") + y = ParamSpecBase("y", "numeric") + idps = InterDependencies_(dependencies={y: (x,)}) + ds.set_interdependencies(idps) + + received: list[int] = [] + + def callback(results, length, state): + received.append(length) + + # Subscribe while still pristine - no results table exists yet. + sub_id = ds.subscribe(callback, min_wait=0, min_count=1) + assert sub_id in ds._pending_subscribers + assert ds.subscribers == {} + + ds.mark_started() + # Deferred subscriber has now been materialised. + assert ds._pending_subscribers == {} + assert sub_id in ds.subscribers + + ds.add_results([{"x": float(i), "y": float(i**2)} for i in range(5)]) + ds.mark_completed() + + assert len(received) > 0 + assert received[-1] == 5 + self._close_ds(ds) """get_parameter_data should read from the raw data file.""" ds, results = self._make_dataset_with_data(n_rows=5) data = ds.get_parameter_data() @@ -299,7 +369,7 @@ def test_missing_raw_data_file_raises(self, tmp_path: Path) -> None: """Loading a dataset whose raw data file is missing should raise.""" ds, _ = self._make_dataset_with_data(n_rows=3) run_id = ds.run_id - raw_path = Path(ds.metadata["raw_data_db_path"]) + raw_path = _raw_file(ds) self._close_ds(ds) # Delete the raw data file @@ -370,7 +440,7 @@ def test_update_after_move(self, tmp_path: Path) -> None: ds.add_results([{"x": 1.0, "y": 2.0}]) ds.mark_completed() - raw_path = Path(ds.metadata["raw_data_db_path"]) + raw_path = _raw_file(ds) db_path = ds.path_to_db assert db_path is not None self._close_ds(ds) @@ -460,7 +530,7 @@ def test_dry_run_reports_orphans(self, tmp_path: Path) -> None: ds.add_results([{"x": 1.0, "y": 2.0}]) ds.mark_completed() - raw_path = Path(ds.metadata["raw_data_db_path"]) + raw_path = _raw_file(ds) db_path = ds.path_to_db assert db_path is not None self._close_ds(ds) @@ -496,7 +566,7 @@ def test_removes_orphans_when_not_dry_run(self, tmp_path: Path) -> None: ds.add_results([{"x": 1.0, "y": 2.0}]) ds.mark_completed() - raw_path = Path(ds.metadata["raw_data_db_path"]) + raw_path = _raw_file(ds) db_path = ds.path_to_db assert db_path is not None run_id = ds.run_id @@ -570,7 +640,7 @@ def test_cleanup_by_sample_name(self, tmp_path: Path) -> None: ds1.mark_started() ds1.add_results([{"x": 1.0, "y": 2.0}]) ds1.mark_completed() - raw_path1 = Path(ds1.metadata["raw_data_db_path"]) + raw_path1 = _raw_file(ds1) db_path = ds1.path_to_db assert db_path is not None self._close_ds(ds1) @@ -582,7 +652,7 @@ def test_cleanup_by_sample_name(self, tmp_path: Path) -> None: ds2.mark_started() ds2.add_results([{"x": 3.0, "y": 4.0}]) ds2.mark_completed() - raw_path2 = Path(ds2.metadata["raw_data_db_path"]) + raw_path2 = _raw_file(ds2) self._close_ds(ds2) # Dry run first @@ -613,7 +683,7 @@ def test_cleanup_by_size(self, tmp_path: Path) -> None: for i in range(100): ds.add_results([{"x": float(i), "y": float(i * 2)}]) ds.mark_completed() - raw_path = Path(ds.metadata["raw_data_db_path"]) + raw_path = _raw_file(ds) db_path = ds.path_to_db assert db_path is not None self._close_ds(ds) @@ -733,6 +803,10 @@ def test_extract_runs_into_db_with_split_data(self, tmp_path: Path) -> None: data["y"]["y"], np.array([r["y"] for r in results]) ) assert target_ds.number_of_results == 5 + # The extracted dataset owns its data in the target DB and must not + # keep referring to the source's per-dataset raw data file. + assert get_raw_data_db_path_for_run(target_conn, target_ds.run_id) is None + assert "raw_data_db_path" not in target_ds.metadata target_conn.close() def test_export_and_create_metadata_db_with_split_data(