columnar series_stats in dataset manifest Bump cache_version 2 -> 3 and flip each shard's series_stats from a list of per-series dicts into eight parallel primitive lists (device_index, field_index, length, min_time, max_time, timeline_length, timeline_min_time, timeline_max_time). JSON parse time is dominated by the number of objects allocated; folding N per-series dicts into eight long primitive lists turns the hot rehydrate path from O(N) Python dicts into O(1) Python lists. In-memory MetadataCatalog.series_stats_by_ref keeps its dict-of-dict shape, so no downstream caller needs to change. Old v2 manifests fail the version check in read_manifest and are rebuilt on the next load.
diff --git a/python/tests/test_dataset_cache.py b/python/tests/test_dataset_cache.py index 82f4b30..f3ededd 100644 --- a/python/tests/test_dataset_cache.py +++ b/python/tests/test_dataset_cache.py
@@ -306,3 +306,45 @@ assert reader._reader is not None df.close() assert reader._reader is None + + +def test_manifest_uses_columnar_series_stats(tmp_path): + path = tmp_path / "columnar.tsfile" + _write_weather_file(path) + TsFileDataFrame(str(path), show_progress=False, cache="auto").close() + + with open(manifest_path([str(path)]), "r") as fh: + payload = json.load(fh) + + series_stats = payload["shards"][str(path)]["catalog"]["series_stats"] + assert isinstance(series_stats, dict) + expected_columns = { + "device_index", + "field_index", + "length", + "min_time", + "max_time", + "timeline_length", + "timeline_min_time", + "timeline_max_time", + } + assert set(series_stats.keys()) == expected_columns + lengths = {len(series_stats[col]) for col in expected_columns} + assert len(lengths) == 1 + + +def test_manifest_version_bump_invalidates_old_payload(tmp_path): + path = tmp_path / "old_version.tsfile" + _write_weather_file(path) + + TsFileDataFrame(str(path), show_progress=False, cache="auto").close() + mp = manifest_path([str(path)]) + + # Forge a manifest that uses the previous schema (cache_version=2). + with open(mp, "r") as fh: + payload = json.load(fh) + payload["cache_version"] = 2 + with open(mp, "w") as fh: + json.dump(payload, fh) + + assert read_manifest(mp) is None
diff --git a/python/tsfile/dataset/README.md b/python/tsfile/dataset/README.md index 4a14b6e..14b8ce0 100644 --- a/python/tsfile/dataset/README.md +++ b/python/tsfile/dataset/README.md
@@ -70,12 +70,12 @@ ```json { - "cache_version": 2, + "cache_version": 3, "shards": { "/abs/path/data/part_0.tsfile": { "size": 1234567, "mtime_ns": 1717050000000000000, - "catalog": { "tables": [...], "devices": [...], "series_stats": [...] } + "catalog": { "tables": [...], "devices": [...], "series_stats": {...} } }, "/abs/path/data/part_1.tsfile": { ... } } @@ -138,21 +138,24 @@ The position of a device in this array is its `device_index`, referenced by `series_stats[].device_index` below. -### `shards[<abs_path>].catalog.series_stats[]` +### `shards[<abs_path>].catalog.series_stats` -Each entry mirrors one `(device_id, field_idx) → stats` entry in -`MetadataCatalog.series_stats_by_ref`: +A columnar table: every key maps to a list of the same length, and the +``i``-th element across all eight lists describes one series. This shape +exists because JSON parsing time is dominated by the number of objects +allocated; folding ``N`` per-series dicts into eight long primitive lists +reduces a hot rehydrate path from O(N) Python dicts to O(1) Python lists. -| Field | Type | Notes | -|----------------------|--------------|------------------------------------------------------------------------------------| -| `device_index` | int | Index into the same shard's `catalog.devices[]`. | -| `field_index` | int | Index into the device's table's `field_columns`. | -| `length` | int | Row count derived from the value column's statistic. | -| `min_time` | int \| null | Minimum timestamp from the value column statistic. `null` when no statistic. | -| `max_time` | int \| null | Maximum timestamp from the value column statistic. `null` when no statistic. | -| `timeline_length` | int | Row count from the device's shared timeline statistic. Used for display + reads. | -| `timeline_min_time` | int \| null | Minimum timestamp from the device's shared timeline statistic. | -| `timeline_max_time` | int \| null | Maximum timestamp from the device's shared timeline statistic. | +| Column | Element type | Notes | +|----------------------|----------------|------------------------------------------------------------------------------------| +| `device_index` | int | Index into the same shard's `catalog.devices[]`. | +| `field_index` | int | Index into the device's table's `field_columns`. | +| `length` | int | Row count derived from the value column's statistic. | +| `min_time` | int \| null | Minimum timestamp from the value column statistic. `null` when no statistic. | +| `max_time` | int \| null | Maximum timestamp from the value column statistic. `null` when no statistic. | +| `timeline_length` | int | Row count from the device's shared timeline statistic. Used for display + reads. | +| `timeline_min_time` | int \| null | Minimum timestamp from the device's shared timeline statistic. | +| `timeline_max_time` | int \| null | Maximum timestamp from the device's shared timeline statistic. | For series with no data, every stat field is `0` or `null` — the slot is present so that `(device_index, field_index)` resolution always succeeds. @@ -165,7 +168,7 @@ 1. The manifest file exists and parses as JSON. 2. The top-level value is an object. -3. `cache_version` equals `_MANIFEST_VERSION` (currently `2`). +3. `cache_version` equals `_MANIFEST_VERSION` (currently `3`). 4. `shards` is an object. A manifest-level failure throws away the entire cache; every shard becomes @@ -267,7 +270,7 @@ ```json { - "cache_version": 2, + "cache_version": 3, "shards": { "/abs/path/data/weather.tsfile": { "size": 4096, @@ -289,18 +292,16 @@ "max_time": 2 } ], - "series_stats": [ - { - "device_index": 0, - "field_index": 0, - "length": 3, - "min_time": 0, - "max_time": 2, - "timeline_length": 3, - "timeline_min_time": 0, - "timeline_max_time": 2 - } - ] + "series_stats": { + "device_index": [0], + "field_index": [0], + "length": [3], + "min_time": [0], + "max_time": [2], + "timeline_length": [3], + "timeline_min_time": [0], + "timeline_max_time": [2] + } } } }
diff --git a/python/tsfile/dataset/cache.py b/python/tsfile/dataset/cache.py index a226991..a35142b 100644 --- a/python/tsfile/dataset/cache.py +++ b/python/tsfile/dataset/cache.py
@@ -34,7 +34,21 @@ from .metadata import MetadataCatalog _MANIFEST_FILENAME = "tsfile_dataset.metacache.json" -_MANIFEST_VERSION = 2 +# v3: series_stats stored as 8 parallel columns instead of a list of dicts. +# JSON parsing time is roughly proportional to the number of objects allocated, +# so flattening per-series dicts into long primitive lists cuts the dominant +# cost when rehydrating large shards. +_MANIFEST_VERSION = 3 +_SERIES_STATS_COLUMNS = ( + "device_index", + "field_index", + "length", + "min_time", + "max_time", + "timeline_length", + "timeline_min_time", + "timeline_max_time", +) def manifest_path(paths: List[str]) -> str: @@ -56,6 +70,25 @@ def catalog_to_dict(catalog: MetadataCatalog) -> dict: + n_series = len(catalog.series_stats_by_ref) + device_index = [0] * n_series + field_index = [0] * n_series + length = [0] * n_series + min_time = [None] * n_series + max_time = [None] * n_series + timeline_length = [0] * n_series + timeline_min_time = [None] * n_series + timeline_max_time = [None] * n_series + for i, ((d, f), stats) in enumerate(catalog.series_stats_by_ref.items()): + device_index[i] = d + field_index[i] = f + length[i] = stats["length"] + min_time[i] = stats["min_time"] + max_time[i] = stats["max_time"] + timeline_length[i] = stats["timeline_length"] + timeline_min_time[i] = stats["timeline_min_time"] + timeline_max_time[i] = stats["timeline_max_time"] + return { "tables": [ { @@ -75,19 +108,16 @@ } for d in catalog.device_entries ], - "series_stats": [ - { - "device_index": device_id, - "field_index": field_idx, - "length": stats["length"], - "min_time": stats["min_time"], - "max_time": stats["max_time"], - "timeline_length": stats["timeline_length"], - "timeline_min_time": stats["timeline_min_time"], - "timeline_max_time": stats["timeline_max_time"], - } - for (device_id, field_idx), stats in catalog.series_stats_by_ref.items() - ], + "series_stats": { + "device_index": device_index, + "field_index": field_index, + "length": length, + "min_time": min_time, + "max_time": max_time, + "timeline_length": timeline_length, + "timeline_min_time": timeline_min_time, + "timeline_max_time": timeline_max_time, + }, } @@ -107,15 +137,34 @@ device["min_time"], device["max_time"], ) - for stats in data["series_stats"]: - catalog.series_stats_by_ref[(stats["device_index"], stats["field_index"])] = { - "length": stats["length"], - "min_time": stats["min_time"], - "max_time": stats["max_time"], - "timeline_length": stats["timeline_length"], - "timeline_min_time": stats["timeline_min_time"], - "timeline_max_time": stats["timeline_max_time"], - } + + stats = data["series_stats"] + device_index = stats["device_index"] + field_index = stats["field_index"] + length = stats["length"] + min_time = stats["min_time"] + max_time = stats["max_time"] + timeline_length = stats["timeline_length"] + timeline_min_time = stats["timeline_min_time"] + timeline_max_time = stats["timeline_max_time"] + series_stats_by_ref = catalog.series_stats_by_ref + # Tight loop: all hot locals are bound and the per-row dict is built via + # one C-level call (`dict(zip(...))`) instead of six bytecode setitems. + keys = _SERIES_STATS_COLUMNS[2:] + for i, d in enumerate(device_index): + series_stats_by_ref[(d, field_index[i])] = dict( + zip( + keys, + ( + length[i], + min_time[i], + max_time[i], + timeline_length[i], + timeline_min_time[i], + timeline_max_time[i], + ), + ) + ) return catalog