diff --git a/cpp/src/dataset/dataset_index.h b/cpp/src/dataset/dataset_index.h index faec4840e..453d75721 100644 --- a/cpp/src/dataset/dataset_index.h +++ b/cpp/src/dataset/dataset_index.h @@ -142,6 +142,8 @@ struct ColumnSchemaRecord { uint32_t name_sid; uint32_t column_ordinal; uint16_t logical_type; + // 0xFFFF denotes mixed physical types on independent tree devices. + // Decode each series using its located TimeseriesMetadata. uint16_t physical_type; uint16_t encoding; uint16_t compression; diff --git a/cpp/src/reader/aligned_chunk_reader.cc b/cpp/src/reader/aligned_chunk_reader.cc index ff4552cc8..fff8211d7 100644 --- a/cpp/src/reader/aligned_chunk_reader.cc +++ b/cpp/src/reader/aligned_chunk_reader.cc @@ -37,16 +37,15 @@ int AlignedChunkReader::init(RandomAccessReadFile* read_file, String m_name, TSDataType data_type, Filter* time_filter) { read_file_ = read_file; measurement_name_.shallow_copy_from(m_name); - time_decoder_ = DecoderFactory::alloc_time_decoder(); + // Aligned time chunks carry their own encoding. Allocate after reading the + // chunk header instead of fixing the decoder to the process default. + time_decoder_ = nullptr; value_decoder_ = nullptr; time_compressor_ = nullptr; value_compressor_ = nullptr; time_filter_ = time_filter; time_uncompressed_buf_ = nullptr; value_uncompressed_buf_ = nullptr; - if (IS_NULL(time_decoder_)) { - return E_OOM; - } return E_OK; } @@ -317,7 +316,7 @@ int AlignedChunkReader::load_by_aligned_meta(ChunkMeta* time_chunk_meta, } else if (RET_FAIL(alloc_compressor_and_decoder( time_decoder_, time_compressor_, time_chunk_header_.encoding_type_, - time_chunk_header_.data_type_, + common::INT64, time_chunk_header_.compression_type_))) { } else if (RET_FAIL(alloc_compressor_and_decoder( value_decoder_, value_compressor_, @@ -1264,7 +1263,7 @@ int AlignedChunkReader::load_by_aligned_meta_multi( if (RET_FAIL(alloc_compressor_and_decoder( time_decoder_, time_compressor_, time_chunk_header_.encoding_type_, - time_chunk_header_.data_type_, + common::INT64, time_chunk_header_.compression_type_))) { return ret; } @@ -1833,7 +1832,9 @@ int AlignedChunkReader::decode_all_planned_pages() { for (size_t w = 0; w < worker_count; w++) { if (time_decoder_pool_[w] == nullptr) { time_decoder_pool_[w] = - DecoderFactory::alloc_time_decoder(); + DecoderFactory::alloc_value_decoder( + time_chunk_header_.encoding_type_, + common::INT64); if (time_decoder_pool_[w] == nullptr) return E_OOM; } if (time_compressor_pool_[w] == nullptr) { diff --git a/python/README-zh.md b/python/README-zh.md index 59d2eb0f1..89fd39b96 100644 --- a/python/README-zh.md +++ b/python/README-zh.md @@ -68,6 +68,26 @@ mvn -P with-cpp,with-python clean verify python setup.py build_ext --inplace ``` +## 树模型数值序列 + +V4 树模型 TsFile 可以使用与表模型相同的 `TsFileDataFrame` 序列读取和窗口切片接口。 +通过完整逻辑路径选择测点。不同设备的同名测点可以具有不同的物理类型； +数值统一返回为 `float64` 数组，空值保留为 `NaN`。 + +```python +from tsfile import TsFileDataFrame + +with TsFileDataFrame("tree_data/", use_index=True) as dataset: + with dataset["root.training.d1.temperature"] as series: + window = series[0:128] + timestamps = series.timestamps + statistics = series.stats # start_time、end_time、count +``` + +统计信息和按位置切片以序列时间线为准，包括对齐数据中的空值位置。 +索引仍会校验同一完整路径在不同文件分片中的物理类型是否一致。 +表模型与树模型文件应分别加载到不同的 DataFrame；数值视图不包含非数值测点。 + ## Dataset 索引 通过 `use_index=True` 启用持久化 Dataset 索引。第一次打开时构建索引，之后 diff --git a/python/README.md b/python/README.md index 60607fe7d..674c039c7 100644 --- a/python/README.md +++ b/python/README.md @@ -122,6 +122,28 @@ with TsFileReader("example.tsfile") as reader: Values do not carry a data type; use an explicit portable encoding when storing numbers or structures. +## Numeric tree-model sequences + +V4 tree-model TsFiles can use the same `TsFileDataFrame` sequence and window APIs +as table-model files. Select each measurement by its full logical path. Different +devices may have different physical types for the same measurement name; numeric +values are returned as `float64` arrays, with null values preserved as `NaN`. + +```python +from tsfile import TsFileDataFrame + +with TsFileDataFrame("tree_data/", use_index=True) as dataset: + with dataset["root.training.d1.temperature"] as series: + window = series[0:128] + timestamps = series.timestamps + statistics = series.stats # start_time, end_time, count +``` + +Statistics and positional slices follow the sequence's timeline, including null +positions in aligned data. The index validates physical types across shards of +the same full path. Load table-model and tree-model files in separate DataFrames; +non-numeric measurements are omitted from the numeric dataset view. + ## Dataset indexes Enable the persistent dataset index with `use_index=True`. The first open diff --git a/python/tests/resources/README.md b/python/tests/resources/README.md index d5ec82b49..c19519232 100644 --- a/python/tests/resources/README.md +++ b/python/tests/resources/README.md @@ -19,6 +19,20 @@ --> +## Aligned numeric tree fixtures + +`tree_aligned_mixed.tsfile` and `tree_aligned_mixed_ts2diff.tsfile` are V4 files +generated by `TreeDatasetFixtureGenerator.java` using the Java TsFile writer. +The generator accepts the output path and an optional time encoding (`PLAIN` +by default, or `TS_2DIFF`). Both use uncompressed two-row pages. + +The fixtures contain six timestamps (0 through 5) on two aligned devices: +`root.training.d1` has DOUBLE temperature and INT32 status, while +`root.training.d2.deep` has INT32 temperature and BOOLEAN status. Temperature +is null at timestamp 2 on the first device and timestamp 3 on the second. +They verify that the Python numeric dataset reads each sequence with its own +physical type, preserves nulls, and counts the shared timeline consistently. + In `simple_tree.tsfile`: ```sql @@ -332,4 +346,3 @@ Total line number = 4 It costs 0.065s IoTDB:mydb> ``` - diff --git a/python/tests/resources/TreeDatasetFixtureGenerator.java b/python/tests/resources/TreeDatasetFixtureGenerator.java new file mode 100644 index 000000000..9b22d47e6 --- /dev/null +++ b/python/tests/resources/TreeDatasetFixtureGenerator.java @@ -0,0 +1,69 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import org.apache.tsfile.common.conf.TSFileDescriptor; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.enums.CompressionType; +import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.read.common.Path; +import org.apache.tsfile.write.TsFileWriter; +import org.apache.tsfile.write.record.TSRecord; +import org.apache.tsfile.write.schema.MeasurementSchema; + +import java.io.File; +import java.util.Arrays; + +/** Generates aligned tree fixtures for the Python sequence training tests. */ +public class TreeDatasetFixtureGenerator { + public static void main(String[] args) throws Exception { + TSFileDescriptor.getInstance() + .getConfig() + .setTimeEncoder(args.length > 1 ? args[1] : "PLAIN"); + TSFileDescriptor.getInstance().getConfig().setCompressor("UNCOMPRESSED"); + TSFileDescriptor.getInstance().getConfig().setMaxNumberOfPointsInPage(2); + TSFileDescriptor.getInstance().getConfig().setMaxDegreeOfIndexNode(2); + try (TsFileWriter writer = new TsFileWriter(new File(args[0]))) { + writer.registerAlignedTimeseries( + new Path("root.training.d1"), + Arrays.asList( + schema("temperature", TSDataType.DOUBLE), schema("status", TSDataType.INT32))); + writer.registerAlignedTimeseries( + new Path("root.training.d2.deep"), + Arrays.asList( + schema("temperature", TSDataType.INT32), schema("status", TSDataType.BOOLEAN))); + for (int t = 0; t < 6; t++) { + TSRecord first = new TSRecord("root.training.d1", t).addPoint("status", t); + if (t != 2) { + first.addPoint("temperature", t + 0.25); + } + writer.writeRecord(first); + TSRecord second = + new TSRecord("root.training.d2.deep", t).addPoint("status", t % 2 == 0); + if (t != 3) { + second.addPoint("temperature", t * 10 + 100); + } + writer.writeRecord(second); + } + } + } + + private static MeasurementSchema schema(String name, TSDataType type) { + return new MeasurementSchema(name, type, TSEncoding.PLAIN, CompressionType.UNCOMPRESSED); + } +} diff --git a/python/tests/resources/tree_aligned_mixed.tsfile b/python/tests/resources/tree_aligned_mixed.tsfile new file mode 100644 index 000000000..a17800511 Binary files /dev/null and b/python/tests/resources/tree_aligned_mixed.tsfile differ diff --git a/python/tests/resources/tree_aligned_mixed_ts2diff.tsfile b/python/tests/resources/tree_aligned_mixed_ts2diff.tsfile new file mode 100644 index 000000000..44689e398 Binary files /dev/null and b/python/tests/resources/tree_aligned_mixed_ts2diff.tsfile differ diff --git a/python/tests/test_tree_dataset_training.py b/python/tests/test_tree_dataset_training.py new file mode 100644 index 000000000..5eb0b0459 --- /dev/null +++ b/python/tests/test_tree_dataset_training.py @@ -0,0 +1,212 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Tree-model sequence reads, statistics, and numeric training windows.""" + +from pathlib import Path +import shutil + +import numpy as np +import pandas as pd +import pytest + +from tsfile import ( + Field, + RowRecord, + TSDataType, + TimeseriesSchema, + TsFileDataFrame, + TsFileWriter, +) + + +def _value(dtype, timestamp): + if dtype == TSDataType.BOOLEAN: + return timestamp % 2 == 0 + if dtype in {TSDataType.FLOAT, TSDataType.DOUBLE}: + return timestamp + 0.25 + return timestamp * 10 + 100 + + +def _write_tree(path, specs, start, stop): + with TsFileWriter(str(path)) as writer: + for device, field, dtype in specs: + writer.register_timeseries(device, TimeseriesSchema(field, dtype)) + for timestamp in range(start, stop): + for device, field, dtype in specs: + writer.write_row_record( + RowRecord( + device, + timestamp, + [Field(field, _value(dtype, timestamp), dtype)], + ) + ) + + +@pytest.mark.parametrize("use_index", [False, True], ids=["no-index", "index"]) +@pytest.mark.parametrize("shards", [1, 2]) +@pytest.mark.parametrize( + "other_type", + [ + TSDataType.INT32, + TSDataType.INT64, + TSDataType.FLOAT, + TSDataType.BOOLEAN, + TSDataType.TIMESTAMP, + ], +) +def test_tree_independent_measurement_types_read_training_windows( + tmp_path, monkeypatch, use_index, shards, other_type +): + specs = [ + ("root.training.d1", "temperature", TSDataType.DOUBLE), + ("root.training.d2.deep", "temperature", other_type), + ] + paths = [] + for shard in range(shards): + path = tmp_path / f"part{shard}.tsfile" + _write_tree(path, specs, shard * 6, (shard + 1) * 6) + paths.append(str(path)) + timestamps = np.arange(shards * 6, dtype=np.int64) + names = [f"{device}.{field}" for device, field, _ in specs] + expected = { + f"{device}.{field}": np.array( + [_value(dtype, int(t)) for t in timestamps], dtype=np.float64 + ) + for device, field, dtype in specs + } + + for load in range(2 if use_index else 1): + if load: + + def fail_scan(*_args, **_kwargs): + raise AssertionError("hot tree reload must use the persisted index") + + monkeypatch.setattr( + TsFileDataFrame, "_load_metadata_without_index", fail_scan + ) + with TsFileDataFrame(paths, show_progress=False, use_index=use_index) as frame: + assert frame.model == "tree" + assert sorted(frame.list_timeseries()) == sorted(names) + metadata = frame.list_timeseries_metadata() + for name in names: + with frame[name] as series: + np.testing.assert_array_equal(series.timestamps, timestamps) + np.testing.assert_array_equal(series[:], expected[name]) + np.testing.assert_array_equal(series[1:5], expected[name][1:5]) + np.testing.assert_array_equal(series[::-1], expected[name][::-1]) + assert series.stats == { + "start_time": 0, + "end_time": int(timestamps[-1]), + "count": len(timestamps), + } + windows = np.stack( + [series[start : start + 3] for start in (0, 2, 3)] + ) + assert windows.shape == (3, 3) + assert windows.dtype == np.float64 + assert np.isfinite(windows).all() + assert metadata.loc[name, "count"] == len(timestamps) + assert metadata.loc[name, "start_time"] == pd.Timestamp(0, unit="ms") + assert metadata.loc[name, "end_time"] == pd.Timestamp( + int(timestamps[-1]), unit="ms" + ) + batch = frame.loc[1:4, names] + np.testing.assert_array_equal(batch.timestamps, timestamps[1:5]) + np.testing.assert_array_equal( + batch.values, np.column_stack([expected[name][1:5] for name in names]) + ) + + +@pytest.mark.parametrize("use_index", [False, True], ids=["no-index", "index"]) +@pytest.mark.parametrize("field", ["_col_1", "col_1"]) +def test_tree_path_column_names_are_valid_measurements(tmp_path, use_index, field): + path = tmp_path / "tree.tsfile" + _write_tree(path, [("root.a.b", field, TSDataType.DOUBLE)], 0, 6) + name = f"root.a.b.{field}" + expected = np.arange(6, dtype=np.float64) + 0.25 + for _ in range(2 if use_index else 1): + with TsFileDataFrame( + str(path), show_progress=False, use_index=use_index + ) as frame: + assert frame.list_timeseries() == [name] + with frame[name] as series: + np.testing.assert_array_equal(series[:], expected) + assert series.stats == {"start_time": 0, "end_time": 5, "count": 6} + assert frame.list_timeseries_metadata().loc[name, "count"] == 6 + np.testing.assert_array_equal( + frame.loc[1:3, [name]].values[:, 0], expected[1:4] + ) + + +def test_tree_index_rejects_type_changes_in_the_same_series(tmp_path): + first, second = tmp_path / "first.tsfile", tmp_path / "second.tsfile" + _write_tree(first, [("root.a.b", "temperature", TSDataType.DOUBLE)], 0, 3) + _write_tree(second, [("root.a.b", "temperature", TSDataType.INT32)], 3, 6) + with pytest.raises( + ValueError, match=r"incompatible physical type for root\.a\.b\.temperature" + ): + TsFileDataFrame([str(first), str(second)], show_progress=False, use_index=True) + + +@pytest.mark.parametrize("use_index", [False, True], ids=["no-index", "index"]) +@pytest.mark.parametrize("time_encoding", ["plain", "ts2diff"]) +def test_java_aligned_tree_training_windows_preserve_nulls( + tmp_path, use_index, time_encoding +): + filename = ( + "tree_aligned_mixed.tsfile" + if time_encoding == "plain" + else "tree_aligned_mixed_ts2diff.tsfile" + ) + source = Path(__file__).parent / "resources" / filename + path = tmp_path / "aligned.tsfile" + shutil.copyfile(source, path) + expected = { + "root.training.d1.temperature": np.array( + [0.25, 1.25, np.nan, 3.25, 4.25, 5.25] + ), + "root.training.d1.status": np.arange(6, dtype=np.float64), + "root.training.d2.deep.temperature": np.array( + [100, 110, 120, np.nan, 140, 150] + ), + "root.training.d2.deep.status": np.array([1, 0, 1, 0, 1, 0], dtype=np.float64), + } + for _ in range(2 if use_index else 1): + with TsFileDataFrame( + str(path), show_progress=False, use_index=use_index + ) as frame: + metadata = frame.list_timeseries_metadata() + for name, values in expected.items(): + with frame[name] as series: + np.testing.assert_allclose(series[:], values, equal_nan=True) + np.testing.assert_allclose(series[1:5], values[1:5], equal_nan=True) + np.testing.assert_array_equal( + series.timestamps, np.arange(6, dtype=np.int64) + ) + assert series.stats == {"start_time": 0, "end_time": 5, "count": 6} + assert metadata.loc[name, "count"] == 6 + names = list(expected) + batch = frame.loc[1:4, names] + np.testing.assert_array_equal( + batch.timestamps, np.arange(1, 5, dtype=np.int64) + ) + np.testing.assert_allclose( + batch.values, + np.column_stack([expected[name][1:5] for name in names]), + equal_nan=True, + ) diff --git a/python/tsfile/dataset/_index.pyx b/python/tsfile/dataset/_index.pyx index fb7453705..1cc7584a1 100644 --- a/python/tsfile/dataset/_index.pyx +++ b/python/tsfile/dataset/_index.pyx @@ -108,6 +108,10 @@ cdef inline int64_t _find_child( uint64_t string_offsets_base, uint64_t string_bytes_base, uint32_t string_count, + uint64_t schema_base, + uint32_t schema_size, + uint32_t schema_count, + int role, ) noexcept nogil: cdef uint32_t low = first cdef uint32_t high = first + count @@ -144,6 +148,12 @@ cdef inline int64_t _find_child( length, ): row_id = _read_u32(view, row_base + 4) + if role >= 0: + if row_id >= schema_count: + return -2 + if _read_u16(view, schema_base + row_id * schema_size + 20) != role: + low += 1 + continue return row_id low += 1 return -1 @@ -207,12 +217,16 @@ cdef class IndexLookup: self._section_offsets[1], self._section_offsets[2], self._section_counts[1], + 0, + 0, + 0, + -1, ) if result < 0: raise KeyError(name) return result - def find_column_id(self, Py_ssize_t table_id, object name): + def find_column_id(self, Py_ssize_t table_id, object name, int role=-1): cdef const char* data cdef Py_ssize_t length cdef const uint8_t[:] view = self._view @@ -244,7 +258,13 @@ cdef class IndexLookup: self._section_offsets[1], self._section_offsets[2], self._section_counts[1], + self._section_offsets[8], + self._section_sizes[8], + self._section_counts[8], + role, ) + if result == -2: + raise IndexError(name) if result < 0: raise KeyError(name) return result diff --git a/python/tsfile/dataset/index.py b/python/tsfile/dataset/index.py index 7c1246ba0..1d6e13c6c 100644 --- a/python/tsfile/dataset/index.py +++ b/python/tsfile/dataset/index.py @@ -34,7 +34,7 @@ import struct from typing import Dict, Iterable, Iterator, List, Mapping, Optional, Sequence, Tuple -from ..constants import ColumnCategory +from ..constants import ColumnCategory, TSDataType from .metadata import MODEL_TREE, _join_series_path from ._index import IndexLookup @@ -47,6 +47,9 @@ ALIGNMENT = 64 INDEX_FILE_PREFIX = ".tsfile_dataframe_index" INDEX_FILE_SUFFIX = ".tsidx" +# The virtual tree column describes float64 output when devices have different +# physical types. Each locator's TimeseriesMetadata determines actual decoding. +MIXED_PHYSICAL_TYPE = 0xFFFF STRING_OFFSETS = 1 STRING_BYTES = 2 @@ -430,9 +433,11 @@ def find_device_id(self, table_id: int, name: str) -> int: self._assert_open() return self._lookup.find_device_id(table_id, name) - def find_column_id(self, table_id: int, name: str) -> int: + def find_column_id( + self, table_id: int, name: str, role: Optional[int] = None + ) -> int: self._assert_open() - return self._lookup.find_column_id(table_id, name) + return self._lookup.find_column_id(table_id, name, -1 if role is None else role) def find_series_id(self, device_id: int, column_id: int) -> int: self._assert_open() @@ -538,10 +543,11 @@ def build_sections_from_dataframe(dataframe) -> Mapping[int, bytes]: table_names = sorted(index.table_entries) table_ids = {name: table_id for table_id, name in enumerate(table_names)} + is_tree = getattr(index, "model", None) == MODEL_TREE field_types: Dict[Tuple[str, str], int] = {} for series_ref, shards in index.series_shards.items(): device_idx, field_idx = series_ref - table_name, _ = index.devices[device_idx] + table_name, tags = index.devices[device_idx] table_entry = index.table_entries[table_name] field_name = table_entry.field_columns[field_idx] declared_type = ( @@ -549,8 +555,16 @@ def build_sections_from_dataframe(dataframe) -> Mapping[int, bytes]: if field_idx < len(table_entry.field_types) else None ) + series_type = None for reader, local_device, local_field in shards: stats = reader.catalog.series_stats_by_ref[(local_device, local_field)] + if is_tree: + # Only shards of the same full path must agree. Measurements + # with the same name on different devices are independent. + if series_type is not None and series_type != stats.data_type: + name = _join_series_path(table_name, tags, field_name) + raise ValueError(f"incompatible physical type for {name}") + series_type = stats.data_type if declared_type is not None and declared_type != stats.data_type: raise ValueError( f"physical type for {table_name}.{field_name} does not match " @@ -558,9 +572,12 @@ def build_sections_from_dataframe(dataframe) -> Mapping[int, bytes]: ) current = field_types.setdefault((table_name, field_name), stats.data_type) if current != stats.data_type: - raise ValueError( - f"incompatible physical type for {table_name}.{field_name}" - ) + if is_tree: + field_types[(table_name, field_name)] = MIXED_PHYSICAL_TYPE + else: + raise ValueError( + f"incompatible physical type for {table_name}.{field_name}" + ) columns = [] columns_by_table = defaultdict(list) @@ -595,13 +612,18 @@ def build_sections_from_dataframe(dataframe) -> Mapping[int, bytes]: ] ) for ordinal, name, data_type, role in definitions: + logical_type = ( + int(TSDataType.DOUBLE) + if data_type == MIXED_PHYSICAL_TYPE + else data_type + ) column_id = len(columns) columns.append( ( table_id, pool.intern(name), ordinal, - data_type, + logical_type, data_type, 0, 0, diff --git a/python/tsfile/dataset/reader.py b/python/tsfile/dataset/reader.py index 98afe2afb..4d9e787d5 100644 --- a/python/tsfile/dataset/reader.py +++ b/python/tsfile/dataset/reader.py @@ -28,6 +28,7 @@ from ..tag_filter import tag_eq, tag_is_null from ..tsfile_reader import TsFileReaderPy from ._arrow import arrow_column_to_float64 +from .merge import build_aligned_matrix from .metadata import ( MetadataCatalog, MODEL_TABLE, @@ -89,6 +90,7 @@ def __init__(self, file_path: str, show_progress: bool = True): raise FileNotFoundError(f"TsFile not found: {file_path}") self.file_path = file_path + self._file_size = os.stat(file_path).st_size self.show_progress = show_progress try: @@ -492,6 +494,10 @@ def get_series_info(self, series_path: str) -> dict: def read_series_by_ref( self, device_id: int, field_idx: int, start_time: int, end_time: int ) -> Tuple[np.ndarray, np.ndarray]: + if self._model_kind == MODEL_TREE: + return self._read_tree_series( + device_id, field_idx, start_time=start_time, end_time=end_time + ) table_entry, _, field_name = self._resolve_series_ref(device_id, field_idx) timestamps, field_values = self.read_device_fields_by_time_range( device_id, [field_idx], start_time, end_time @@ -518,8 +524,9 @@ def read_series_by_row( ) 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) + return self._read_tree_series( + device_id, field_idx, offset=offset, limit=limit + ) tag_values = _device_exact_tag_values(table_entry, device_entry) tag_filter = _build_exact_tag_filter(tag_values) if tag_values else None @@ -568,74 +575,49 @@ def read_series_by_row( 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)) + def _read_tree_series(self, device_id: int, field_idx: int, **query): + """Read an exact series with its own physical type and timeline. - 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), + The all-device tree table query assumes a shared type for each + measurement name. Exact metadata locators avoid that assumption and + the synthetic path columns, even without a persisted Dataset Index. + """ + stats = self._catalog.series_stats_by_ref[(device_id, field_idx)] + locator = ( + id(self._catalog), + 0, + self._file_size, + 0, + 0, + stats.layout, + stats.locator_flags, + stats.value_metadata_offset, + stats.value_metadata_length, + stats.time_metadata_offset, + stats.time_metadata_length, ) + timestamp_parts = [] + value_parts = [] + with self._reader.prepare_series(locator) as prepared: + with self._reader.query_prepared(prepared, **query) as result_set: + while True: + batch = result_set.read_arrow_batch() + if batch is None: + break + if batch.num_rows == 0: + continue + timestamp_parts.append( + np.asarray( + batch.column(0).to_numpy(zero_copy_only=False), + dtype=np.int64, + ) + ) + value_parts.append(arrow_column_to_float64(batch.column(1))) + if not timestamp_parts: + return np.array([], dtype=np.int64), np.array([], dtype=np.float64) + if len(timestamp_parts) == 1: + return timestamp_parts[0], value_parts[0] + return np.concatenate(timestamp_parts), np.concatenate(value_parts) def read_device_fields_by_time_range( self, device_id: int, field_indices: List[int], start_time: int, end_time: int @@ -647,10 +629,16 @@ def read_device_fields_by_time_range( table_entry.field_columns[field_idx] for field_idx in field_indices ] 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 - ) + parts = { + name: self._read_tree_series( + device_id, field_idx, start_time=start_time, end_time=end_time + ) + for name, field_idx in zip(requested_field_columns, field_indices) + } + timestamps, values = build_aligned_matrix(requested_field_columns, parts) + return timestamps, { + name: values[:, idx] for idx, name in enumerate(requested_field_columns) + } return self._read_arrow_table( table_entry.table_name, requested_field_columns, @@ -660,34 +648,6 @@ def read_device_fields_by_time_range( end_time, ) - @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, @@ -756,94 +716,3 @@ def _read_arrow_table( } 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 diff --git a/python/tsfile/dataset/runtime.py b/python/tsfile/dataset/runtime.py index ab9df4b2a..c43bbba3b 100644 --- a/python/tsfile/dataset/runtime.py +++ b/python/tsfile/dataset/runtime.py @@ -32,7 +32,7 @@ import numpy as np -from ..constants import NUMERIC_DATASET_FIELD_TYPES, TSDataType +from ..constants import ColumnCategory, NUMERIC_DATASET_FIELD_TYPES, TSDataType from ..tag_filter import tag_eq, tag_is_null from ..tsfile_reader import TsFileReaderPy from ._arrow import arrow_column_to_float64 @@ -776,7 +776,9 @@ def _series_id(self, ref): self._catalog.index.table_name_id(table_id) ) field_name = self._catalog.table_entries[table_name].field_columns[field_idx] - column_id = self._catalog.index.find_column_id(table_id, field_name) + column_id = self._catalog.index.find_column_id( + table_id, field_name, role=int(ColumnCategory.FIELD) + ) return self._catalog.index.find_series_id(device_id, column_id) def describe(self, ref, series_id=None, column_id=None): @@ -885,7 +887,9 @@ def resolve_series_descriptor(self, table_name, tags, field_name): device_id = self.index.find_device_id( table_id, _join_series_path(table_name, tags, "") ) - column_id = self.index.find_column_id(table_id, field_name) + column_id = self.index.find_column_id( + table_id, field_name, role=int(ColumnCategory.FIELD) + ) series_id = self.index.find_series_id(device_id, column_id) result = self.series_shards.describe( (device_id, field_idx), series_id=series_id, column_id=column_id