diff --git a/cpp/src/common/device_id.cc b/cpp/src/common/device_id.cc index e88cdac8a..0b5e6c107 100644 --- a/cpp/src/common/device_id.cc +++ b/cpp/src/common/device_id.cc @@ -103,6 +103,13 @@ std::string StringArrayDeviceID::get_device_name() const { } void StringArrayDeviceID::init_prefix_segments() { + // Idempotent: device IDs are cached and reused across queries, so clear + // previous prefixes before rebuilding to avoid accumulation and leaks. + for (const auto& prefix_segment : prefix_segments_) { + delete prefix_segment; + } + prefix_segments_.clear(); + #ifdef ENABLE_ANTLR4 auto splits = storage::PathNodesGenerator::invokeParser(*segments_[0]); #else @@ -130,6 +137,7 @@ int StringArrayDeviceID::serialize(common::ByteStream& write_stream) { int StringArrayDeviceID::deserialize(common::ByteStream& read_stream) { int ret = common::E_OK; + uint32_t num_segments; if (RET_FAIL(common::SerializationUtil::read_var_uint(num_segments, read_stream))) { diff --git a/cpp/test/common/device_id_test.cc b/cpp/test/common/device_id_test.cc index 9d97607ab..839258f67 100644 --- a/cpp/test/common/device_id_test.cc +++ b/cpp/test/common/device_id_test.cc @@ -99,4 +99,24 @@ TEST(DeviceIdTest, NullTagVsLiteralNullAreDistinct) { ASSERT_FALSE(null_first == literal_null); ASSERT_TRUE(null_first != literal_null); } + +// Regression: cached device IDs are reused across queries, so +// split_table_name() must be idempotent and not accumulate prefix segments. +TEST(DeviceIdTest, SplitTableNameIsIdempotent) { + StringArrayDeviceID device_id("root.ln.wf01.wt01"); + + const std::vector expected = {"root", "ln", "wf01", "wt01"}; + + for (int round = 0; round < 3; ++round) { + device_id.split_table_name(); + + ASSERT_EQ(static_cast(expected.size()), + device_id.get_split_seg_num()); + for (int i = 0; i < device_id.get_split_seg_num(); ++i) { + std::string* seg = device_id.get_split_segname_at(i); + ASSERT_NE(nullptr, seg); + ASSERT_EQ(expected[static_cast(i)], *seg); + } + } +} } // namespace storage diff --git a/python/tests/test_tsfile_dataset.py b/python/tests/test_tsfile_dataset.py index 12b847e9e..49d7ca379 100644 --- a/python/tests/test_tsfile_dataset.py +++ b/python/tests/test_tsfile_dataset.py @@ -1507,12 +1507,84 @@ def test_dataset_tree_model_series_access(tmp_path): np.testing.assert_array_equal(aligned.timestamps, np.arange(5, dtype=np.int64)) +def test_dataset_tree_model_reads_uppercase_measurement_names(tmp_path): + """Tree-model series with uppercase measurement names must read data. + + Regression: a measurement like ``Temperature``/``STATUS`` must return its + values, not an empty array, when read back through TsFileDataFrame. + """ + from tsfile import Field, RowRecord, TimeseriesSchema, TsFileWriter + + path = tmp_path / "tree_upper.tsfile" + writer = TsFileWriter(str(path)) + writer.register_timeseries( + "root.ln.wf01.wt01", TimeseriesSchema("Temperature", TSDataType.DOUBLE) + ) + writer.register_timeseries( + "root.ln.wf01.wt01", TimeseriesSchema("STATUS", TSDataType.INT32) + ) + for t in range(5): + writer.write_row_record( + RowRecord( + "root.ln.wf01.wt01", + t, + [ + Field("Temperature", float(t) + 0.5, TSDataType.DOUBLE), + Field("STATUS", t * 2, TSDataType.INT32), + ], + ) + ) + writer.close() + + with TsFileDataFrame(str(path), show_progress=False) as tsdf: + assert sorted(tsdf.list_timeseries()) == [ + "root.ln.wf01.wt01.STATUS", + "root.ln.wf01.wt01.Temperature", + ] + np.testing.assert_array_equal( + tsdf["root.ln.wf01.wt01.Temperature"][:], + np.array([0.5, 1.5, 2.5, 3.5, 4.5]), + ) + np.testing.assert_array_equal( + tsdf["root.ln.wf01.wt01.STATUS"][:], + np.array([0.0, 2.0, 4.0, 6.0, 8.0]), + ) + + +def test_tree_reader_handles_stale_path_columns_after_reused_queries(tmp_path): + """Reusing a reader must not leak prefix path state across queries. + + Reading one device series then another reuses the cached device id; stale + prefix segments used to mismatch the device and return empty data. + """ + path = tmp_path / "tree.tsfile" + _write_tree_file(path) + + with TsFileDataFrame(str(path), show_progress=False) as tsdf: + # First read establishes query state on the reader. + np.testing.assert_array_equal( + tsdf["root.ln.wf01.wt01.temperature"][:], + np.array([0.5, 1.5, 2.5, 3.5, 4.5]), + ) + # Second read reuses the same reader for another device. + np.testing.assert_array_equal( + tsdf["root.ln.wf02.wt02.status"][:], + np.array([0.0, 2.0, 4.0, 6.0, 8.0]), + ) + # Read back the first series to confirm alternating reads stay stable. + np.testing.assert_array_equal( + tsdf["root.ln.wf01.wt01.temperature"][:], + np.array([0.5, 1.5, 2.5, 3.5, 4.5]), + ) + + 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", diff --git a/python/tsfile/dataset/reader.py b/python/tsfile/dataset/reader.py index 9b77190e1..4e716af5b 100644 --- a/python/tsfile/dataset/reader.py +++ b/python/tsfile/dataset/reader.py @@ -575,16 +575,23 @@ def _read_series_by_row_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. + # The native tree query normalizes column names and path segments to + # lower case (table model is case-insensitive), so match the field name + # and device path case-insensitively to preserve the original casing. + target_path_lower = [seg.lower() for seg in target_path_segments] 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)] + lower_col_names = [name.lower() for name in col_names] try: - field_idx = col_names.index(field_name) + 1 + field_idx = lower_col_names.index(field_name.lower()) + 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_") + idx + 1 + for idx, name in enumerate(lower_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. @@ -607,7 +614,11 @@ def _read_series_by_row_tree( # 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: + row_path_lower = [ + seg.lower() if isinstance(seg, str) else seg + for seg in row_path_segments + ] + if row_path_lower != target_path_lower: continue if skipped < offset: skipped += 1 @@ -763,6 +774,7 @@ def _read_arrow_tree( ) target_path_segments = device_path.split(".") + target_path_lower = [seg.lower() for seg in target_path_segments] expected_path_len = ( max(len(t.tag_columns) for t in self._catalog.table_entries) + 1 ) @@ -773,16 +785,19 @@ def _read_arrow_tree( # 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. + # The native tree query lower-cases column names and path segments + # (table model is case-insensitive), so match both case-insensitively. 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)] + lower_col_names = [name.lower() for name in col_names] value_indices = {} for col in field_columns: try: - value_indices[col] = col_names.index(col) + 1 + value_indices[col] = lower_col_names.index(col.lower()) + 1 except ValueError: # Column missing (no device in file owns it). Yield empty. return ( @@ -793,7 +808,9 @@ def _read_arrow_tree( }, ) all_col_indices = [ - idx + 1 for idx, name in enumerate(col_names) if name.startswith("col_") + idx + 1 + for idx, name in enumerate(lower_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 @@ -813,7 +830,11 @@ def _read_arrow_tree( ] while row_path_segments and row_path_segments[-1] is None: row_path_segments.pop() - if row_path_segments != target_path_segments: + row_path_lower = [ + seg.lower() if isinstance(seg, str) else seg + for seg in row_path_segments + ] + if row_path_lower != target_path_lower: continue ts = int(result_set.get_value_by_index(1)) # The on-tree scan already honors start/end_time at the