This is an automated email from the ASF dual-hosted git repository.
ColinLeeo pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/tsfile.git
The following commit(s) were added to refs/heads/develop by this push:
new e6c0ddd23 fix(python): make dataframe index opt-in (#919)
e6c0ddd23 is described below
commit e6c0ddd237df5f00068a5da2beb12fb40763f2fb
Author: Colin Lee <[email protected]>
AuthorDate: Thu Aug 27 09:07:35 2026 +0800
fix(python): make dataframe index opt-in (#919)
* fix(python): make dataframe index opt-in
* test(python): cover dataframe index modes
---
python/tests/test_dataset_index.py | 77 ++++++++--
python/tests/test_tsfile_dataset.py | 295 ++++++++++++++++++++++++++----------
python/tsfile/dataset/dataframe.py | 80 +++++++---
python/tsfile/dataset/index.py | 16 +-
4 files changed, 355 insertions(+), 113 deletions(-)
diff --git a/python/tests/test_dataset_index.py
b/python/tests/test_dataset_index.py
index 417b54d58..91761a11a 100644
--- a/python/tests/test_dataset_index.py
+++ b/python/tests/test_dataset_index.py
@@ -244,14 +244,14 @@ def _write_runtime_devices_file(path):
def test_hot_construction_maps_index_without_opening_readers(tmp_path,
monkeypatch):
source = tmp_path / "part.tsfile"
_write_runtime_file(source, 0)
- with TsFileDataFrame(str(source), show_progress=False) as first:
+ with TsFileDataFrame(str(source), show_progress=False, use_index=True) as
first:
assert len(first) == 1
def fail_legacy_scan(*_args, **_kwargs):
raise AssertionError("hot construction must not build a legacy
catalog")
monkeypatch.setattr("tsfile.dataset.reader.TsFileSeriesReader",
fail_legacy_scan)
- with TsFileDataFrame(str(source), show_progress=False) as second:
+ with TsFileDataFrame(str(source), show_progress=False, use_index=True) as
second:
assert len(second) == 1
assert second._runtime.readers.open_count == 0
series = second[0]
@@ -264,11 +264,58 @@ def
test_hot_construction_maps_index_without_opening_readers(tmp_path, monkeypat
series.close()
-def test_named_selection_reuses_bounded_runtime_descriptor(tmp_path,
monkeypatch):
+def test_dataframe_does_not_use_or_create_index_by_default(tmp_path):
source = tmp_path / "part.tsfile"
_write_runtime_file(source, 0)
+ index_path = index_module.index_path_for([str(source)])
with TsFileDataFrame(str(source), show_progress=False) as dataframe:
+ assert dataframe._runtime is None
+ assert len(dataframe) == 1
+ np.testing.assert_array_equal(dataframe[0][:], np.array([0.0, 1.0]))
+ aligned = dataframe.loc[0:1, [0]]
+ np.testing.assert_array_equal(aligned.timestamps, np.array([0, 1]))
+ np.testing.assert_array_equal(aligned.values, np.array([[0.0], [1.0]]))
+
+ assert not os.path.exists(index_path)
+
+
+def test_persistent_index_path_is_scoped_to_expanded_file_set(tmp_path,
monkeypatch):
+ first = tmp_path / "part1.tsfile"
+ second = tmp_path / "part2.tsfile"
+ third = tmp_path / "part3.tsfile"
+ _write_runtime_file(first, 0)
+ _write_runtime_file(second, 2)
+ _write_runtime_file(third, 10)
+
+ first_set = [str(first), str(second)]
+ second_set = [str(first), str(third)]
+ first_index = index_module.index_path_for(first_set)
+ second_index = index_module.index_path_for(second_set)
+ assert first_index != second_index
+
+ with TsFileDataFrame(first_set, show_progress=False, use_index=True) as
dataframe:
+ assert len(dataframe) == 1
+ with TsFileDataFrame(second_set, show_progress=False, use_index=True) as
dataframe:
+ assert len(dataframe) == 1
+ assert os.path.exists(first_index)
+ assert os.path.exists(second_index)
+
+ def fail_legacy_scan(*_args, **_kwargs):
+ raise AssertionError("matching file-set index should be reused")
+
+ monkeypatch.setattr("tsfile.dataset.reader.TsFileSeriesReader",
fail_legacy_scan)
+ with TsFileDataFrame(first_set, show_progress=False, use_index=True) as
dataframe:
+ np.testing.assert_array_equal(dataframe[0][:], np.array([0.0, 1.0,
2.0, 3.0]))
+ with TsFileDataFrame(second_set, show_progress=False, use_index=True) as
dataframe:
+ np.testing.assert_array_equal(dataframe[0][:], np.array([0.0, 1.0,
10.0, 11.0]))
+
+
+def test_named_selection_reuses_bounded_runtime_descriptor(tmp_path,
monkeypatch):
+ source = tmp_path / "part.tsfile"
+ _write_runtime_file(source, 0)
+
+ with TsFileDataFrame(str(source), show_progress=False, use_index=True) as
dataframe:
name = str(dataframe.list_timeseries()[0])
find_device_calls = 0
original_find_device = dataframe._runtime.index.find_device_id
@@ -307,7 +354,7 @@ def
test_listed_series_path_resolves_directly_by_snapshot_series_id(
source = tmp_path / "part.tsfile"
_write_runtime_file(source, 0)
- with TsFileDataFrame(str(source), show_progress=False) as dataframe:
+ with TsFileDataFrame(str(source), show_progress=False, use_index=True) as
dataframe:
path = dataframe.list_timeseries()[0]
assert isinstance(path, str)
assert path.series_id == 0
@@ -346,9 +393,13 @@ def
test_series_path_from_another_index_falls_back_to_its_name(tmp_path, monkeyp
_write_runtime_file(first_source, 0)
_write_runtime_file(second_source, 10)
- with TsFileDataFrame(str(first_source), show_progress=False) as first:
+ with TsFileDataFrame(
+ str(first_source), show_progress=False, use_index=True
+ ) as first:
foreign_path = first.list_timeseries()[0]
- with TsFileDataFrame(str(second_source), show_progress=False) as
second:
+ with TsFileDataFrame(
+ str(second_source), show_progress=False, use_index=True
+ ) as second:
assert foreign_path._index_identity !=
second._runtime.index.identity
find_device_calls = 0
original_find_device = second._runtime.index.find_device_id
@@ -372,7 +423,7 @@ def
test_runtime_descriptor_cache_evicts_least_recent_name(tmp_path, monkeypatch
_write_runtime_devices_file(source)
monkeypatch.setattr(runtime_module, "_SERIES_DESCRIPTOR_CACHE_SIZE", 2)
- with TsFileDataFrame(str(source), show_progress=False) as dataframe:
+ with TsFileDataFrame(str(source), show_progress=False, use_index=True) as
dataframe:
names = [str(name) for name in dataframe.list_timeseries()]
find_device_calls = 0
original_find_device = dataframe._runtime.index.find_device_id
@@ -403,7 +454,9 @@ def test_reader_pool_enforces_open_file_cap(tmp_path,
monkeypatch):
_write_runtime_file(first, 0)
_write_runtime_file(second, 2)
monkeypatch.setenv("TSFILE_DATAFRAME_MAX_OPEN_FILES", "1")
- with TsFileDataFrame([str(first), str(second)], show_progress=False) as
dataframe:
+ with TsFileDataFrame(
+ [str(first), str(second)], show_progress=False, use_index=True
+ ) as dataframe:
series = dataframe[0]
assert list(series[:]) == [0.0, 1.0, 2.0, 3.0]
assert dataframe._runtime.readers.open_count == 1
@@ -481,7 +534,7 @@ def
test_prepared_query_reads_nullable_offset_window_in_arrow_batches(tmp_path):
)
)
- with TsFileDataFrame(str(source), show_progress=False) as dataframe:
+ with TsFileDataFrame(str(source), show_progress=False, use_index=True) as
dataframe:
runtime = dataframe._runtime
series = runtime.index.record(LOGICAL_SERIES, 0)
span = runtime.index.record(SERIES_FILE_SPAN, series[2])
@@ -514,7 +567,7 @@ def
test_prepared_query_reads_nullable_offset_window_in_arrow_batches(tmp_path):
def test_prepared_locator_rejects_stale_generation_and_bad_range(tmp_path):
source = tmp_path / "part.tsfile"
_write_runtime_file(source, 0)
- with TsFileDataFrame(str(source), show_progress=False) as dataframe:
+ with TsFileDataFrame(str(source), show_progress=False, use_index=True) as
dataframe:
runtime = dataframe._runtime
series = runtime.index.record(LOGICAL_SERIES, 0)
span = runtime.index.record(SERIES_FILE_SPAN, series[2])
@@ -534,7 +587,7 @@ def
test_prepared_locator_rejects_stale_generation_and_bad_range(tmp_path):
def test_reader_session_revalidates_generation_when_reused(tmp_path):
source = tmp_path / "part.tsfile"
_write_runtime_file(source, 0)
- with TsFileDataFrame(str(source), show_progress=False) as dataframe:
+ with TsFileDataFrame(str(source), show_progress=False, use_index=True) as
dataframe:
pool = dataframe._runtime.readers
with pool.acquire(0):
pass
@@ -548,7 +601,7 @@ def
test_reader_session_revalidates_generation_when_reused(tmp_path):
def test_runtime_lease_close_waits_for_query_lease(tmp_path):
source = tmp_path / "part.tsfile"
_write_runtime_file(source, 0)
- with TsFileDataFrame(str(source), show_progress=False) as dataframe:
+ with TsFileDataFrame(str(source), show_progress=False, use_index=True) as
dataframe:
runtime = DatasetRuntime(str(dataframe._runtime.index.path),
query_workers=1)
lease = runtime.lease()
entered = threading.Event()
diff --git a/python/tests/test_tsfile_dataset.py
b/python/tests/test_tsfile_dataset.py
index 55115987e..499f2cf2c 100644
--- a/python/tests/test_tsfile_dataset.py
+++ b/python/tests/test_tsfile_dataset.py
@@ -45,6 +45,11 @@ from tsfile.dataset.reader import (
from tsfile.dataset.runtime import RuntimeSeriesReader
[email protected](params=[False, True], ids=["no-index", "index"])
+def dataframe_use_index(request):
+ return request.param
+
+
def _write_weather_file(path, start):
schema = TableSchema(
"weather",
@@ -259,13 +264,17 @@ def
test_format_timestamp_preserves_millisecond_precision():
assert format_timestamp(1).endswith(".001")
-def test_dataset_basic_access_patterns(tmp_path, capsys):
+def test_dataset_basic_access_patterns(tmp_path, capsys, dataframe_use_index):
path1 = tmp_path / "part1.tsfile"
path2 = tmp_path / "part2.tsfile"
_write_weather_file(path1, 0)
_write_weather_file(path2, 3)
- with TsFileDataFrame([str(path1), str(path2)], show_progress=False) as
tsdf:
+ with TsFileDataFrame(
+ [str(path1), str(path2)],
+ show_progress=False,
+ use_index=dataframe_use_index,
+ ) as tsdf:
assert len(tsdf) == 2
first = tsdf[0]
@@ -303,7 +312,9 @@ def test_dataset_basic_access_patterns(tmp_path, capsys):
assert "AlignedTimeseries(6 rows, 2 series)" in capsys.readouterr().out
-def
test_dataset_loc_aligns_timestamp_union_and_preserves_requested_order(tmp_path):
+def test_dataset_loc_aligns_timestamp_union_and_preserves_requested_order(
+ tmp_path, dataframe_use_index
+):
path = tmp_path / "weather_sparse.tsfile"
_write_weather_rows_file(
path,
@@ -315,7 +326,9 @@ def
test_dataset_loc_aligns_timestamp_union_and_preserves_requested_order(tmp_pa
},
)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
aligned = tsdf.loc[
0:2,
[
@@ -341,7 +354,9 @@ def
test_dataset_loc_aligns_timestamp_union_and_preserves_requested_order(tmp_pa
assert aligned.values[2, 1] == 30.0
-def
test_dataset_loc_batches_aligned_fields_per_device_then_unions_devices(tmp_path):
+def test_dataset_loc_batches_aligned_fields_per_device_then_unions_devices(
+ tmp_path, dataframe_use_index
+):
path = tmp_path / "weather_multi_device.tsfile"
_write_weather_rows_file(
path,
@@ -358,7 +373,9 @@ def
test_dataset_loc_batches_aligned_fields_per_device_then_unions_devices(tmp_p
"weather.device_a.humidity",
"weather.device_b.humidity",
]
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
aligned = tsdf.loc[0:2, requested]
assert aligned.series_names == requested
@@ -407,7 +424,7 @@ def
test_dataset_loc_runs_independent_device_groups_concurrently(tmp_path, monke
observed_read,
)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(str(path), show_progress=False, use_index=True) as
tsdf:
aligned = tsdf.loc[
0:1,
[
@@ -452,7 +469,7 @@ def
test_dataset_loc_keeps_small_device_groups_inline(tmp_path, monkeypatch):
observed_read,
)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(str(path), show_progress=False, use_index=True) as
tsdf:
tsdf.loc[
0:1,
[
@@ -464,7 +481,7 @@ def
test_dataset_loc_keeps_small_device_groups_inline(tmp_path, monkeypatch):
assert worker_ids == {caller_id}
-def test_dataset_reads_nullable_tag_devices_in_isolation(tmp_path):
+def test_dataset_reads_nullable_tag_devices_in_isolation(tmp_path,
dataframe_use_index):
path = tmp_path / "nullable_tags.tsfile"
schema = TableSchema(
"sensors",
@@ -506,7 +523,9 @@ def
test_dataset_reads_nullable_tag_devices_in_isolation(tmp_path):
writer.write_dataframe(null_device)
writer.write_dataframe(full)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
series = tsdf.list_timeseries()
# Null tags keep their position via the \N marker; trailing nulls drop.
assert set(series) == {
@@ -592,7 +611,9 @@ def
test_split_logical_series_path_null_marker_only_whole_component():
split_logical_series_path(bad)
-def test_dataset_null_tag_positions_and_string_null_are_distinct(tmp_path):
+def test_dataset_null_tag_positions_and_string_null_are_distinct(
+ tmp_path, dataframe_use_index
+):
path = tmp_path / "null_positions.tsfile"
schema = TableSchema(
"a",
@@ -622,7 +643,9 @@ def
test_dataset_null_tag_positions_and_string_null_are_distinct(tmp_path):
)
)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
series = tsdf.list_timeseries()
# Nothing collapses: three physically distinct devices stay distinct.
assert len(series) == 3
@@ -648,11 +671,15 @@ def
test_dataset_null_tag_positions_and_string_null_are_distinct(tmp_path):
)
-def
test_dataset_loc_supports_single_timestamp_and_mixed_series_specifiers(tmp_path):
+def test_dataset_loc_supports_single_timestamp_and_mixed_series_specifiers(
+ tmp_path, dataframe_use_index
+):
path = tmp_path / "weather.tsfile"
_write_weather_file(path, 0)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
aligned = tsdf.loc[1, [0, "weather.device_a.humidity"]]
assert isinstance(aligned, AlignedTimeseries)
@@ -664,11 +691,13 @@ 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):
+def test_dataset_loc_dedups_repeated_series_specifiers(tmp_path,
dataframe_use_index):
path = tmp_path / "weather.tsfile"
_write_weather_file(path, 0)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
humidity = "weather.device_a.humidity"
humidity_idx = tsdf.list_timeseries().index(humidity)
@@ -712,11 +741,15 @@ def
test_dataset_loc_dedups_repeated_series_specifiers(tmp_path):
)
-def
test_dataset_loc_supports_open_ended_ranges_and_negative_series_index(tmp_path):
+def test_dataset_loc_supports_open_ended_ranges_and_negative_series_index(
+ tmp_path, dataframe_use_index
+):
path = tmp_path / "weather.tsfile"
_write_weather_file(path, 100)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
aligned = tsdf.loc[:101, [-1]]
assert isinstance(aligned, AlignedTimeseries)
@@ -727,7 +760,9 @@ def
test_dataset_loc_supports_open_ended_ranges_and_negative_series_index(tmp_pa
np.testing.assert_array_equal(aligned.values, np.array([[50.0],
[52.0]]))
-def
test_dataset_loc_with_nulls_does_not_expand_beyond_requested_time_range(tmp_path):
+def test_dataset_loc_with_nulls_does_not_expand_beyond_requested_time_range(
+ tmp_path, dataframe_use_index
+):
path = tmp_path / "weather_sparse_range.tsfile"
_write_weather_rows_file(
path,
@@ -739,7 +774,9 @@ def
test_dataset_loc_with_nulls_does_not_expand_beyond_requested_time_range(tmp_
},
)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
aligned = tsdf.loc[
1:2,
[
@@ -759,7 +796,9 @@ def
test_dataset_loc_with_nulls_does_not_expand_beyond_requested_time_range(tmp_
assert np.isnan(aligned.values[1, 1])
-def
test_dataset_loc_single_timestamp_with_nulls_keeps_exact_time_window(tmp_path):
+def test_dataset_loc_single_timestamp_with_nulls_keeps_exact_time_window(
+ tmp_path, dataframe_use_index
+):
path = tmp_path / "weather_sparse_point.tsfile"
_write_weather_rows_file(
path,
@@ -771,7 +810,9 @@ def
test_dataset_loc_single_timestamp_with_nulls_keeps_exact_time_window(tmp_pat
},
)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
aligned = tsdf.loc[
1,
[
@@ -787,11 +828,15 @@ def
test_dataset_loc_single_timestamp_with_nulls_keeps_exact_time_window(tmp_pat
assert aligned.values[0, 1] == 20.0
-def test_dataset_repr_only_builds_preview_rows(tmp_path, monkeypatch):
+def test_dataset_repr_only_builds_preview_rows(
+ tmp_path, monkeypatch, dataframe_use_index
+):
path = tmp_path / "weather.tsfile"
_write_weather_file(path, 0)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
tsdf._index.series = [(0, 0)] * 1000
built_rows = []
@@ -822,11 +867,15 @@ def test_dataset_repr_only_builds_preview_rows(tmp_path,
monkeypatch):
assert len(built_rows) == 20
-def test_dataset_exposes_only_numeric_fields_and_keeps_nan(tmp_path):
+def test_dataset_exposes_only_numeric_fields_and_keeps_nan(
+ tmp_path, dataframe_use_index
+):
path = tmp_path / "numeric_and_text.tsfile"
_write_numeric_and_text_file(path)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
assert tsdf.list_timeseries() == ["weather.device_a.temperature"]
series = tsdf[0]
@@ -844,7 +893,9 @@ 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):
+def test_dataset_omits_table_model_phantom_series_for_skipped_cells(
+ tmp_path, dataframe_use_index
+):
"""Schema-declared fields that a device never wrote must NOT appear.
The dataset surface treats a series as "data physically written for one
@@ -900,7 +951,9 @@ def
test_dataset_omits_table_model_phantom_series_for_skipped_cells(tmp_path):
t2.add_value_by_name("v3", 0, 330.0)
writer.write_table(t2)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
# 4 real cells: (d1,v1), (d1,v2), (d2,v1), (d2,v3); NO phantoms.
assert len(tsdf) == 4
assert sorted(tsdf.list_timeseries()) == [
@@ -926,21 +979,29 @@ def
test_dataset_omits_table_model_phantom_series_for_skipped_cells(tmp_path):
reader.close()
-def test_dataset_timeseries_supports_negative_step_slices(tmp_path):
+def test_dataset_timeseries_supports_negative_step_slices(
+ tmp_path, dataframe_use_index
+):
path = tmp_path / "weather.tsfile"
_write_weather_file(path, 0)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
series = tsdf[0]
np.testing.assert_array_equal(series[::-1], np.array([23.0, 21.5,
20.0]))
np.testing.assert_array_equal(series[::-2], np.array([23.0, 20.0]))
-def test_dataset_metadata_discovery_uses_all_numeric_fields(tmp_path):
+def test_dataset_metadata_discovery_uses_all_numeric_fields(
+ tmp_path, dataframe_use_index
+):
path = tmp_path / "partial_numeric_rows.tsfile"
_write_partial_numeric_rows_file(path)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
assert tsdf.list_timeseries() == [
"weather.device_a.temperature",
"weather.device_a.humidity",
@@ -951,14 +1012,20 @@ def
test_dataset_metadata_discovery_uses_all_numeric_fields(tmp_path):
assert list(tsdf["end_time"]) == [1, 1]
-def test_dataset_rejects_duplicate_timestamps_across_shards(tmp_path):
+def test_dataset_rejects_duplicate_timestamps_across_shards(
+ tmp_path, dataframe_use_index
+):
path1 = tmp_path / "part1.tsfile"
path2 = tmp_path / "part2.tsfile"
_write_weather_file(path1, 0)
_write_weather_file(path2, 2)
with pytest.raises(ValueError, match="Duplicate timestamp"):
- TsFileDataFrame([str(path1), str(path2)], show_progress=False)
+ TsFileDataFrame(
+ [str(path1), str(path2)],
+ show_progress=False,
+ use_index=dataframe_use_index,
+ )
def test_dataset_overlap_position_access_avoids_full_timestamp_materialization(
@@ -1009,7 +1076,7 @@ def
test_dataset_close_only_releases_current_handle(tmp_path):
path = tmp_path / "weather.tsfile"
_write_weather_file(path, 0)
- tsdf = TsFileDataFrame(str(path), show_progress=False)
+ tsdf = TsFileDataFrame(str(path), show_progress=False, use_index=True)
series = tsdf[0]
tsdf.close()
@@ -1026,7 +1093,7 @@ def
test_subset_close_releases_only_subset_lease(tmp_path):
path = tmp_path / "weather.tsfile"
_write_weather_file(path, 0)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(str(path), show_progress=False, use_index=True) as
tsdf:
subset = tsdf[:1]
subset.close()
@@ -1037,44 +1104,64 @@ def
test_subset_close_releases_only_subset_lease(tmp_path):
assert series[0] == 20.0
-def test_dataset_rejects_incompatible_table_schemas_across_shards(tmp_path):
+def test_dataset_rejects_incompatible_table_schemas_across_shards(
+ tmp_path, dataframe_use_index
+):
path1 = tmp_path / "part1.tsfile"
path2 = tmp_path / "part2.tsfile"
_write_weather_file(path1, 0)
_write_weather_with_extra_field_file(path2, 2)
with pytest.raises(ValueError, match="Incompatible schema for table
'weather'"):
- TsFileDataFrame([str(path1), str(path2)], show_progress=False)
+ TsFileDataFrame(
+ [str(path1), str(path2)],
+ show_progress=False,
+ use_index=dataframe_use_index,
+ )
-def test_dataset_rejects_same_field_name_with_different_type(tmp_path):
+def test_dataset_rejects_same_field_name_with_different_type(
+ tmp_path, dataframe_use_index
+):
path1 = tmp_path / "double.tsfile"
path2 = tmp_path / "int64.tsfile"
_write_weather_file(path1, 0)
_write_weather_int_temperature_file(path2, 3)
with pytest.raises(ValueError, match="Incompatible schema for table
'weather'"):
- TsFileDataFrame([str(path1), str(path2)], show_progress=False)
+ TsFileDataFrame(
+ [str(path1), str(path2)],
+ show_progress=False,
+ use_index=dataframe_use_index,
+ )
-def test_dataset_rejects_nonnumeric_declared_schema_difference(tmp_path):
+def test_dataset_rejects_nonnumeric_declared_schema_difference(
+ tmp_path, dataframe_use_index
+):
path1 = tmp_path / "text-status.tsfile"
path2 = tmp_path / "boolean-status.tsfile"
_write_numeric_and_text_file(path1)
_write_weather_boolean_status_file(path2, 3)
with pytest.raises(ValueError, match="Incompatible schema for table
'weather'"):
- TsFileDataFrame([str(path1), str(path2)], show_progress=False)
+ TsFileDataFrame(
+ [str(path1), str(path2)],
+ show_progress=False,
+ use_index=dataframe_use_index,
+ )
-def test_dataset_table_model_omits_boolean_fields(tmp_path):
+def test_dataset_table_model_omits_boolean_fields(tmp_path,
dataframe_use_index):
# The numeric dataset surface must drop BOOLEAN fields (they are not
# numeric) rather than surfacing them as a series that crashes the runtime
# read path. Regression for the reader/runtime field-type whitelist drift.
path = tmp_path / "weather_boolean.tsfile"
_write_weather_boolean_status_file(path, 0)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
assert tsdf.model == "table"
assert tsdf.list_timeseries() == ["weather.device_a.temperature"]
with pytest.raises(KeyError):
@@ -1103,7 +1190,7 @@ def
test_dataset_close_waits_for_an_active_public_query(tmp_path, monkeypatch):
blocked_read,
)
- dataframe = TsFileDataFrame(str(path), show_progress=False)
+ dataframe = TsFileDataFrame(str(path), show_progress=False, use_index=True)
query_done = threading.Event()
def run_query():
@@ -1137,14 +1224,16 @@ def
test_dataset_close_waits_for_an_active_public_query(tmp_path, monkeypatch):
assert query_error == []
-def test_dataset_skips_empty_tsfile_shards(tmp_path):
+def test_dataset_skips_empty_tsfile_shards(tmp_path, dataframe_use_index):
empty_path = tmp_path / "empty.tsfile"
data_path = tmp_path / "part.tsfile"
_write_empty_weather_file(empty_path)
_write_weather_file(data_path, 0)
with TsFileDataFrame(
- [str(empty_path), str(data_path)], show_progress=False
+ [str(empty_path), str(data_path)],
+ show_progress=False,
+ use_index=dataframe_use_index,
) as tsdf:
assert tsdf.list_timeseries() == [
"weather.device_a.temperature",
@@ -1164,11 +1253,13 @@ def test_reader_allows_empty_tsfile(tmp_path):
reader.close()
-def test_dataset_multi_tag_metadata_discovery(tmp_path):
+def test_dataset_multi_tag_metadata_discovery(tmp_path, dataframe_use_index):
path = tmp_path / "multi_tag.tsfile"
_write_multi_tag_file(path)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
assert tsdf.list_timeseries() == [
"weather.beijing.device_a.temperature",
"weather.beijing.device_a.humidity",
@@ -1218,11 +1309,13 @@ def test_dataset_multi_tag_metadata_discovery(tmp_path):
assert list(summary["count"]) == [2, 2, 2, 2]
-def test_dataset_series_paths_escape_special_tag_values(tmp_path):
+def test_dataset_series_paths_escape_special_tag_values(tmp_path,
dataframe_use_index):
path = tmp_path / "special_tag.tsfile"
_write_special_tag_file(path)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
expected_path = r"weather.bei\.jing.dev\\1.temperature"
assert tsdf.list_timeseries() == [expected_path]
@@ -1596,12 +1689,14 @@ def
test_dataframe_list_timeseries_filters_named_sparse_tag_prefix():
def test_dataframe_list_timeseries_prefix_can_skip_full_name_build(
- tmp_path, monkeypatch
+ tmp_path, monkeypatch, dataframe_use_index
):
path = tmp_path / "weather.tsfile"
_write_weather_file(path, 0)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
tsdf._index.series = [(0, 0)] * 1000
def fail_build_series_name(_series_ref):
@@ -1719,11 +1814,13 @@ def _write_tree_file(path):
writer.close()
-def test_dataset_tree_model_metadata_and_repr(tmp_path):
+def test_dataset_tree_model_metadata_and_repr(tmp_path, dataframe_use_index):
path = tmp_path / "tree.tsfile"
_write_tree_file(path)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
assert tsdf.model == "tree"
assert len(tsdf) == 3
assert sorted(tsdf.list_timeseries()) == [
@@ -1746,11 +1843,13 @@ def test_dataset_tree_model_metadata_and_repr(tmp_path):
tsdf["table"]
-def test_dataset_tree_model_series_access(tmp_path):
+def test_dataset_tree_model_series_access(tmp_path, dataframe_use_index):
path = tmp_path / "tree.tsfile"
_write_tree_file(path)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
ts = tsdf["root.ln.wf01.wt01.temperature"]
assert isinstance(ts, Timeseries)
assert ts.name == "root.ln.wf01.wt01.temperature"
@@ -1773,7 +1872,9 @@ 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):
+def test_dataset_tree_model_reads_uppercase_measurement_names(
+ tmp_path, dataframe_use_index
+):
"""Tree-model series with uppercase measurement names must read data.
Regression: a measurement like ``Temperature``/``STATUS`` must return its
@@ -1802,7 +1903,9 @@ def
test_dataset_tree_model_reads_uppercase_measurement_names(tmp_path):
)
writer.close()
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
assert sorted(tsdf.list_timeseries()) == [
"root.ln.wf01.wt01.STATUS",
"root.ln.wf01.wt01.Temperature",
@@ -1817,7 +1920,9 @@ def
test_dataset_tree_model_reads_uppercase_measurement_names(tmp_path):
)
-def
test_dataset_tree_model_case_distinct_measurements_do_not_collide(tmp_path,
capsys):
+def test_dataset_tree_model_case_distinct_measurements_do_not_collide(
+ tmp_path, capsys, dataframe_use_index
+):
"""Case-distinct measurements on one device must not be conflated.
``temperature`` and ``Temperature`` are two independent series; each must
@@ -1847,7 +1952,9 @@ def
test_dataset_tree_model_case_distinct_measurements_do_not_collide(tmp_path,
)
writer.close()
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
# Two independent tree-model series are discovered (case preserved).
assert len(tsdf) == 2
assert sorted(tsdf.list_timeseries()) == [
@@ -1909,7 +2016,9 @@ def
test_dataset_tree_model_case_distinct_measurements_do_not_collide(tmp_path,
)
-def test_tree_reader_handles_stale_path_columns_after_reused_queries(tmp_path):
+def test_tree_reader_handles_stale_path_columns_after_reused_queries(
+ tmp_path, dataframe_use_index
+):
"""Reusing a reader must not leak prefix path state across queries.
Reading one device series then another reuses the cached device id; stale
@@ -1918,7 +2027,9 @@ def
test_tree_reader_handles_stale_path_columns_after_reused_queries(tmp_path):
path = tmp_path / "tree.tsfile"
_write_tree_file(path)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
# First read establishes query state on the reader.
np.testing.assert_array_equal(
tsdf["root.ln.wf01.wt01.temperature"][:],
@@ -1936,11 +2047,13 @@ def
test_tree_reader_handles_stale_path_columns_after_reused_queries(tmp_path):
)
-def test_dataset_tree_model_list_timeseries_metadata(tmp_path):
+def test_dataset_tree_model_list_timeseries_metadata(tmp_path,
dataframe_use_index):
path = tmp_path / "tree.tsfile"
_write_tree_file(path)
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
meta = tsdf.list_timeseries_metadata()
assert isinstance(meta, pd.DataFrame)
@@ -1962,14 +2075,18 @@ def
test_dataset_tree_model_list_timeseries_metadata(tmp_path):
assert meta.loc["root.ln.wf02.wt02.status", "count"] == 5
-def test_dataset_rejects_mixed_model_load(tmp_path):
+def test_dataset_rejects_mixed_model_load(tmp_path, dataframe_use_index):
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)
+ TsFileDataFrame(
+ [str(table_path), str(tree_path)],
+ show_progress=False,
+ use_index=dataframe_use_index,
+ )
def _write_tree_rows(path, device_measurements, t_start=0, t_count=3):
@@ -1995,7 +2112,9 @@ def _write_tree_rows(path, device_measurements,
t_start=0, t_count=3):
writer.close()
-def test_dataset_tree_model_merges_identical_structure_across_files(tmp_path):
+def test_dataset_tree_model_merges_identical_structure_across_files(
+ tmp_path, dataframe_use_index
+):
# Two tree files, same device/measurement, disjoint time ranges: one
logical
# series whose shards and time bounds merge.
path1 = tmp_path / "t1.tsfile"
@@ -2003,7 +2122,11 @@ def
test_dataset_tree_model_merges_identical_structure_across_files(tmp_path):
_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:
+ with TsFileDataFrame(
+ [str(path1), str(path2)],
+ show_progress=False,
+ use_index=dataframe_use_index,
+ ) as tsdf:
assert tsdf.model == "tree"
assert tsdf.list_timeseries() == ["root.a.b.m1"]
ts = tsdf["root.a.b.m1"]
@@ -2015,14 +2138,18 @@ def
test_dataset_tree_model_merges_identical_structure_across_files(tmp_path):
assert meta.loc["root.a.b.m1", "count"] == 6
-def test_dataset_tree_model_unions_fields_across_files(tmp_path):
+def test_dataset_tree_model_unions_fields_across_files(tmp_path,
dataframe_use_index):
# 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:
+ with TsFileDataFrame(
+ [str(path1), str(path2)],
+ show_progress=False,
+ use_index=dataframe_use_index,
+ ) 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.
@@ -2036,7 +2163,9 @@ def
test_dataset_tree_model_unions_fields_across_files(tmp_path):
)
-def test_dataset_tree_model_unions_different_depths_across_files(tmp_path):
+def test_dataset_tree_model_unions_different_depths_across_files(
+ tmp_path, dataframe_use_index
+):
# Files with different max depth -> global tag layout widens; the shallower
# device pads its deepest tag column with null.
path1 = tmp_path / "t1.tsfile"
@@ -2044,7 +2173,11 @@ def
test_dataset_tree_model_unions_different_depths_across_files(tmp_path):
_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:
+ with TsFileDataFrame(
+ [str(path1), str(path2)],
+ show_progress=False,
+ use_index=dataframe_use_index,
+ ) 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()
@@ -2060,7 +2193,9 @@ def
test_dataset_tree_model_unions_different_depths_across_files(tmp_path):
)
-def test_dataset_tree_model_omits_non_numeric_measurements(tmp_path):
+def test_dataset_tree_model_omits_non_numeric_measurements(
+ tmp_path, dataframe_use_index
+):
# 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
@@ -2084,7 +2219,9 @@ def
test_dataset_tree_model_omits_non_numeric_measurements(tmp_path):
)
writer.close()
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) 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):
@@ -2094,7 +2231,9 @@ def
test_dataset_tree_model_omits_non_numeric_measurements(tmp_path):
)
-def test_dataset_tree_model_loc_aligns_sparse_non_aligned_fields(tmp_path):
+def test_dataset_tree_model_loc_aligns_sparse_non_aligned_fields(
+ tmp_path, dataframe_use_index
+):
# Non-aligned (tree) device whose two measurements are sampled at
# different timestamps must produce a timestamp union with NaN fill,
# instead of raising when the per-field timelines differ. Regression for
@@ -2122,7 +2261,9 @@ def
test_dataset_tree_model_loc_aligns_sparse_non_aligned_fields(tmp_path):
)
writer.close()
- with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ with TsFileDataFrame(
+ str(path), show_progress=False, use_index=dataframe_use_index
+ ) as tsdf:
assert tsdf.model == "tree"
aligned = tsdf.loc[0:5, ["root.a.b.m1", "root.a.b.m2"]]
assert isinstance(aligned, AlignedTimeseries)
diff --git a/python/tsfile/dataset/dataframe.py
b/python/tsfile/dataset/dataframe.py
index e3ce78372..9e706b0f2 100644
--- a/python/tsfile/dataset/dataframe.py
+++ b/python/tsfile/dataset/dataframe.py
@@ -663,8 +663,28 @@ class _LocIndexer:
_, table_entry, _ = self._df._get_series_components(series_ref)
field_name = table_entry.field_columns[field_idx]
- descriptor = self._df._index.series_shards.describe(series_ref)
- for shard in descriptor.shards:
+ describe = getattr(self._df._index.series_shards, "describe", None)
+ if describe is None:
+ shard_entries = []
+ for (
+ reader,
+ device_id,
+ reader_field_idx,
+ ) in self._df._index.series_shards[series_ref]:
+ info = reader.get_series_info_by_ref(device_id,
reader_field_idx)
+ shard_entries.append(
+ SimpleNamespace(
+ reader=reader,
+ device_id=device_id,
+ column_id=reader_field_idx,
+ min_time=info["timeline_min_time"],
+ max_time=info["timeline_max_time"],
+ timeline_length=info["timeline_length"],
+ )
+ )
+ else:
+ shard_entries = describe(series_ref).shards
+ for shard in shard_entries:
overlap_start = max(start_time, shard.min_time)
overlap_end = min(end_time, shard.max_time)
if shard.timeline_length <= 0 or overlap_start > overlap_end:
@@ -710,13 +730,16 @@ class _LocIndexer:
return entries, ts_arr, field_vals
group_entries = list(groups.values())
- group_results = self._df._runtime.map_query_groups(
- query_group,
- group_entries,
- estimated_rows=[
- max(entry[6] for entry in entries) for entries in group_entries
- ],
- )
+ if self._df._runtime is None:
+ group_results = [query_group(entries) for entries in group_entries]
+ else:
+ group_results = self._df._runtime.map_query_groups(
+ query_group,
+ group_entries,
+ estimated_rows=[
+ max(entry[6] for entry in entries) for entries in
group_entries
+ ],
+ )
series_time_parts = defaultdict(list)
series_value_parts = defaultdict(list)
@@ -751,9 +774,17 @@ class _LocIndexer:
class TsFileDataFrame:
"""Lazy-loaded unified numeric dataset view over multiple TsFile shards."""
- def __init__(self, paths: Union[str, List[str]], show_progress: bool =
True):
+ def __init__(
+ self,
+ paths: Union[str, List[str]],
+ show_progress: bool = True,
+ use_index: bool = False,
+ ):
+ if not isinstance(use_index, bool):
+ raise TypeError("use_index must be a bool")
self._paths = _expand_paths(paths)
self._show_progress = show_progress
+ self._use_index = use_index
self._readers: Dict[str, object] = {}
self._index = _DataFrameCatalog()
self._is_view = False
@@ -773,6 +804,7 @@ class TsFileDataFrame:
obj._is_view = True
obj._paths = parent._paths
obj._show_progress = parent._show_progress
+ obj._use_index = parent._use_index
obj._readers = parent._readers
subset_refs = list(series_refs)
obj._index = SimpleNamespace(
@@ -808,9 +840,24 @@ class TsFileDataFrame:
with self._runtime_lease.query_lease():
yield
+ def _load_metadata_without_index(self, reader_class):
+ if len(self._paths) >= 2:
+ self._load_metadata_parallel(reader_class)
+ else:
+ self._load_metadata_serial(reader_class)
+
+ if not self._index.series:
+ raise ValueError("No valid time series found in the provided
TsFile files")
+ _validate_unique_shard_timestamps(self._index)
+
def _load_metadata(self):
- """Map a valid persistent index, or build it once under a file lock."""
+ """Load metadata, optionally through a persistent mmap-backed index."""
from .reader import TsFileSeriesReader
+
+ if not self._use_index:
+ self._load_metadata_without_index(TsFileSeriesReader)
+ return
+
from .index import (
build_index_from_dataframe,
index_matches_paths,
@@ -830,17 +877,8 @@ class TsFileDataFrame:
except ImportError:
pass
if not index_matches_paths(index_path, self._paths):
- if len(self._paths) >= 2:
- self._load_metadata_parallel(TsFileSeriesReader)
- else:
- self._load_metadata_serial(TsFileSeriesReader)
-
- if not self._index.series:
- raise ValueError(
- "No valid time series found in the provided TsFile
files"
- )
+ self._load_metadata_without_index(TsFileSeriesReader)
try:
- _validate_unique_shard_timestamps(self._index)
build_index_from_dataframe(self, index_path)
finally:
for reader in self._readers.values():
diff --git a/python/tsfile/dataset/index.py b/python/tsfile/dataset/index.py
index 394860d54..4bd9c3767 100644
--- a/python/tsfile/dataset/index.py
+++ b/python/tsfile/dataset/index.py
@@ -28,6 +28,7 @@ from __future__ import annotations
from collections import defaultdict
import contextlib
+import hashlib
import mmap
import os
import struct
@@ -43,7 +44,8 @@ HEADER_SIZE = 64
DIRECTORY_ENTRY_SIZE = 32
SECTION_COUNT = 13
ALIGNMENT = 64
-INDEX_FILE_NAME = ".tsfile_dataframe_index.tsidx"
+INDEX_FILE_PREFIX = ".tsfile_dataframe_index"
+INDEX_FILE_SUFFIX = ".tsidx"
STRING_OFFSETS = 1
STRING_BYTES = 2
@@ -460,10 +462,18 @@ class MappedDatasetIndex:
def index_path_for(paths: Sequence[str]) -> str:
- common = os.path.commonpath([os.path.abspath(path) for path in paths])
+ canonical_paths = sorted(os.path.abspath(path) for path in paths)
+ common = os.path.commonpath(canonical_paths)
if not os.path.isdir(common):
common = os.path.dirname(common)
- return os.path.join(common, INDEX_FILE_NAME)
+ digest = hashlib.sha256()
+ for path in canonical_paths:
+ digest.update(path.encode("utf-8"))
+ digest.update(b"\0")
+ return os.path.join(
+ common,
+ f"{INDEX_FILE_PREFIX}.{digest.hexdigest()[:32]}{INDEX_FILE_SUFFIX}",
+ )
def index_matches_paths(path: str, paths: Sequence[str]) -> bool:
