From 78b67ef3f82942a97a119dfca0ce77c01fcd356f Mon Sep 17 00:00:00 2001 From: ColinLee Date: Tue, 8 Sep 2026 14:27:50 +0800 Subject: [PATCH 1/2] fix(python): preserve shared readers when closing dataframe subsets --- python/tests/test_tsfile_dataset.py | 119 ++++++++++++++++++++++++++++ python/tsfile/dataset/dataframe.py | 8 +- 2 files changed, 126 insertions(+), 1 deletion(-) diff --git a/python/tests/test_tsfile_dataset.py b/python/tests/test_tsfile_dataset.py index a5d7e9e57..fd0aaab2b 100644 --- a/python/tests/test_tsfile_dataset.py +++ b/python/tests/test_tsfile_dataset.py @@ -19,6 +19,9 @@ import numpy as np import pandas as pd import pytest +import subprocess +import sys +import textwrap import threading from tsfile.dataset import dataframe as dataframe_module @@ -1149,6 +1152,122 @@ def test_subset_close_releases_only_subset_lease(tmp_path): assert series[0] == 20.0 +@pytest.fixture(params=["tree", "table"]) +def subset_lifecycle_dataset(tmp_path, request): + path = tmp_path / "lifecycle.tsfile" + if request.param == "tree": + _write_tree_rows(path, {"root.name.lower": [("value", TSDataType.DOUBLE)]}) + return str(path), "root.name.lower.value", [0.5, 1.5, 2.5] + _write_weather_file(path, 0) + return str(path), "weather.device_a.temperature", [20.0, 21.5, 23.0] + + +@pytest.mark.parametrize("release", ["close", "collect", "context", "filter"]) +def test_subset_release_preserves_shared_readers( + subset_lifecycle_dataset, dataframe_use_index, release +): + path, name, expected = subset_lifecycle_dataset + # A regression here can dereference a closed native reader. Keep the + # assertions in a child process so it cannot terminate the pytest suite. + result = subprocess.run( + [ + sys.executable, + "-X", + "faulthandler", + "-c", + textwrap.dedent(""" + import gc + import json + import sys + import numpy as np + import pytest + from tsfile import TsFileDataFrame + + path, name, expected, use_index, release = sys.argv[1:] + expected = json.loads(expected) + with TsFileDataFrame(path, show_progress=False, + use_index=use_index == "True") as root: + index = root.list_timeseries().index(name) + sibling = root[[index]] + parent = root[[index]] + nested = parent[[0]] + series = root[name] + if release == "close": + parent.close() + parent.close() + with pytest.raises(RuntimeError, match="closed"): + parent.loc[:, [name]] + elif release == "collect": + del parent + gc.collect() + elif release == "context": + with parent: + np.testing.assert_allclose(parent[name][:], expected) + else: + field = root.list_timeseries_metadata().loc[name, "field"] + assert name in root[root["field"] == field].list_timeseries() + gc.collect() + np.testing.assert_allclose(series[:], expected) + for frame in (root, sibling, nested): + aligned = frame.loc[:, [name, name]] + np.testing.assert_array_equal(aligned.timestamps, [0, 1, 2]) + np.testing.assert_allclose(aligned.values, + np.column_stack([expected, expected])) + np.testing.assert_allclose(frame[name][:], expected) + sibling.close() + nested.close() + """), + path, + name, + str(expected), + str(dataframe_use_index), + release, + ], + capture_output=True, + text=True, + timeout=30, + ) + assert result.returncode == 0, result.stdout + result.stderr + + +def test_no_index_subset_rejects_reads_after_root_close(subset_lifecycle_dataset): + path, name, _ = subset_lifecycle_dataset + result = subprocess.run( + [ + sys.executable, + "-X", + "faulthandler", + "-c", + textwrap.dedent(""" + import sys + import pytest + from tsfile import TsFileDataFrame + + root = TsFileDataFrame(sys.argv[1], show_progress=False) + subset = root[[root.list_timeseries().index(sys.argv[2])]] + nested = subset[[0]] + series = nested[0] + root.close() + for frame in (root, subset, nested): + with pytest.raises(RuntimeError, match="closed"): + frame.loc[:, [0]] + with pytest.raises(RuntimeError, match="closed"): + frame[0][:] + with pytest.raises(RuntimeError, match="closed"): + series[:] + subset.close() + nested.close() + """), + path, + name, + ], + capture_output=True, + text=True, + timeout=30, + ) + assert result.returncode == 0, result.stdout + result.stderr + + def test_dataset_rejects_incompatible_table_schemas_across_shards( tmp_path, dataframe_use_index ): diff --git a/python/tsfile/dataset/dataframe.py b/python/tsfile/dataset/dataframe.py index 9e706b0f2..7ccdc68f9 100644 --- a/python/tsfile/dataset/dataframe.py +++ b/python/tsfile/dataset/dataframe.py @@ -830,6 +830,10 @@ def _owner(self) -> "TsFileDataFrame": def _assert_open(self): if self._closed: raise RuntimeError("Current TsFileDataFrame is closed.") + if self._is_view and self._runtime_lease is None: + # Direct readers belong to the root; a view cannot outlive an + # explicit root close. Indexed views have independent leases. + self._root._assert_open() @contextlib.contextmanager def _query_guard(self): @@ -1416,7 +1420,9 @@ def close(self): self._closed = True if self._runtime_lease is not None: self._runtime_lease.close() - else: + elif not self._is_view: + # Subsets share these readers and their dictionary with the root. + # Closing or collecting a view must not release the root's readers. for reader in self._readers.values(): reader.close() self._readers.clear() From b1427cfed2a619eef17d778446bee9c4740fb354 Mon Sep 17 00:00:00 2001 From: ColinLee Date: Tue, 8 Sep 2026 15:01:45 +0800 Subject: [PATCH 2/2] style(python): keep lifecycle tests compatible with Black 25 and 26 --- python/tests/test_tsfile_dataset.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/python/tests/test_tsfile_dataset.py b/python/tests/test_tsfile_dataset.py index fd0aaab2b..02baa9bec 100644 --- a/python/tests/test_tsfile_dataset.py +++ b/python/tests/test_tsfile_dataset.py @@ -1175,7 +1175,8 @@ def test_subset_release_preserves_shared_readers( "-X", "faulthandler", "-c", - textwrap.dedent(""" + textwrap.dedent( + """ import gc import json import sys @@ -1216,7 +1217,8 @@ def test_subset_release_preserves_shared_readers( np.testing.assert_allclose(frame[name][:], expected) sibling.close() nested.close() - """), + """, + ), path, name, str(expected), @@ -1238,7 +1240,8 @@ def test_no_index_subset_rejects_reads_after_root_close(subset_lifecycle_dataset "-X", "faulthandler", "-c", - textwrap.dedent(""" + textwrap.dedent( + """ import sys import pytest from tsfile import TsFileDataFrame @@ -1257,7 +1260,8 @@ def test_no_index_subset_rejects_reads_after_root_close(subset_lifecycle_dataset series[:] subset.close() nested.close() - """), + """, + ), path, name, ],