From 3d571562afc0500759ffc5b5918ae2dc2c602d8e Mon Sep 17 00:00:00 2001 From: Young-Leo <562593859@qq.com> Date: Tue, 13 Oct 2026 20:20:11 +0800 Subject: [PATCH] fix(python): read and index tree-model sequences independently --- cpp/src/dataset/dataset_index.h | 2 + cpp/src/reader/aligned_chunk_reader.cc | 15 +- python/README-zh.md | 20 ++ python/README.md | 22 ++ python/tests/resources/README.md | 15 +- .../TreeDatasetFixtureGenerator.java | 69 +++++ .../tests/resources/tree_aligned_mixed.tsfile | Bin 0 -> 1603 bytes .../tree_aligned_mixed_ts2diff.tsfile | Bin 0 -> 1653 bytes python/tests/test_tree_dataset_training.py | 212 +++++++++++++++ python/tsfile/dataset/_index.pyx | 22 +- python/tsfile/dataset/index.py | 38 ++- python/tsfile/dataset/reader.py | 251 +++++------------- python/tsfile/dataset/runtime.py | 10 +- 13 files changed, 465 insertions(+), 211 deletions(-) create mode 100644 python/tests/resources/TreeDatasetFixtureGenerator.java create mode 100644 python/tests/resources/tree_aligned_mixed.tsfile create mode 100644 python/tests/resources/tree_aligned_mixed_ts2diff.tsfile create mode 100644 python/tests/test_tree_dataset_training.py 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 0000000000000000000000000000000000000000..a1780051131b6d21fa01becb2b73850c4d430f61 GIT binary patch literal 1603 zcmc&#J#Q015S^V{+h;2gL5ia&4Hi%VIuU*V*$6@bQ4~d{NNL4Ouw?Ah`9L_;Au1#a zstEBL;6Gd>ij*OBdQzsKpk&s!bDt!XthD!LcHi8-o!zgS^};F<8kPPL2ob` zjP_-J^&JfA01FEU#U_d_h>XHi$5K*1qewedvzYqpOQC+{>!26D4)nR8@phZzc}1~y zl9+41ESCM>Oa+X~Z2Tz38&OCJ{D)9(clzvtiiXXPVQ_BqYII(l26!2 zVjNOuBvag^2gaZAEApuhX(UTHIkRT8TZ;f=k>ju`=L$}$qr$-m>Mj*-p>egXD0|n$ zr4*(zUQ1pi-m83pN|)z(of?;$vNcfl)xC(h20vV(6CbipsOn62Y^&JWEhrOJWwe^P z5$3EI$wYH92EU~VKh1A3nU!}e9ym;s2Y;b`PiQsil)j{*Ww2T z?r<)gA2$>nxyi7*2)A2NCz2iYNdc&B0q$IcwfeFi?FCam+|+OM5brNq_nDr;lC`^2 qUq2lO`(DrQ?Y-1qIC(+82hYxW&$l+V9_{RIcQ+o>T0gt9pT{3JY^)am literal 0 HcmV?d00001 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 0000000000000000000000000000000000000000..44689e3986afec2cb7d35f8e561c84fd82fc61f5 GIT binary patch literal 1653 zcmd5-zi-n(6ux`DI8M|GAvIOhg$Njt1q`gcik6|EszN~s35KXSN~Dljj*H0VN^E6d zs!-LbbN&dkF(PIpMi?0oV&uK}p6%GQtekZI-n;j`yZ7GxaCdTVFf>ZQS~wm@eiZfw zqrqrj^=}=DL+J|DZi85C5>@KlGC8NRi-VlZT#zBBDmzWBV5T8S^?D^~>&n3w5fTBp2YXU_Bo~n;q1r>D+}tBe04*dFgAqm1 zzz&Jm6wp~XH-7)N*Ec5kFAvWN)v?jwYET~3sBV4iqXMERK4XXEqx>T4b*?S@622~V zROe$5|B(4A#Y(Jxi?rd?*0eo@@>~j2884-t@>CaZuJYKPn(<8WJ=FGcly!Y0W-j5`1#!Y-j#E_Q z%rdsu+c{c@OjaYq)#yafUNNEwv@?WXx!_6_Q%tSusl}O-(Pn?4{Yq)IJ5K5Cl_eeD zVy0SrCM&m2o^uh|!{QsRg>tTF@xe=ny%ll28FymU(eJbndRGc>DTbQcHlw|88blAw zD>H= 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