This is an automated email from the ASF dual-hosted git repository. ColinLeeo pushed a commit to branch opt_dataframe in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit 8490458228565d2fbbd50edbdcef090bb9004f16 Author: Le Yang <[email protected]> AuthorDate: Thu Jul 2 10:06:04 2026 +0800 [Python] Support tree-model TsFiles in TsFileDataFrame (#816) * [Python] Support tree-model TsFiles in TsFileDataFrame Squash of the tsdf-tree branch: tree-model support plus dataset catalog memory and indexing improvements. - support tree-model TsFiles in TsFileDataFrame - shrink the dataframe catalog memory footprint - drop table-model phantom cells - inline iter_owned_series_refs - rename _LogicalIndex to _DataFrameCatalog - dedup repeated series specifiers in tsdf.loc * Fix tree-model metadata segments * [Python] Dedupe dataset model constants and harden tree-model reads - metadata: define MODEL_TABLE/MODEL_TREE once and import them in reader and dataframe instead of redefining the same literals in both modules, removing the drift risk between duplicate definitions. - metadata: MetadataCatalog.series_count now returns the count of physically-present series (len(series_stats_by_ref)) instead of the device x declared-field cross-product, so it matches len(tsdf) and the reader property for sparse schemas; TsFileSeriesReader.series_count delegates to it as the single source of truth. - reader: document why _metadata_field_stats gates on the value statistic's non-null row_count (not the timeline) and why the timeline_statistic reads below cannot raise. - reader: mark the tree-model full-scan reads as O(total_rows) per device (cwrapper per-device pushdown limitation) so the hot path is easy to locate when profiling. - reader: fail fast with an assertion if the cwrapper stale col_ leak pattern changes, instead of silently slicing the wrong path columns and returning wrong-device data. - tests: add a sparse-schema regression assertion locking series_count to the physically-present count. * [Python] Support unioning multiple tree-model TsFiles in TsFileDataFrame - Tree TsFiles synthesize their schema per file, so loading several into one TsFileDataFrame previously raised "Incompatible schema" whenever the files differed in measurement subset or device depth. Widen the dataframe-level synthetic table to the union instead: _merge_tree_table_entries unions the field columns (existing order first) and grows the tag columns to the deepest file's depth so shallower devices pad with nulls. - Key each logical series by the global field index resolved by measurement name against the merged table, while each shard keeps its reader-local field index for the read path. This keeps per-(device, field) ownership exact and avoids mis-mapping measurements across files. For table model the global and local indices always coincide, so behavior is unchanged (incompatible table schemas are still rejected). - tests: add multi-file tree coverage -- identical-structure merge (shards and time bounds), field union across files, and differing-depth union with tag padding. * [Python] Filter non-numeric measurements in tree-model TsFileDataFrame The dataset surface is numeric (values are read as float64), and table mode already drops non-numeric (STRING/TEXT) fields from its schema. Tree mode did not, so a string/text measurement was surfaced as a series that then crashed on read (float() on a string). Apply the numeric-type filter once in _metadata_field_stats so both models behave consistently -- this also makes the existing tree-model "measurements that pass the numeric filter" comment accurate. - reader: skip measurements whose data_type is not numeric in _metadata_field_stats (shared by table and tree paths); table mode is unaffected since it already filters at the schema level. - tests: add a tree-model regression test asserting a STRING measurement is dropped (not listed, raises KeyError) while the numeric one reads back. --- python/tests/test_tsfile_dataset.py | 510 +++++++++++++++++++++++++++++++----- python/tsfile/dataset/dataframe.py | 385 ++++++++++++++++++--------- python/tsfile/dataset/formatting.py | 31 ++- python/tsfile/dataset/metadata.py | 27 +- python/tsfile/dataset/reader.py | 418 +++++++++++++++++++++++++---- 5 files changed, 1129 insertions(+), 242 deletions(-) diff --git a/python/tests/test_tsfile_dataset.py b/python/tests/test_tsfile_dataset.py index f58d5117c..12b847e9e 100644 --- a/python/tests/test_tsfile_dataset.py +++ b/python/tests/test_tsfile_dataset.py @@ -32,10 +32,15 @@ from tsfile import AlignedTimeseries, SeriesPath, Timeseries, TsFileDataFrame from tsfile.dataset.formatting import format_timestamp from tsfile.dataset.metadata import ( MetadataCatalog, + SeriesStats, build_series_path, resolve_series_path, ) -from tsfile.dataset.reader import TsFileSeriesReader, _build_exact_tag_filter +from tsfile.dataset.reader import ( + MODEL_TABLE, + TsFileSeriesReader, + _build_exact_tag_filter, +) def _write_weather_file(path, start): @@ -247,7 +252,7 @@ def test_dataset_basic_access_patterns(tmp_path, capsys): assert list(tsdf["field"]) == ["temperature", "humidity"] - assert "TsFileDataFrame(2 time series, 2 files)" in repr(tsdf) + assert "TsFileDataFrame(table model, 2 time series, 2 files)" in repr(tsdf) aligned.show(2) assert "AlignedTimeseries(6 rows, 2 series)" in capsys.readouterr().out @@ -490,6 +495,54 @@ def test_dataset_loc_supports_single_timestamp_and_mixed_series_specifiers(tmp_p np.testing.assert_array_equal(aligned.values, np.array([[21.5, 52.0]])) +def test_dataset_loc_dedups_repeated_series_specifiers(tmp_path): + path = tmp_path / "weather.tsfile" + _write_weather_file(path, 0) + + with TsFileDataFrame(str(path), show_progress=False) as tsdf: + humidity = "weather.device_a.humidity" + humidity_idx = tsdf.list_timeseries().index(humidity) + + # 1) name + matching idx pointing at the same series. + aligned_two_dup = tsdf.loc[0:2, [humidity, humidity_idx]] + assert aligned_two_dup.shape == (3, 2) + np.testing.assert_array_equal( + aligned_two_dup.timestamps, np.array([0, 1, 2], dtype=np.int64) + ) + np.testing.assert_array_equal( + aligned_two_dup.values, + np.array([[50.0, 50.0], [52.0, 52.0], [55.0, 55.0]]), + ) + + # 2) same name twice -- single-group, single-key dedup path. + aligned_name_twice = tsdf.loc[0:2, [humidity, humidity]] + assert aligned_name_twice.shape == (3, 2) + np.testing.assert_array_equal( + aligned_name_twice.values, + np.array([[50.0, 50.0], [52.0, 52.0], [55.0, 55.0]]), + ) + + # 3) duplicate among other distinct series must not regress the + # historically-passing case either. + aligned_mixed = tsdf.loc[ + 0:2, [humidity, "weather.device_a.temperature", humidity_idx] + ] + assert aligned_mixed.shape == (3, 3) + np.testing.assert_array_equal( + aligned_mixed.timestamps, np.array([0, 1, 2], dtype=np.int64) + ) + np.testing.assert_array_equal( + aligned_mixed.values, + np.array( + [ + [50.0, 20.0, 50.0], + [52.0, 21.5, 52.0], + [55.0, 23.0, 55.0], + ] + ), + ) + + def test_dataset_loc_supports_open_ended_ranges_and_negative_series_index(tmp_path): path = tmp_path / "weather.tsfile" _write_weather_file(path, 100) @@ -570,7 +623,7 @@ def test_dataset_repr_only_builds_preview_rows(tmp_path, monkeypatch): _write_weather_file(path, 0) with TsFileDataFrame(str(path), show_progress=False) as tsdf: - tsdf._index.series_refs_ordered = [(0, 0)] * 1000 + tsdf._index.series = [(0, 0)] * 1000 built_rows = [] @@ -595,7 +648,7 @@ def test_dataset_repr_only_builds_preview_rows(tmp_path, monkeypatch): monkeypatch.setattr(tsdf, "_build_series_name", fail_build_series_name) rendered = repr(tsdf) - assert "TsFileDataFrame(1000 time series, 1 files)" in rendered + assert "TsFileDataFrame(table model, 1000 time series, 1 files)" in rendered assert "..." in rendered assert len(built_rows) == 20 @@ -622,6 +675,88 @@ def test_dataset_exposes_only_numeric_fields_and_keeps_nan(tmp_path): assert series[1:1].shape == (0,) +def test_dataset_omits_table_model_phantom_series_for_skipped_cells(tmp_path): + """Schema-declared fields that a device never wrote must NOT appear. + + The dataset surface treats a series as "data physically written for one + (device, field) pair". A Tablet that skips ``add_value_by_name`` for a + column produces a chunk with ``length=0``; that cell is not a real series + and must not be exposed via ``list_timeseries`` / ``len(tsdf)`` / + ``series_shards`` -- table-model and tree-model behave identically here. + """ + from tsfile import Tablet + + path = tmp_path / "sparse_table.tsfile" + schema = TableSchema( + "bench", + [ + ColumnSchema("device", TSDataType.STRING, ColumnCategory.TAG), + ColumnSchema("v1", TSDataType.DOUBLE, ColumnCategory.FIELD), + ColumnSchema("v2", TSDataType.DOUBLE, ColumnCategory.FIELD), + ColumnSchema("v3", TSDataType.DOUBLE, ColumnCategory.FIELD), + ], + ) + with TsFileTableWriter(str(path), schema) as writer: + # d1: writes only v1 and v2 (skip v3) + t1 = Tablet( + ["device", "v1", "v2", "v3"], + [ + TSDataType.STRING, + TSDataType.DOUBLE, + TSDataType.DOUBLE, + TSDataType.DOUBLE, + ], + 1, + ) + t1.add_timestamp(0, 1) + t1.add_value_by_name("device", 0, "d1") + t1.add_value_by_name("v1", 0, 100.0) + t1.add_value_by_name("v2", 0, 200.0) + writer.write_table(t1) + + # d2: writes only v1 and v3 (skip v2) + t2 = Tablet( + ["device", "v1", "v2", "v3"], + [ + TSDataType.STRING, + TSDataType.DOUBLE, + TSDataType.DOUBLE, + TSDataType.DOUBLE, + ], + 1, + ) + t2.add_timestamp(0, 2) + t2.add_value_by_name("device", 0, "d2") + t2.add_value_by_name("v1", 0, 110.0) + t2.add_value_by_name("v3", 0, 330.0) + writer.write_table(t2) + + with TsFileDataFrame(str(path), show_progress=False) as tsdf: + # 4 real cells: (d1,v1), (d1,v2), (d2,v1), (d2,v3); NO phantoms. + assert len(tsdf) == 4 + assert sorted(tsdf.list_timeseries()) == [ + "bench.d1.v1", + "bench.d1.v2", + "bench.d2.v1", + "bench.d2.v3", + ] + + with pytest.raises(KeyError): + tsdf["bench.d1.v3"] + with pytest.raises(KeyError): + tsdf["bench.d2.v2"] + + # series_count must report the 4 physically-present series, not the 2x3 + # schema cross-product -- regression guard for the sparse-schema case + # (catalog and reader must agree). + reader = TsFileSeriesReader(str(path), show_progress=False) + try: + assert reader.series_count == 4 + assert reader.catalog.series_count == 4 + finally: + reader.close() + + def test_dataset_timeseries_supports_negative_step_slices(tmp_path): path = tmp_path / "weather.tsfile" _write_weather_file(path, 0) @@ -934,6 +1069,7 @@ def test_reader_read_series_by_row_retries_across_native_row_query_boundaries(): ) reader = object.__new__(TsFileSeriesReader) + reader._model_kind = MODEL_TABLE reader._reader = _FakeNativeReader( np.arange(30, dtype=np.int64), np.arange(30, dtype=np.float64), boundary=10 ) @@ -955,14 +1091,14 @@ def test_series_path_resolution_allows_prefix_tag_values(): ("temperature",), ) device_id = catalog.add_device(table_id, ("site_a", "device_a"), 0, 1) - catalog.series_stats_by_ref[(device_id, 0)] = { - "length": 1, - "min_time": 0, - "max_time": 0, - "timeline_length": 1, - "timeline_min_time": 0, - "timeline_max_time": 0, - } + catalog.series_stats_by_ref[(device_id, 0)] = SeriesStats( + length=1, + min_time=0, + max_time=0, + timeline_length=1, + timeline_min_time=0, + timeline_max_time=0, + ) series_path = build_series_path(catalog, device_id, 0) assert series_path == "weather.site_a.device_a.temperature" @@ -978,14 +1114,14 @@ def test_series_path_resolution_allows_missing_trailing_tag_value(): ("temperature",), ) device_id = catalog.add_device(table_id, (), 0, 1) - catalog.series_stats_by_ref[(device_id, 0)] = { - "length": 1, - "min_time": 0, - "max_time": 0, - "timeline_length": 1, - "timeline_min_time": 0, - "timeline_max_time": 0, - } + catalog.series_stats_by_ref[(device_id, 0)] = SeriesStats( + length=1, + min_time=0, + max_time=0, + timeline_length=1, + timeline_min_time=0, + timeline_max_time=0, + ) series_path = build_series_path(catalog, device_id, 0) assert series_path == "weather.temperature" @@ -1001,14 +1137,14 @@ def test_series_path_resolution_uses_named_tags_for_sparse_non_prefix_values(): ("temperature",), ) device_id = catalog.add_device(table_id, (None, "device_a", None), 0, 1) - catalog.series_stats_by_ref[(device_id, 0)] = { - "length": 1, - "min_time": 0, - "max_time": 0, - "timeline_length": 1, - "timeline_min_time": 0, - "timeline_max_time": 0, - } + catalog.series_stats_by_ref[(device_id, 0)] = SeriesStats( + length=1, + min_time=0, + max_time=0, + timeline_length=1, + timeline_min_time=0, + timeline_max_time=0, + ) series_path = build_series_path(catalog, device_id, 0) # The leading null tag is preserved at its position via the \N marker. @@ -1114,6 +1250,7 @@ def test_reader_exact_match_with_none_tag_values_issues_is_null_query(): return _EmptyResultSet() reader = object.__new__(TsFileSeriesReader) + reader._model_kind = MODEL_TABLE reader._reader = _FakeNativeReader() reader._catalog = MetadataCatalog() table_id = reader._catalog.add_table( @@ -1123,14 +1260,14 @@ def test_reader_exact_match_with_none_tag_values_issues_is_null_query(): ("temperature",), ) device_id = reader._catalog.add_device(table_id, (None, "device_a", "north"), 0, 1) - reader._catalog.series_stats_by_ref[(device_id, 0)] = { - "length": 2, - "min_time": 0, - "max_time": 1, - "timeline_length": 2, - "timeline_min_time": 0, - "timeline_max_time": 1, - } + reader._catalog.series_stats_by_ref[(device_id, 0)] = SeriesStats( + length=2, + min_time=0, + max_time=1, + timeline_length=2, + timeline_min_time=0, + timeline_max_time=1, + ) # Both read paths now issue a native query that encodes the null tag as # IS NULL instead of failing fast. @@ -1143,7 +1280,7 @@ def test_reader_exact_match_with_none_tag_values_issues_is_null_query(): def test_dataframe_resolves_named_sparse_tag_series_path(): tsdf = object.__new__(TsFileDataFrame) - tsdf._index = dataframe_module._LogicalIndex() + tsdf._index = dataframe_module._DataFrameCatalog() tsdf._index.table_entries["weather"] = dataframe_module.TableEntry( table_name="weather", tag_columns=("city", "device", "region"), @@ -1151,12 +1288,11 @@ def test_dataframe_resolves_named_sparse_tag_series_path(): field_columns=("temperature",), ) device_key = ("weather", (None, "device_a")) - tsdf._index.device_order = [device_key] - tsdf._index.device_index_by_key = {device_key: 0} - tsdf._index.device_refs = [[]] - tsdf._index.series_refs_ordered = [(0, 0)] - tsdf._index.series_ref_set = {(0, 0)} - tsdf._index.series_ref_map = {(0, 0): []} + tsdf._index.devices = [device_key] + tsdf._index.device_index = {device_key: 0} + tsdf._index.device_time_bounds = [(0, 1)] + tsdf._index.series = [(0, 0)] + tsdf._index.series_shards = {(0, 0): []} assert tsdf.list_timeseries() == ["weather.\\N.device_a.temperature"] # Resolvable by the \N string form and by the returned SeriesPath itself. @@ -1166,25 +1302,24 @@ def test_dataframe_resolves_named_sparse_tag_series_path(): def test_dataframe_list_timeseries_filters_named_sparse_tag_prefix(): tsdf = object.__new__(TsFileDataFrame) - tsdf._index = dataframe_module._LogicalIndex() + tsdf._index = dataframe_module._DataFrameCatalog() tsdf._index.table_entries["weather"] = dataframe_module.TableEntry( table_name="weather", tag_columns=("city", "device", "region"), tag_types=(TSDataType.STRING, TSDataType.STRING, TSDataType.STRING), field_columns=("temperature",), ) - tsdf._index.device_order = [ + tsdf._index.devices = [ ("weather", (None, "device_a")), ("weather", ("beijing", "device_b")), ] - tsdf._index.device_index_by_key = { + tsdf._index.device_index = { ("weather", (None, "device_a")): 0, ("weather", ("beijing", "device_b")): 1, } - tsdf._index.device_refs = [[], []] - tsdf._index.series_refs_ordered = [(0, 0), (1, 0)] - tsdf._index.series_ref_set = {(0, 0), (1, 0)} - tsdf._index.series_ref_map = {(0, 0): [], (1, 0): []} + tsdf._index.device_time_bounds = [(0, 1), (0, 1)] + tsdf._index.series = [(0, 0), (1, 0)] + tsdf._index.series_shards = {(0, 0): [], (1, 0): []} # Prefix matching is position-aware: "weather.\N" selects the null-city # device, "weather.beijing" selects the fully specified one. @@ -1201,7 +1336,7 @@ def test_dataframe_list_timeseries_prefix_can_skip_full_name_build( _write_weather_file(path, 0) with TsFileDataFrame(str(path), show_progress=False) as tsdf: - tsdf._index.series_refs_ordered = [(0, 0)] * 1000 + tsdf._index.series = [(0, 0)] * 1000 def fail_build_series_name(_series_ref): raise AssertionError( @@ -1223,14 +1358,14 @@ def test_series_path_resolution_distinguishes_null_position(): first_id = catalog.add_device(table_id, ("beijing", None), 0, 1) # device IS NULL second_id = catalog.add_device(table_id, (None, "beijing"), 0, 1) # city IS NULL for device_id in (first_id, second_id): - catalog.series_stats_by_ref[(device_id, 0)] = { - "length": 1, - "min_time": 0, - "max_time": 0, - "timeline_length": 1, - "timeline_min_time": 0, - "timeline_max_time": 0, - } + catalog.series_stats_by_ref[(device_id, 0)] = SeriesStats( + length=1, + min_time=0, + max_time=0, + timeline_length=1, + timeline_min_time=0, + timeline_max_time=0, + ) # Null position is preserved, so these two devices get distinct paths # (previously both compressed to "weather.beijing.temperature" -> ambiguous). @@ -1270,3 +1405,260 @@ def test_dataframe_parallel_show_progress_reports_start_immediately(tmp_path, ca stderr = capsys.readouterr().err assert "Loading TsFile shards: 0/2" in stderr assert "Loading TsFile shards: 2/2 (4 series) ... done" in stderr + + +# --- Tree-model tests ------------------------------------------------------- + + +def _write_tree_file(path): + """Tree-model TsFile with two devices; the second device is shorter and + only has one of the two declared measurements, exercising None-pad + + union-field paths in the synthetic table layer. + """ + from tsfile import ( + Field, + RowRecord, + TimeseriesSchema, + TsFileWriter, + ) + + writer = TsFileWriter(str(path)) + writer.register_timeseries( + "root.ln.wf01.wt01", TimeseriesSchema("status", TSDataType.INT32) + ) + writer.register_timeseries( + "root.ln.wf01.wt01", TimeseriesSchema("temperature", TSDataType.DOUBLE) + ) + writer.register_timeseries( + "root.ln.wf02.wt02", TimeseriesSchema("status", TSDataType.INT32) + ) + for t in range(5): + writer.write_row_record( + RowRecord( + "root.ln.wf01.wt01", + t, + [ + Field("status", t, TSDataType.INT32), + Field("temperature", float(t) + 0.5, TSDataType.DOUBLE), + ], + ) + ) + writer.write_row_record( + RowRecord( + "root.ln.wf02.wt02", + t, + [Field("status", t * 2, TSDataType.INT32)], + ) + ) + writer.close() + + +def test_dataset_tree_model_metadata_and_repr(tmp_path): + path = tmp_path / "tree.tsfile" + _write_tree_file(path) + + with TsFileDataFrame(str(path), show_progress=False) as tsdf: + assert tsdf.model == "tree" + assert len(tsdf) == 3 + assert sorted(tsdf.list_timeseries()) == [ + "root.ln.wf01.wt01.status", + "root.ln.wf01.wt01.temperature", + "root.ln.wf02.wt02.status", + ] + + rendered = repr(tsdf) + # Header carries the model marker; tag headers use _col_i (1-based). + assert "TsFileDataFrame(tree model, 3 time series, 1 files)" in rendered + assert "_col_1" in rendered and "_col_2" in rendered and "_col_3" in rendered + assert "table" not in rendered.splitlines()[1] # no 'table' header + + # Metadata column projection: _col_i and field; 'table' is rejected. + assert list(tsdf["_col_1"]) == ["ln", "ln", "ln"] + assert list(tsdf["_col_3"]) == ["wt01", "wt01", "wt02"] + assert list(tsdf["field"]) == ["status", "temperature", "status"] + with pytest.raises(KeyError): + tsdf["table"] + + +def test_dataset_tree_model_series_access(tmp_path): + path = tmp_path / "tree.tsfile" + _write_tree_file(path) + + with TsFileDataFrame(str(path), show_progress=False) as tsdf: + ts = tsdf["root.ln.wf01.wt01.temperature"] + assert isinstance(ts, Timeseries) + assert ts.name == "root.ln.wf01.wt01.temperature" + assert len(ts) == 5 + np.testing.assert_array_equal(ts.timestamps, np.arange(5, dtype=np.int64)) + # __getitem__ slice routes through _read_series_by_row_tree. + first_three = ts[0:3] + np.testing.assert_array_equal(first_three, np.array([0.5, 1.5, 2.5])) + + # Aligned read across two co-located series. + aligned = tsdf.loc[ + 0:5, + [ + "root.ln.wf01.wt01.temperature", + "root.ln.wf01.wt01.status", + ], + ] + assert isinstance(aligned, AlignedTimeseries) + assert aligned.shape == (5, 2) + np.testing.assert_array_equal(aligned.timestamps, np.arange(5, dtype=np.int64)) + + +def test_dataset_tree_model_list_timeseries_metadata(tmp_path): + path = tmp_path / "tree.tsfile" + _write_tree_file(path) + + with TsFileDataFrame(str(path), show_progress=False) as tsdf: + meta = tsdf.list_timeseries_metadata() + assert isinstance(meta, pd.DataFrame) + assert list(meta.columns) == [ + "field", + "start_time", + "end_time", + "count", + "_col_1", + "_col_2", + "_col_3", + ] + assert sorted(meta.index.tolist()) == sorted(tsdf.list_timeseries()) + # Time bounds surface as pandas.Timestamp for ergonomic comparison. + assert pd.api.types.is_datetime64_any_dtype(meta["start_time"]) + assert pd.api.types.is_datetime64_any_dtype(meta["end_time"]) + # Per-series count comes from the catalog, not the synthetic union. + assert meta.loc["root.ln.wf01.wt01.temperature", "count"] == 5 + assert meta.loc["root.ln.wf02.wt02.status", "count"] == 5 + + +def test_dataset_rejects_mixed_model_load(tmp_path): + table_path = tmp_path / "weather.tsfile" + tree_path = tmp_path / "tree.tsfile" + _write_weather_file(table_path, 0) + _write_tree_file(tree_path) + + with pytest.raises(ValueError, match="Mixed table-model and tree-model"): + TsFileDataFrame([str(table_path), str(tree_path)], show_progress=False) + + +def _write_tree_rows(path, device_measurements, t_start=0, t_count=3): + """Write a tree-model TsFile from a device -> [(measurement, dtype)] map. + + Each device's measurements are written for timestamps + ``t_start .. t_start + t_count - 1``; numeric values are ``float(t) + 0.5`` + (INT32 measurements use ``t``), so series read back deterministically. + """ + from tsfile import Field, RowRecord, TimeseriesSchema, TsFileWriter + + writer = TsFileWriter(str(path)) + for device, measurements in device_measurements.items(): + for name, dtype in measurements: + writer.register_timeseries(device, TimeseriesSchema(name, dtype)) + for t in range(t_start, t_start + t_count): + for device, measurements in device_measurements.items(): + fields = [ + Field(name, (t if dtype == TSDataType.INT32 else float(t) + 0.5), dtype) + for name, dtype in measurements + ] + writer.write_row_record(RowRecord(device, t, fields)) + writer.close() + + +def test_dataset_tree_model_merges_identical_structure_across_files(tmp_path): + # Two tree files, same device/measurement, disjoint time ranges: one logical + # series whose shards and time bounds merge. + path1 = tmp_path / "t1.tsfile" + path2 = tmp_path / "t2.tsfile" + _write_tree_rows(path1, {"root.a.b": [("m1", TSDataType.DOUBLE)]}, t_start=0) + _write_tree_rows(path2, {"root.a.b": [("m1", TSDataType.DOUBLE)]}, t_start=10) + + with TsFileDataFrame([str(path1), str(path2)], show_progress=False) as tsdf: + assert tsdf.model == "tree" + assert tsdf.list_timeseries() == ["root.a.b.m1"] + ts = tsdf["root.a.b.m1"] + assert len(ts) == 6 + np.testing.assert_array_equal( + ts.timestamps, np.array([0, 1, 2, 10, 11, 12], dtype=np.int64) + ) + meta = tsdf.list_timeseries_metadata() + assert meta.loc["root.a.b.m1", "count"] == 6 + + +def test_dataset_tree_model_unions_fields_across_files(tmp_path): + # Same device, different measurement subsets across files -> union of fields. + path1 = tmp_path / "t1.tsfile" + path2 = tmp_path / "t2.tsfile" + _write_tree_rows(path1, {"root.a.b": [("m1", TSDataType.DOUBLE)]}, t_start=0) + _write_tree_rows(path2, {"root.a.b": [("m2", TSDataType.DOUBLE)]}, t_start=0) + + with TsFileDataFrame([str(path1), str(path2)], show_progress=False) as tsdf: + assert tsdf.model == "tree" + assert sorted(tsdf.list_timeseries()) == ["root.a.b.m1", "root.a.b.m2"] + # Each field reads its own file's data, not the other's. + np.testing.assert_array_equal(tsdf["root.a.b.m1"][:], np.array([0.5, 1.5, 2.5])) + np.testing.assert_array_equal(tsdf["root.a.b.m2"][:], np.array([0.5, 1.5, 2.5])) + # Aligned read spans the unioned fields on the one device. + aligned = tsdf.loc[0:2, ["root.a.b.m1", "root.a.b.m2"]] + assert aligned.shape == (3, 2) + np.testing.assert_array_equal( + aligned.timestamps, np.array([0, 1, 2], dtype=np.int64) + ) + + +def test_dataset_tree_model_unions_different_depths_across_files(tmp_path): + # Files with different max depth -> global tag layout widens; the shallower + # device pads its deepest tag column with null. + path1 = tmp_path / "t1.tsfile" + path2 = tmp_path / "t2.tsfile" + _write_tree_rows(path1, {"root.a.b": [("m1", TSDataType.DOUBLE)]}, t_start=0) + _write_tree_rows(path2, {"root.a.b.c": [("m1", TSDataType.DOUBLE)]}, t_start=0) + + with TsFileDataFrame([str(path1), str(path2)], show_progress=False) as tsdf: + assert tsdf.model == "tree" + assert sorted(tsdf.list_timeseries()) == ["root.a.b.c.m1", "root.a.b.m1"] + meta = tsdf.list_timeseries_metadata() + # Global depth widened to the deeper file (a, b, c -> _col_3 exists). + assert "_col_3" in meta.columns + assert meta.loc["root.a.b.c.m1", "_col_3"] == "c" + # The shallower device's deepest tag column is null (padded). + assert pd.isna(meta.loc["root.a.b.m1", "_col_3"]) + # Each device still reads its own data. + np.testing.assert_array_equal(tsdf["root.a.b.m1"][:], np.array([0.5, 1.5, 2.5])) + np.testing.assert_array_equal( + tsdf["root.a.b.c.m1"][:], np.array([0.5, 1.5, 2.5]) + ) + + +def test_dataset_tree_model_omits_non_numeric_measurements(tmp_path): + # The dataset surface is numeric (float64); a STRING tree measurement must + # be dropped, not surfaced as a series that crashes on read. + from tsfile import Field, RowRecord, TimeseriesSchema, TsFileWriter + + path = tmp_path / "tree_mixed.tsfile" + writer = TsFileWriter(str(path)) + writer.register_timeseries("root.a.b", TimeseriesSchema("temp", TSDataType.DOUBLE)) + writer.register_timeseries( + "root.a.b", TimeseriesSchema("status", TSDataType.STRING) + ) + for t in range(3): + writer.write_row_record( + RowRecord( + "root.a.b", + t, + [ + Field("temp", float(t) + 0.5, TSDataType.DOUBLE), + Field("status", "ok", TSDataType.STRING), + ], + ) + ) + writer.close() + + with TsFileDataFrame(str(path), show_progress=False) as tsdf: + # Only the numeric measurement is exposed; the STRING one is dropped. + assert tsdf.list_timeseries() == ["root.a.b.temp"] + with pytest.raises(KeyError): + tsdf["root.a.b.status"] + np.testing.assert_array_equal( + tsdf["root.a.b.temp"][:], np.array([0.5, 1.5, 2.5]) + ) diff --git a/python/tsfile/dataset/dataframe.py b/python/tsfile/dataset/dataframe.py index 77c278289..a65b231bc 100644 --- a/python/tsfile/dataset/dataframe.py +++ b/python/tsfile/dataset/dataframe.py @@ -23,13 +23,15 @@ from dataclasses import dataclass, field import heapq import os import sys -from typing import Dict, List, Set, Tuple, Union +from typing import Dict, List, Optional, Tuple, Union import warnings import numpy as np from .formatting import format_dataframe_table from .metadata import ( + MODEL_TABLE, + MODEL_TREE, SeriesPath, TableEntry, _normalize_tag_values, @@ -43,7 +45,6 @@ from .timeseries import AlignedTimeseries, Timeseries DeviceKey = Tuple[str, tuple] SeriesRefKey = Tuple[int, int] SeriesRef = Tuple[object, int, int] -DeviceRef = Tuple[object, int] _QUERY_START = np.iinfo(np.int64).min _QUERY_END = np.iinfo(np.int64).max @@ -56,35 +57,35 @@ _OVERLAP_ROW_CHUNK_SIZE = 256 @dataclass(**_DATACLASS_SLOTS) -class _LogicalIndex: - """Cross-reader logical mapping for devices and series.""" +class _DataFrameCatalog: + """TsFileDataFrame's cross-file unified catalog: merges each tsfile's + ``MetadataCatalog`` into one user-facing global view. + """ + + # Model kind for the entire load set: "table" or "tree". A single + # TsFileDataFrame is not allowed to mix table-model and tree-model files. + model: Optional[str] = None # Shared table schema references keyed by table name. table_entries: Dict[str, TableEntry] = field(default_factory=dict) # Stable logical device order, each item is (table_name, tag_values). - device_order: List[DeviceKey] = field(default_factory=list) + devices: List[DeviceKey] = field(default_factory=list) # Map one logical device key to its dataframe-local device index. The key's # tag tuple keeps interior nulls (None) and drops trailing ones, so every # device -- including null-tagged ones -- resolves by a single direct lookup. - device_index_by_key: Dict[DeviceKey, int] = field(default_factory=dict) - # For each logical device, keep the contributing reader-local device refs. - device_refs: List[List[DeviceRef]] = field(default_factory=list) + device_index: Dict[DeviceKey, int] = field(default_factory=dict) + # Aggregated (min_time, max_time) per logical device, computed once at + # registration so query-time lookups are O(1). + device_time_bounds: List[Tuple[Optional[int], Optional[int]]] = field( + default_factory=list + ) # Stable logical series order, each item is (device_idx, field_idx). - series_refs_ordered: List[SeriesRefKey] = field(default_factory=list) - # Map one logical series ref to the contributing reader-local series refs. - series_ref_map: Dict[SeriesRefKey, List[SeriesRef]] = field(default_factory=dict) - # Fast membership check for resolved series refs. - series_ref_set: Set[SeriesRefKey] = field(default_factory=set) - - -@dataclass(**_DATACLASS_SLOTS) -class _DerivedCache: - """Merged metadata derived from the logical index.""" - - devices: List[dict] = field(default_factory=list) - field_stats: Dict[SeriesRefKey, dict] = field(default_factory=dict) + series: List[SeriesRefKey] = field(default_factory=list) + # For each logical series: which shards hold it, plus its (device_id, + # field_idx) coordinates inside each shard. + series_shards: Dict[SeriesRefKey, List[SeriesRef]] = field(default_factory=dict) def _expand_paths(paths: Union[str, List[str]]) -> List[str]: @@ -139,13 +140,56 @@ def _validate_table_schema( ) +def _merge_tree_table_entries(existing: TableEntry, incoming: TableEntry) -> TableEntry: + """Widen two per-file synthetic tree tables into one global table. + + Tree TsFiles carry no authored schema, so each file synthesizes its own + table (root name, ``_col_i`` tag columns sized to that file's deepest + device, fields = the measurements present in that file). When several tree + files are loaded together we widen to their union: field columns are + unioned (existing order first, then new measurements) and the tag + columns/types grow to the deepest file's depth so shallower devices pad + with nulls. Per-(device, field) ownership stays exact because each shard is + keyed by the global field index resolved by measurement name. + """ + if len(incoming.tag_columns) > len(existing.tag_columns): + tag_columns, tag_types = incoming.tag_columns, incoming.tag_types + else: + tag_columns, tag_types = existing.tag_columns, existing.tag_types + + field_columns = list(existing.field_columns) + seen = set(field_columns) + for name in incoming.field_columns: + if name not in seen: + seen.add(name) + field_columns.append(name) + + return TableEntry( + table_name=existing.table_name, + tag_columns=tag_columns, + tag_types=tag_types, + field_columns=tuple(field_columns), + ) + + def _register_reader( readers: Dict[str, object], - index: _LogicalIndex, + index: _DataFrameCatalog, file_path: str, reader, ) -> None: """Merge one reader's catalog into the dataframe-wide logical index.""" + cur_tsfile_model = reader.model_kind + if index.model is None: + index.model = cur_tsfile_model + elif index.model != cur_tsfile_model: + raise ValueError( + f"Mixed table-model and tree-model TsFiles detected. The first " + f"loaded file is {index.model!r} but '{file_path}' is " + f"{cur_tsfile_model!r}. A single TsFileDataFrame load set must be " + f"entirely table-model or entirely tree-model." + ) + readers[file_path] = reader catalog = reader.catalog @@ -153,44 +197,61 @@ def _register_reader( existing_entry = index.table_entries.get(table_entry.table_name) if existing_entry is None: index.table_entries[table_entry.table_name] = table_entry + elif index.model == MODEL_TREE: + # Tree shards carry no authored schema; widen the synthetic table to + # the union of fields and the deepest tag layout across files + # instead of rejecting differing subsets/depths. + index.table_entries[table_entry.table_name] = _merge_tree_table_entries( + existing_entry, table_entry + ) else: _validate_table_schema(existing_entry, table_entry, file_path) for device_id, device_entry in enumerate(catalog.device_entries): table_entry = catalog.table_entries[device_entry.table_id] device_key = (table_entry.table_name, tuple(device_entry.tag_values)) - device_idx = index.device_index_by_key.get(device_key) + device_idx = index.device_index.get(device_key) if device_idx is None: - device_idx = len(index.device_order) - index.device_index_by_key[device_key] = device_idx - index.device_order.append(device_key) - index.device_refs.append([]) - index.device_refs[device_idx].append((reader, device_id)) - - for field_idx in range(len(table_entry.field_columns)): - series_ref = (device_idx, field_idx) - if series_ref not in index.series_ref_map: - index.series_refs_ordered.append(series_ref) - index.series_ref_map[series_ref] = [] - index.series_ref_map[series_ref].append((reader, device_id, field_idx)) - - -def _build_device_entry(refs: List[DeviceRef]) -> dict: - """Compute per-device time bounds from cheap metadata only. - - We intentionally do not validate duplicates at the device level because - table-model fields do not necessarily share one complete timestamp axis. - Duplicate detection stays on the logical-series paths that materialize or - merge one field's timestamps. - """ - infos = [reader.get_device_info(device_id) for reader, device_id in refs] - min_time = min(info["min_time"] for info in infos) - max_time = max(info["max_time"] for info in infos) - - return { - "min_time": min_time, - "max_time": max_time, - } + device_idx = len(index.devices) + index.device_index[device_key] = device_idx + index.devices.append(device_key) + index.device_time_bounds.append( + (device_entry.min_time, device_entry.max_time) + ) + else: + cur_min, cur_max = index.device_time_bounds[device_idx] + new_min = ( + device_entry.min_time + if cur_min is None + else min(cur_min, device_entry.min_time) + ) + new_max = ( + device_entry.max_time + if cur_max is None + else max(cur_max, device_entry.max_time) + ) + index.device_time_bounds[device_idx] = (new_min, new_max) + + # Register every (device, field) pair the reader physically holds. The + # logical series is keyed by the GLOBAL field index (resolved by measurement + # name against the merged table), while the per-shard tuple keeps the + # reader-local field index for the read path. This lets tree shards with + # different field subsets/orders merge without mis-mapping measurements + # (for table model the global and local indices always coincide). + for device_id, field_idx in catalog.series_stats_by_ref: + device_entry = catalog.device_entries[device_id] + reader_table_entry = catalog.table_entries[device_entry.table_id] + field_name = reader_table_entry.field_columns[field_idx] + device_key = (reader_table_entry.table_name, tuple(device_entry.tag_values)) + device_idx = index.device_index[device_key] + global_field_idx = index.table_entries[ + reader_table_entry.table_name + ].get_field_index(field_name) + series_ref = (device_idx, global_field_idx) + if series_ref not in index.series_shards: + index.series.append(series_ref) + index.series_shards[series_ref] = [] + index.series_shards[series_ref].append((reader, device_id, field_idx)) def _build_runtime_series_stats(refs: List[SeriesRef]) -> dict: @@ -470,10 +531,10 @@ class _LocIndexer: if isinstance(item, (int, np.integer)): idx = int(item) if idx < 0: - idx += len(self._df._index.series_refs_ordered) - if idx < 0 or idx >= len(self._df._index.series_refs_ordered): + idx += len(self._df._index.series) + if idx < 0 or idx >= len(self._df._index.series): raise IndexError(f"Series index {item} out of range") - series_ref = self._df._index.series_refs_ordered[idx] + series_ref = self._df._index.series[idx] elif isinstance(item, str): series_ref = self._df._resolve_series_name(item) else: @@ -497,17 +558,17 @@ class _LocIndexer: groups = defaultdict(list) for col_idx, series_ref in enumerate(series_refs): device_idx, field_idx = series_ref - device_info = self._df._cache.devices[device_idx] + min_time_dev, max_time_dev = self._df._index.device_time_bounds[device_idx] if ( - device_info["max_time"] is None - or device_info["max_time"] < start_time - or device_info["min_time"] > end_time + max_time_dev is None + or max_time_dev < start_time + or (min_time_dev is not None and min_time_dev > end_time) ): continue _, table_entry, _ = self._df._get_series_components(series_ref) field_name = table_entry.field_columns[field_idx] - for reader, device_id, reader_field_idx in self._df._index.series_ref_map[ + for reader, device_id, reader_field_idx in self._df._index.series_shards[ series_ref ]: groups[(id(reader), device_id)].append( @@ -530,10 +591,15 @@ class _LocIndexer: ts_arr, field_vals = reader.read_device_fields_by_time_range( device_id, field_indices, start_time, end_time ) + if len(ts_arr) == 0: + continue + appended_series = set() for _, _, field_name, series_name, _, _ in entries: - if len(ts_arr) > 0: - series_time_parts[series_name].append(ts_arr) - series_value_parts[series_name].append(field_vals[field_name]) + if series_name in appended_series: + continue + appended_series.add(series_name) + series_time_parts[series_name].append(ts_arr) + series_value_parts[series_name].append(field_vals[field_name]) series_data = {} for name in series_names: @@ -558,8 +624,7 @@ class TsFileDataFrame: self._paths = _expand_paths(paths) self._show_progress = show_progress self._readers: Dict[str, object] = {} - self._index = _LogicalIndex() - self._cache = _DerivedCache() + self._index = _DataFrameCatalog() self._is_view = False self._root = None self._closed = False @@ -576,17 +641,19 @@ class TsFileDataFrame: obj._paths = parent._paths obj._show_progress = parent._show_progress obj._readers = parent._readers - obj._index = _LogicalIndex( + # Reuse the parent's full mapping but restrict the membership scope to + # the requested subset. + subset_refs = list(series_refs) + parent_shards = parent._index.series_shards + subset_shards = {ref: parent_shards[ref] for ref in subset_refs} + obj._index = _DataFrameCatalog( + model=parent._index.model, table_entries=parent._index.table_entries, - device_order=parent._index.device_order, - device_index_by_key=parent._index.device_index_by_key, - device_refs=parent._index.device_refs, - series_refs_ordered=list(series_refs), - series_ref_map=parent._index.series_ref_map, - series_ref_set=set(series_refs), - ) - obj._cache = _DerivedCache( - devices=parent._cache.devices, field_stats=parent._cache.field_stats + devices=parent._index.devices, + device_index=parent._index.device_index, + device_time_bounds=parent._index.device_time_bounds, + series=subset_refs, + series_shards=subset_shards, ) obj._closed = False return obj @@ -607,16 +674,7 @@ class TsFileDataFrame: else: self._load_metadata_serial(TsFileSeriesReader) - self._cache.devices = [ - _build_device_entry(refs) for refs in self._index.device_refs - ] - for series_ref in self._index.series_refs_ordered: - self._cache.field_stats[series_ref] = _build_field_stats( - self._index.series_ref_map[series_ref] - ) - - self._index.series_ref_set = set(self._index.series_refs_ordered) - if not self._index.series_refs_ordered: + if not self._index.series: raise ValueError("No valid time series found in the provided TsFile files") def _show_loading_progress(self, done: int, total: int, total_series: int = None): @@ -687,7 +745,7 @@ class TsFileDataFrame: self, series_ref: SeriesRefKey ) -> Tuple[DeviceKey, TableEntry, int]: device_idx, field_idx = series_ref - device_key = self._index.device_order[device_idx] + device_key = self._index.devices[device_idx] return device_key, self._index.table_entries[device_key[0]], field_idx def _build_series_name(self, series_ref: SeriesRefKey) -> SeriesPath: @@ -728,37 +786,50 @@ class TsFileDataFrame: raise KeyError(_series_lookup_hint(series_name)) from exc device_key = (table_name, _normalize_tag_values(tag_parts)) - device_idx = self._index.device_index_by_key.get(device_key) + device_idx = self._index.device_index.get(device_key) if device_idx is None: raise KeyError(_series_lookup_hint(series_name)) series_ref = (device_idx, field_idx) - if series_ref not in self._index.series_ref_set: + if series_ref not in self._index.series_shards: raise KeyError(_series_lookup_hint(series_name)) return series_ref def _build_series_info(self, series_ref: SeriesRefKey) -> dict: device_idx, field_idx = series_ref device_key, table_entry, _ = self._get_series_components(series_ref) - field_stats = self._cache.field_stats[series_ref] + # Aggregate per-shard timeline stats lazily on demand for this series. + field_stats = _build_field_stats(self._index.series_shards[series_ref]) + # Pad short tag tuples (tree-model devices whose path is shorter than + # the synthetic table's max depth) with None so positional access by + # `_col_i` index always lands on a defined cell. + tag_values_ordered = list(device_key[1]) + if len(tag_values_ordered) < len(table_entry.tag_columns): + tag_values_ordered.extend( + [None] * (len(table_entry.tag_columns) - len(tag_values_ordered)) + ) return { "table_name": table_entry.table_name, "field": table_entry.field_columns[field_idx], "tag_columns": table_entry.tag_columns, - "tag_values": dict(zip(table_entry.tag_columns, device_key[1])), + "tag_values": dict(zip(table_entry.tag_columns, tag_values_ordered)), + "tag_values_ordered": tag_values_ordered, "min_time": field_stats["min_time"], "max_time": field_stats["max_time"], "count": field_stats["count"], } def __len__(self) -> int: - return len(self._index.series_refs_ordered) + return len(self._index.series) + + @property + def model(self) -> str: + return self._index.model def list_timeseries(self, path_prefix: str = "") -> List[SeriesPath]: if not path_prefix: return [ - self._build_series_name(series_ref) - for series_ref in self._index.series_refs_ordered + self._build_series_name(series_ref) for series_ref in self._index.series ] try: @@ -767,7 +838,7 @@ class TsFileDataFrame: return [] matched = [] - for series_ref in self._index.series_refs_ordered: + for series_ref in self._index.series: device_key, table_entry, field_idx = self._get_series_components(series_ref) components = build_logical_series_components( table_entry.table_name, @@ -779,19 +850,79 @@ class TsFileDataFrame: matched.append(self._build_series_name(series_ref)) return matched + def list_timeseries_metadata(self, path_prefix: str = ""): + """Return a pandas DataFrame of per-series metadata. + + The returned frame is indexed by the logical series name and includes + per-series ``field``, time-bound (start/end) statistics, observation + ``count``, and the per-device tag values (named ``_col_1``, ``_col_2``, + ... in tree mode, or by their declared tag-column names in table + mode). Time bounds are exposed as ``pandas.Timestamp`` for ergonomic + comparison; ``count`` is an integer. + + ``path_prefix`` filters by the same logical-path prefix semantics as + ``list_timeseries`` (no prefix returns the full catalog). + """ + import pandas as pd + + # Reuse list_timeseries to apply prefix filtering, then map names back + # to the underlying series_ref (this respects view subsetting too). + names = self.list_timeseries(path_prefix) + + rows = [] + for series_name in names: + series_ref = self._resolve_series_name(series_name) + info = self._build_series_info(series_ref) + row = { + "field": info["field"], + "start_time": pd.to_datetime(info["min_time"], unit="ms"), + "end_time": pd.to_datetime(info["max_time"], unit="ms"), + "count": int(info["count"]), + } + if self._index.model != MODEL_TREE: + row["table"] = info["table_name"] + tag_columns = info["tag_columns"] + tag_values_ordered = info["tag_values_ordered"] + for column, value in zip(tag_columns, tag_values_ordered): + row[column] = value + rows.append((series_name, row)) + + if not rows: + columns = ["field", "start_time", "end_time", "count"] + if self._index.model != MODEL_TREE: + columns.insert(0, "table") + columns.extend(self._collect_tag_columns()) + return pd.DataFrame(columns=columns) + + index = [name for name, _ in rows] + data = [row for _, row in rows] + df = pd.DataFrame(data, index=index) + + # Stable, predictable column order: leading bookkeeping, then tags. + leading = ["field", "start_time", "end_time", "count"] + if self._index.model != MODEL_TREE: + leading.insert(0, "table") + tag_order = list(self._collect_tag_columns()) + ordered_columns = leading + [c for c in tag_order if c in df.columns] + # Preserve any extra columns at the end (defensive against schema drift). + for extra in df.columns: + if extra not in ordered_columns: + ordered_columns.append(extra) + return df.reindex(columns=ordered_columns) + def _get_timeseries(self, series_ref: SeriesRefKey) -> Timeseries: self._assert_open() series_name = self._build_series_name(series_ref) return Timeseries( series_name, - self._index.series_ref_map[series_ref], - _build_runtime_series_stats(self._index.series_ref_map[series_ref]), + self._index.series_shards[series_ref], + _build_runtime_series_stats(self._index.series_shards[series_ref]), self._assert_open, lambda: _merge_field_timestamps( - series_name, self._index.series_ref_map[series_ref] + series_name, self._index.series_shards[series_ref] ), lambda offset, limit: _read_field_by_position( - series_name, self._index.series_ref_map[series_ref], offset, limit + series_name, self._index.series_shards[series_ref], offset, limit ), ) @@ -800,9 +931,7 @@ class TsFileDataFrame: import pandas as pd if isinstance(key, pd.Series) and key.dtype == bool: - selected = [ - self._index.series_refs_ordered[idx] for idx in key.index[key] - ] + selected = [self._index.series[idx] for idx in key.index[key]] return TsFileDataFrame._from_subset(self, selected) except ImportError: pass @@ -810,12 +939,12 @@ class TsFileDataFrame: if isinstance(key, (int, np.integer)): idx = int(key) if idx < 0: - idx += len(self._index.series_refs_ordered) - if idx < 0 or idx >= len(self._index.series_refs_ordered): + idx += len(self._index.series) + if idx < 0 or idx >= len(self._index.series): raise IndexError( - f"Index {idx} out of range [0, {len(self._index.series_refs_ordered)})" + f"Index {idx} out of range [0, {len(self._index.series)})" ) - return self._get_timeseries(self._index.series_refs_ordered[idx]) + return self._get_timeseries(self._index.series[idx]) if isinstance(key, str): try: @@ -823,7 +952,9 @@ class TsFileDataFrame: except KeyError: pass - valid_columns = {"table", "field", "start_time", "end_time", "count"} + valid_columns = {"field", "start_time", "end_time", "count"} + if self._index.model != MODEL_TREE: + valid_columns.add("table") valid_columns.update(self._collect_tag_columns()) if key not in valid_columns: raise KeyError(_series_lookup_hint(key)) @@ -831,7 +962,7 @@ class TsFileDataFrame: import pandas as pd values = [] - for series_ref in self._index.series_refs_ordered: + for series_ref in self._index.series: info = self._build_series_info(series_ref) if key == "table": values.append(info["table_name"]) @@ -844,15 +975,15 @@ class TsFileDataFrame: elif key == "count": values.append(info["count"]) else: - values.append(info["tag_values"].get(key, "")) + values.append(info["tag_values"].get(key)) return pd.Series(values, name=key) if isinstance(key, slice): return TsFileDataFrame._from_subset( self, [ - self._index.series_refs_ordered[idx] - for idx in range(*key.indices(len(self._index.series_refs_ordered))) + self._index.series[idx] + for idx in range(*key.indices(len(self._index.series))) ], ) @@ -865,12 +996,12 @@ class TsFileDataFrame: ) idx = int(item) if idx < 0: - idx += len(self._index.series_refs_ordered) - if idx < 0 or idx >= len(self._index.series_refs_ordered): + idx += len(self._index.series) + if idx < 0 or idx >= len(self._index.series): raise IndexError( - f"Index {item} out of range [0, {len(self._index.series_refs_ordered)})" + f"Index {item} out of range [0, {len(self._index.series)})" ) - selected.append(self._index.series_refs_ordered[idx]) + selected.append(self._index.series[idx]) return TsFileDataFrame._from_subset(self, selected) raise TypeError(f"Unsupported key type: {type(key)}") @@ -881,7 +1012,7 @@ class TsFileDataFrame: def _collect_tag_columns(self) -> List[str]: seen = {} - for table_name, _ in self._index.device_order: + for table_name, _ in self._index.devices: for column in self._index.table_entries[table_name].tag_columns: seen.setdefault(column, True) return list(seen.keys()) @@ -900,25 +1031,27 @@ class TsFileDataFrame: def _format_table(self, indices=None, max_rows: int = 20) -> str: if indices is None: - indices = list(range(len(self._index.series_refs_ordered))) + indices = list(range(len(self._index.series))) else: indices = list(indices) preview_indices, truncated, split_index = self._preview_indices( indices, max_rows ) + is_tree = self._index.model == MODEL_TREE rows = [] for idx in preview_indices: - series_ref = self._index.series_refs_ordered[idx] + series_ref = self._index.series[idx] info = self._build_series_info(series_ref) row = { "index": idx, - "table": info["table_name"], "field": info["field"], "start_time": info["min_time"], "end_time": info["max_time"], "count": info["count"], } + if not is_tree: + row["table"] = info["table_name"] row.update(info["tag_values"]) rows.append(row) @@ -928,13 +1061,21 @@ class TsFileDataFrame: total_count=len(indices), truncated=truncated, split_index=split_index, + is_table_model=not is_tree, ) def _repr_header(self) -> str: - total = len(self._index.series_refs_ordered) + total = len(self._index.series) + model_marker = self._index.model if self._is_view: - return f"TsFileDataFrame({total} time series, subset of {len(self._root._index.series_refs_ordered)})\n" - return f"TsFileDataFrame({total} time series, {len(self._paths)} files)\n" + return ( + f"TsFileDataFrame({model_marker} model, {total} time series, " + f"subset of {len(self._root._index.series)})\n" + ) + return ( + f"TsFileDataFrame({model_marker} model, {total} time series, " + f"{len(self._paths)} files)\n" + ) def __repr__(self): return self._repr_header() + self._format_table() diff --git a/python/tsfile/dataset/formatting.py b/python/tsfile/dataset/formatting.py index 8c387c102..f39acfb0a 100644 --- a/python/tsfile/dataset/formatting.py +++ b/python/tsfile/dataset/formatting.py @@ -106,8 +106,14 @@ def format_dataframe_table( total_count: int, truncated: bool = False, split_index: Optional[int] = None, + is_table_model: bool = True, ) -> str: - """Render the metadata table used by TsFileDataFrame.__repr__.""" + """Render the metadata table used by TsFileDataFrame.__repr__. + + When ``is_table_model`` is False (tree-model layout) the leading + ``table`` column is omitted; ``tag_columns`` is then expected to carry the + tree-model headers (e.g. ``_col_1``, ``_col_2``, ...). + """ if not rows: return "Empty TsFileDataFrame" @@ -115,23 +121,31 @@ def format_dataframe_table( for row in rows: rendered = { "index": row["index"], - "table": row["table"], "field": row["field"], "start_time": format_timestamp(row["start_time"]), "end_time": format_timestamp(row["end_time"]), "count": row["count"], } + if is_table_model: + rendered["table"] = row["table"] for tag_col in tag_columns: - rendered[tag_col] = row.get(tag_col, "") + value = row.get(tag_col) + rendered[tag_col] = "None" if value is None else value rendered_rows.append(rendered) - headers = ["", "table"] + tag_columns + ["field", "start_time", "end_time", "count"] + headers = [""] + if is_table_model: + headers.append("table") + headers.extend(tag_columns) + headers.extend(["field", "start_time", "end_time", "count"]) + widths = {header: len(header) for header in headers} widths[""] = max(len(str(row["index"])) for row in rendered_rows) for row in rendered_rows: widths[""] = max(widths[""], len(str(row["index"]))) - widths["table"] = max(widths["table"], len(row["table"])) + if is_table_model: + widths["table"] = max(widths["table"], len(row["table"])) widths["field"] = max(widths["field"], len(row["field"])) widths["start_time"] = max(widths["start_time"], len(row["start_time"])) widths["end_time"] = max(widths["end_time"], len(row["end_time"])) @@ -144,10 +158,9 @@ def format_dataframe_table( for row_idx, row in enumerate(rendered_rows): if truncated and row_idx == split: lines.append("...") - parts = [ - str(row["index"]).rjust(widths[""]), - row["table"].rjust(widths["table"]), - ] + parts = [str(row["index"]).rjust(widths[""])] + if is_table_model: + parts.append(row["table"].rjust(widths["table"])) for tag_col in tag_columns: parts.append(str(row[tag_col]).rjust(widths[tag_col])) parts.extend( diff --git a/python/tsfile/dataset/metadata.py b/python/tsfile/dataset/metadata.py index c6d7e5887..5eb367273 100644 --- a/python/tsfile/dataset/metadata.py +++ b/python/tsfile/dataset/metadata.py @@ -20,7 +20,7 @@ from dataclasses import dataclass, field import sys -from typing import Any, Dict, Iterable, Iterator, List, Tuple +from typing import Any, Dict, Iterable, Iterator, List, NamedTuple, Tuple from ..constants import TSDataType @@ -28,6 +28,20 @@ _PATH_SEPARATOR = "." _PATH_ESCAPE = "\\" _DATACLASS_SLOTS = {"slots": True} if sys.version_info >= (3, 10) else {} +MODEL_TABLE = "table" +MODEL_TREE = "tree" + + +class SeriesStats(NamedTuple): + """Statistics for a single time series.""" + + length: int + min_time: int + max_time: int + timeline_length: int + timeline_min_time: int + timeline_max_time: int + @dataclass(**_DATACLASS_SLOTS) class TableEntry: @@ -73,7 +87,7 @@ class MetadataCatalog: device_entries: List[DeviceEntry] = field(default_factory=list) table_id_by_name: Dict[str, int] = field(default_factory=dict) device_id_by_key: Dict[Tuple[int, tuple], int] = field(default_factory=dict) - series_stats_by_ref: Dict[Tuple[int, int], Dict[str, int]] = field( + series_stats_by_ref: Dict[Tuple[int, int], SeriesStats] = field( default_factory=dict ) @@ -122,10 +136,11 @@ class MetadataCatalog: @property def series_count(self) -> int: - return sum( - len(self.table_entries[device.table_id].field_columns) - for device in self.device_entries - ) + # Count only physically-present series -- the (device, field) pairs that + # actually carry data -- so this matches len(tsdf) / series_paths and + # the reader's series_count. A schema cross-product (devices x declared + # fields) would overcount sparse schemas where a device skips fields. + return len(self.series_stats_by_ref) # Path marker for a null tag value: a single backslash followed by N. A real diff --git a/python/tsfile/dataset/reader.py b/python/tsfile/dataset/reader.py index 2cf0f4a55..9b77190e1 100644 --- a/python/tsfile/dataset/reader.py +++ b/python/tsfile/dataset/reader.py @@ -29,8 +29,10 @@ from ..tag_filter import tag_eq, tag_is_null from ..tsfile_reader import TsFileReaderPy from .metadata import ( MetadataCatalog, + MODEL_TABLE, + MODEL_TREE, + SeriesStats, build_series_path, - iter_series_refs, resolve_series_path, ) @@ -78,6 +80,13 @@ def _device_exact_tag_values(table_entry, device_entry) -> Dict[str, object]: } +def _expand_tree_device_segments(segments: Tuple[object, ...]) -> Tuple[object, ...]: + """Expand the native tree DeviceID prefix into logical path segments.""" + if not segments or segments[0] is None: + return tuple(segments) + return tuple(str(segments[0]).split(".")) + tuple(segments[1:]) + + class TsFileSeriesReader: """Wrap ``TsFileReaderPy`` with numeric dataset discovery and batch reads.""" @@ -93,6 +102,10 @@ class TsFileSeriesReader: except Exception as e: raise ValueError(f"Failed to open TsFile: {e}") from e + # Probe the file model: an empty table-schema map signals tree model + self._table_schemas = self._reader.get_all_table_schemas() + self._model_kind: str = MODEL_TREE if not self._table_schemas else MODEL_TABLE + self._catalog = MetadataCatalog() self._cache_metadata() @@ -103,6 +116,10 @@ class TsFileSeriesReader: def catalog(self) -> MetadataCatalog: return self._catalog + @property + def model_kind(self) -> str: + return self._model_kind + @property def series_paths(self) -> List[str]: return list(self.iter_series_paths()) @@ -112,11 +129,11 @@ class TsFileSeriesReader: return self._catalog.series_count def iter_series_paths(self) -> Iterator[str]: - for device_id, field_idx in iter_series_refs(self._catalog): + for device_id, field_idx in self._catalog.series_stats_by_ref: yield build_series_path(self._catalog, device_id, field_idx) def iter_series_refs(self) -> Iterator[Tuple[str, int, int]]: - for device_id, field_idx in iter_series_refs(self._catalog): + for device_id, field_idx in self._catalog.series_stats_by_ref: yield build_series_path( self._catalog, device_id, field_idx ), device_id, field_idx @@ -131,8 +148,10 @@ class TsFileSeriesReader: def _cache_metadata(self): """Wrap metadata discovery so reader construction surfaces one stable error shape.""" try: - self._cache_metadata_table_model() - # todo: we should support tree model + if self._model_kind == MODEL_TABLE: + self._cache_metadata_table_model() + else: + self._cache_metadata_tree_model() except Exception as e: raise ValueError( f"Failed to read TsFile metadata. Please ensure the TsFile is valid and readable. Error: {e}" @@ -140,7 +159,7 @@ class TsFileSeriesReader: def _cache_metadata_table_model(self): """Build the in-memory catalog from table schemas and native metadata.""" - table_schemas = self._reader.get_all_table_schemas() + table_schemas = self._table_schemas if not table_schemas: raise ValueError("No tables found in TsFile") @@ -204,18 +223,13 @@ class TsFileSeriesReader: for field_idx, field_name in enumerate(table_entry.field_columns): field_stats = stats_by_field.get(field_name) if field_stats is None: - self._catalog.series_stats_by_ref[(device_id, field_idx)] = { - "length": 0, - "min_time": None, - "max_time": None, - "timeline_length": 0, - "timeline_min_time": None, - "timeline_max_time": None, - } - else: - self._catalog.series_stats_by_ref[(device_id, field_idx)] = ( - field_stats - ) + # Schema declares this field but the device never + # wrote it (or wrote it entirely as NaN). Skip -- + # the dataset surface only carries real series. + continue + self._catalog.series_stats_by_ref[(device_id, field_idx)] = ( + field_stats + ) if self.show_progress: sys.stderr.write( @@ -230,6 +244,105 @@ class TsFileSeriesReader: ) sys.stderr.flush() + def _cache_metadata_tree_model(self): + """Build the in-memory catalog by synthesizing one virtual table. + + Tree TsFiles have no schema, so we materialize a single + ``TableEntry``: table name = the shared root segment, tag columns + = ``_col_1..._col_{N_max}`` (one per remaining path segment), + fields = union of measurements across devices. Per-device + ownership is preserved by registering only the + ``(device_id, field_idx)`` pairs actually written on disk in + ``series_stats_by_ref``. + """ + metadata_groups = self._reader.get_timeseries_metadata(None) + if not metadata_groups: + raise ValueError("No devices found in tree-model TsFile") + + # 1) Walk every device once to collect: root-segment, max depth, and + # the union of measurements that pass the numeric filter. + root_name = None + max_depth = 0 # segments after the root (i.e. virtual tag depth) + device_specs = [] # list of (tail_segments, group, stats_by_field) + union_fields = [] # ordered union of measurement names + seen_field_names = set() + + for group in metadata_groups.values(): + full_segments = _expand_tree_device_segments(tuple(group.segments)) + if not full_segments: + continue + current_root = full_segments[0] + if root_name is None: + root_name = current_root + elif current_root != root_name: + raise ValueError( + f"Tree-model TsFile contains multiple root segments: " + f"{root_name!r} vs {current_root!r}. A single load set " + f"must share one tree root." + ) + + tail = full_segments[1:] + depth = len(tail) + if depth > max_depth: + max_depth = depth + + stats_by_field = self._metadata_field_stats(group) + if not stats_by_field: + continue + for measurement in stats_by_field.keys(): + if measurement not in seen_field_names: + seen_field_names.add(measurement) + union_fields.append(measurement) + device_specs.append((tail, group, stats_by_field)) + + if root_name is None: + raise ValueError("No devices found in tree-model TsFile") + if not union_fields: + raise ValueError("No numeric measurements found in tree-model TsFile") + + # 2) Materialize the synthetic table entry. Tag columns are 1-based + # so the rendered headers match the requirement ("_col_1"). + tag_columns = tuple(f"_col_{i + 1}" for i in range(max_depth)) + tag_types = (TSDataType.STRING,) * max_depth + + self._catalog = MetadataCatalog() + table_id = self._catalog.add_table( + root_name, tag_columns, tag_types, union_fields + ) + table_entry = self._catalog.table_entries[table_id] + + # 3) Stable device order: keep input iteration order so the dataset + # layer's row index stays deterministic across reloads. + total = len(device_specs) + if self.show_progress: + sys.stderr.write(f"\rReading TsFile metadata: 0/{total} devices") + sys.stderr.flush() + + for idx, (tail, group, stats_by_field) in enumerate(device_specs, start=1): + device_stats = self._metadata_device_stats(group) + if device_stats is None: + continue + # Pad shorter devices with None to length max_depth; the catalog + # normalizer strips trailing Nones so this never enters the + # sparse-tag bookkeeping. + padded = tuple(list(tail) + [None] * (max_depth - len(tail))) + device_id = self._add_device( + table_id, padded, device_stats["min_time"], device_stats["max_time"] + ) + for measurement, field_stats in stats_by_field.items(): + field_idx = table_entry.get_field_index(measurement) + self._catalog.series_stats_by_ref[(device_id, field_idx)] = field_stats + if self.show_progress and (idx % 64 == 0 or idx == total): + sys.stderr.write(f"\rReading TsFile metadata: {idx}/{total} devices") + sys.stderr.flush() + + if self.show_progress: + sys.stderr.write( + f"\rReading TsFile metadata (tree): {total} device(s), " + f"{self.series_count} series ... done\n" + ) + sys.stderr.flush() + @staticmethod def _metadata_device_stats(group) -> dict: """Derive cheap device-level metadata hints from native field statistics. @@ -269,28 +382,44 @@ class TsFileSeriesReader: return tuple(values) @staticmethod - def _metadata_field_stats(group) -> Dict[str, dict]: - stats = {} + def _metadata_field_stats(group) -> Dict[str, SeriesStats]: + """Collect per-measurement stats for numeric cells that have real values. + + A measurement appears iff it is numeric AND its native ``statistic`` + block is populated with a positive ``row_count``. Non-numeric + measurements (STRING/TEXT) are dropped because the dataset surface reads + values as ``float64`` -- the same reason table mode filters non-numeric + fields out of its schema. Columns that the device never wrote (Tablet + skip / all-NaN pandas column) carry no real values and are intentionally + absent -- the dataset layer surfaces only series that physically exist. + """ + stats: Dict[str, SeriesStats] = {} for timeseries in group.timeseries: + # Drop non-numeric measurements (see docstring): tree mode relies on + # this to avoid surfacing a string/text series that crashes on read. + if timeseries.data_type not in _NUMERIC_FIELD_TYPES: + continue statistic = timeseries.statistic - timeline_statistic = timeseries.timeline_statistic - if ( - not timeline_statistic.has_statistic - or timeline_statistic.row_count <= 0 - ): + # Gate on the value statistic's non-null row_count, not the + # timeline. A skipped/all-NaN field still carries a value-stat + # block (has_statistic=True) and shares the device's timeline + # rows, so a timeline gate would surface a phantom series whose + # value row_count is 0. See + # test_dataset_omits_table_model_phantom_series_for_skipped_cells. + if not statistic.has_statistic or statistic.row_count <= 0: continue - stats[timeseries.measurement_name] = { - "length": int(statistic.row_count) if statistic.has_statistic else 0, - "min_time": ( - int(statistic.start_time) if statistic.has_statistic else None - ), - "max_time": ( - int(statistic.end_time) if statistic.has_statistic else None - ), - "timeline_length": int(timeline_statistic.row_count), - "timeline_min_time": int(timeline_statistic.start_time), - "timeline_max_time": int(timeline_statistic.end_time), - } + # timeline_statistic is always a TimeseriesStatistic (never None; + # the native binding fills row_count/start/end even when + # has_statistic is False), so the reads below cannot raise. + timeline_statistic = timeseries.timeline_statistic + stats[timeseries.measurement_name] = SeriesStats( + length=int(statistic.row_count), + min_time=int(statistic.start_time), + max_time=int(statistic.end_time), + timeline_length=int(timeline_statistic.row_count), + timeline_min_time=int(timeline_statistic.start_time), + timeline_max_time=int(timeline_statistic.end_time), + ) return stats def _add_device( @@ -330,12 +459,12 @@ class TsFileSeriesReader: ) field_stats = self._catalog.series_stats_by_ref[(device_id, field_idx)] return { - "length": field_stats["length"], - "min_time": field_stats["min_time"], - "max_time": field_stats["max_time"], - "timeline_length": field_stats["timeline_length"], - "timeline_min_time": field_stats["timeline_min_time"], - "timeline_max_time": field_stats["timeline_max_time"], + "length": field_stats.length, + "min_time": field_stats.min_time, + "max_time": field_stats.max_time, + "timeline_length": field_stats.timeline_length, + "timeline_min_time": field_stats.timeline_min_time, + "timeline_max_time": field_stats.timeline_max_time, "table_name": table_entry.table_name, "column_name": field_name, "device_id": device_id, @@ -375,6 +504,11 @@ class TsFileSeriesReader: table_entry, device_entry, field_name = self._resolve_series_ref( device_id, field_idx ) + + if self._model_kind == MODEL_TREE: + device_path = self._build_tree_device_path(table_entry, device_entry) + return self._read_series_by_row_tree(device_path, field_name, offset, limit) + tag_values = _device_exact_tag_values(table_entry, device_entry) tag_filter = _build_exact_tag_filter(tag_values) if tag_values else None @@ -423,6 +557,75 @@ class TsFileSeriesReader: return timestamp_parts[0], value_parts[0] return np.concatenate(timestamp_parts), np.concatenate(value_parts) + def _read_series_by_row_tree( + self, device_path: str, field_name: str, offset: int, limit: int + ) -> Tuple[np.ndarray, np.ndarray]: + """Tree-model row read: scan on-tree result, filter device, apply offset/limit.""" + target_path_segments = device_path.split(".") + # +1 because cwrapper prepends the root as an extra col_i cell. + expected_path_len = ( + max(len(t.tag_columns) for t in self._catalog.table_entries) + 1 + ) + timestamps = [] + values = [] + skipped = 0 + # PERF: O(total_rows). query_table_on_tree scans every device's rows + # for this field and we filter down to one device client-side, so an + # aligned read over N devices is O(N * total_rows). Per-device tree + # pushdown (query_tree_by_row) isn't reliable in the cwrapper yet + # (stale col_i path columns leak across queries on a reused reader); + # see PR #816. Hot path for profilers. + with self._reader.query_table_on_tree([field_name]) as result_set: + md = result_set.get_metadata() + num_cols = md.get_column_num() + col_names = [md.get_column_name(i + 1) for i in range(num_cols)] + try: + field_idx = col_names.index(field_name) + 1 + except ValueError: + return np.array([], dtype=np.int64), np.array([], dtype=np.float64) + all_col_indices = [ + idx + 1 for idx, name in enumerate(col_names) if name.startswith("col_") + ] + # Only the trailing expected_path_len col_i cells are genuine; the + # leading duplicates are stale from prior queries on this reader. + col_indices = all_col_indices[-expected_path_len:] + # Fail fast if the cwrapper col_ leak pattern changes: the trailing + # slice assumes at least expected_path_len col_ cells. A count + # mismatch means the heuristic is stale and could silently pick the + # wrong path columns (returning wrong-device data). Guards the count + # only -- it can't catch a leak where the genuine columns are no + # longer the trailing ones. + assert len(col_indices) == expected_path_len, ( + f"tree path col_ columns: expected {expected_path_len}, got " + f"{len(col_indices)} of {len(all_col_indices)}; cwrapper col_ " + f"leak pattern may have changed" + ) + while result_set.next(): + row_path_segments = [ + result_set.get_value_by_index(ci) for ci in col_indices + ] + # Trim trailing Nones for the (possibly-shorter) device path. + while row_path_segments and row_path_segments[-1] is None: + row_path_segments.pop() + if row_path_segments != target_path_segments: + continue + if skipped < offset: + skipped += 1 + continue + if len(timestamps) >= limit: + break + ts = result_set.get_value_by_index(1) + raw = result_set.get_value_by_index(field_idx) + timestamps.append(int(ts)) + values.append(np.nan if raw is None else float(raw)) + + if not timestamps: + return np.array([], dtype=np.int64), np.array([], dtype=np.float64) + return ( + np.asarray(timestamps, dtype=np.int64), + np.asarray(values, dtype=np.float64), + ) + def read_device_fields_by_time_range( self, device_id: int, field_indices: List[int], start_time: int, end_time: int ) -> Tuple[np.ndarray, Dict[str, np.ndarray]]: @@ -432,7 +635,12 @@ class TsFileSeriesReader: requested_field_columns = [ table_entry.field_columns[field_idx] for field_idx in field_indices ] - timestamps, field_values = self._read_arrow( + if self._model_kind == MODEL_TREE: + device_path = self._build_tree_device_path(table_entry, device_entry) + return self._read_arrow_tree( + device_path, requested_field_columns, start_time, end_time + ) + return self._read_arrow_table( table_entry.table_name, requested_field_columns, table_entry.tag_columns, @@ -440,9 +648,36 @@ class TsFileSeriesReader: start_time, end_time, ) - return timestamps, field_values - def _read_arrow( + @staticmethod + def _build_tree_device_path(table_entry, device_entry) -> str: + """Reassemble the cwrapper-facing tree device path from catalog state. + + The native ``query_timeseries`` / ``query_tree_by_row`` APIs split the + device path on ``.`` internally, so segments themselves must not + contain ``.``. Tree-model writers enforce this convention; we surface + an explicit error if a future writer ever violates it. + """ + components = [str(table_entry.table_name)] + for value in device_entry.tag_values: + if value is None: + # Should not happen: trailing-None devices are normalized to a + # shorter tag tuple. An interior None signals a sparse-tag + # device, which is not part of the tree-model contract. + raise ValueError( + f"Tree device path cannot include a null segment: " + f"{device_entry.tag_values!r}" + ) + text = str(value) + if "." in text: + raise NotImplementedError( + f"Tree device segment with '.' is not supported by the " + f"underlying cwrapper path API: {text!r}" + ) + components.append(text) + return ".".join(components) + + def _read_arrow_table( self, table_name: str, field_columns: List[str], @@ -511,3 +746,94 @@ class TsFileSeriesReader: } return timestamps, field_values + + def _read_arrow_tree( + self, + device_path: str, + field_columns: List[str], + start_time: int, + end_time: int, + ) -> Tuple[np.ndarray, Dict[str, np.ndarray]]: + """Tree-model time-range read for one device (multi-field).""" + field_columns = list(field_columns) + if not field_columns: + return ( + np.array([], dtype=np.int64), + {}, + ) + + target_path_segments = device_path.split(".") + expected_path_len = ( + max(len(t.tag_columns) for t in self._catalog.table_entries) + 1 + ) + timestamps = [] + value_buckets = {col: [] for col in field_columns} + + # PERF: O(total_rows) full tree scan filtered to one device + # client-side; aligned reads over N devices are O(N * total_rows). + # See _read_series_by_row_tree / PR #816 for the cwrapper limitation + # that blocks per-device pushdown (query_timeseries). Hot path. + with self._reader.query_table_on_tree( + field_columns, start_time, end_time + ) as result_set: + md = result_set.get_metadata() + num_cols = md.get_column_num() + col_names = [md.get_column_name(i + 1) for i in range(num_cols)] + value_indices = {} + for col in field_columns: + try: + value_indices[col] = col_names.index(col) + 1 + except ValueError: + # Column missing (no device in file owns it). Yield empty. + return ( + np.array([], dtype=np.int64), + { + col2: np.array([], dtype=np.float64) + for col2 in field_columns + }, + ) + all_col_indices = [ + idx + 1 for idx, name in enumerate(col_names) if name.startswith("col_") + ] + col_indices = all_col_indices[-expected_path_len:] + # Fail fast if the cwrapper col_ leak pattern changes: the trailing + # slice assumes at least expected_path_len col_ cells. A count + # mismatch means the heuristic is stale and could silently pick the + # wrong path columns (returning wrong-device data). Guards the count + # only -- it can't catch a leak where the genuine columns are no + # longer the trailing ones. + assert len(col_indices) == expected_path_len, ( + f"tree path col_ columns: expected {expected_path_len}, got " + f"{len(col_indices)} of {len(all_col_indices)}; cwrapper col_ " + f"leak pattern may have changed" + ) + while result_set.next(): + row_path_segments = [ + result_set.get_value_by_index(ci) for ci in col_indices + ] + while row_path_segments and row_path_segments[-1] is None: + row_path_segments.pop() + if row_path_segments != target_path_segments: + continue + ts = int(result_set.get_value_by_index(1)) + # The on-tree scan already honors start/end_time at the + # cwrapper level, but defensively re-clip on the boundary. + if ts < start_time or ts > end_time: + continue + timestamps.append(ts) + for col, vidx in value_indices.items(): + raw = result_set.get_value_by_index(vidx) + value_buckets[col].append(np.nan if raw is None else float(raw)) + + if not timestamps: + return ( + np.array([], dtype=np.int64), + {col: np.array([], dtype=np.float64) for col in field_columns}, + ) + + timestamps_arr = np.asarray(timestamps, dtype=np.int64) + field_values = { + col: np.asarray(value_buckets[col], dtype=np.float64) + for col in field_columns + } + return timestamps_arr, field_values
