fix typo and address close problem

Author: ycycse

python/tests/test_tsfile_dataset.py

@@ -264,6 +264,34 @@ def test_dataset_rejects_duplicate_timestamps_across_shards(tmp_path):
         TsFileDataFrame([str(path1), str(path2)], show_progress=False)
 
 
+def test_dataset_rejects_data_access_after_close(tmp_path):
+    path = tmp_path / "weather.tsfile"
+    _write_weather_file(path, 0)
+
+    tsdf = TsFileDataFrame(str(path), show_progress=False)
+    series = tsdf[0]
+    tsdf.close()
+
+    with pytest.raises(RuntimeError, match="TsFileDataFrame is closed"):
+        _ = tsdf[0]
+
+    with pytest.raises(RuntimeError, match="TsFileDataFrame is closed"):
+        _ = series[0]
+
+
+def test_subset_close_warns_and_does_not_close_root(tmp_path):
+    path = tmp_path / "weather.tsfile"
+    _write_weather_file(path, 0)
+
+    with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+        subset = tsdf[:1]
+        with pytest.warns(RuntimeWarning, match="no-op"):
+            subset.close()
+
+        series = tsdf[0]
+        assert series[0] == 20.0
+
+
 def test_dataset_rejects_incompatible_table_schemas_across_shards(tmp_path):
     path1 = tmp_path / "part1.tsfile"
     path2 = tmp_path / "part2.tsfile"

python/tsfile/dataset/dataframe.py

@@ -23,6 +23,7 @@
 import os
 import sys
 from typing import Dict, List, Set, Tuple, Union
+import warnings
 
 import numpy as np
 
@@ -271,6 +272,7 @@ def _parse_key(self, key):
 
     def _query_aligned(self, start_time: int, end_time: int, series_refs: List[SeriesRefKey], series_names: List[str]):
         """Batch aligned reads by reader/device, then merge per-series fragments."""
+        self._df._assert_open()
         groups = defaultdict(list)
         for col_idx, series_ref in enumerate(series_refs):
             device_idx, field_idx = series_ref
@@ -320,6 +322,7 @@ def __init__(self, paths: Union[str, List[str]], show_progress: bool = True):
         self._cache = _DerivedCache()
         self._is_view = False
         self._root = None
+        self._closed = False
         self._load_metadata()
 
     @classmethod
@@ -341,8 +344,16 @@ def _from_subset(cls, parent: "TsFileDataFrame", series_refs: List[SeriesRefKey]
             series_ref_set=set(series_refs),
         )
         obj._cache = _DerivedCache(devices=parent._cache.devices, field_stats=parent._cache.field_stats)
+        obj._closed = False
         return obj
 
+    def _owner(self) -> "TsFileDataFrame":
+        return self._root if self._is_view else self
+
+    def _assert_open(self):
+        if self._owner()._closed:
+            raise RuntimeError("Current TsFileDataFrame is closed.")
+
     def _load_metadata(self):
         """Build the logical cross-file index and the derived per-series caches."""
         from .reader import TsFileSeriesReader
@@ -474,11 +485,13 @@ def list_timeseries(self, path_prefix: str = "") -> List[str]:
         return [name for name in names if name.startswith(prefix) or name == path_prefix]
 
     def _get_timeseries(self, series_ref: SeriesRefKey) -> Timeseries:
+        self._assert_open()
         series_name = self._build_series_name(series_ref)
         return Timeseries(
             series_name,
             self._index.series_ref_map[series_ref],
             self._cache.field_stats[series_ref],
+            self._assert_open,
             lambda: _merge_field_timestamps(series_name, self._index.series_ref_map[series_ref]),
         )
 
@@ -595,10 +608,18 @@ def show(self, max_rows: int = 20):
 
     def close(self):
         if self._is_view:
+            warnings.warn(
+                "close() on a subset TsFileDataFrame is a no-op; only the root dataframe owns the readers.",
+                RuntimeWarning,
+                stacklevel=2,
+            )
+            return
+        if self._closed:
             return
         for reader in self._readers.values():
             reader.close()
         self._readers.clear()
+        self._closed = True
 
     def __del__(self):
         try:

python/tsfile/dataset/metadata.py

@@ -51,7 +51,10 @@ def get_field_index(self, field_name: str) -> int:
 
 @dataclass(slots=True)
 class DeviceEntry:
-    """One logical device identified by table name + ordered tag values."""
+    """One logical device identified by table_id + ordered tag values.
+
+    The table_id refers to MetadataCatalog.table_entries[table_id].
+    """
 
     table_id: int
     tag_values: Tuple[Any, ...]

python/tsfile/dataset/timeseries.py

@@ -68,11 +68,13 @@ def __init__(
         name: str,
         series_refs: list,
         stats: dict,
+        ensure_open: Callable[[], None],
         load_timestamps: Callable[[], np.ndarray],
     ):
         self._name = name
         self._series_refs = series_refs
         self._stats = dict(stats)
+        self._ensure_open = ensure_open
         self._load_timestamps = load_timestamps
         self._timestamps = None
 
@@ -82,6 +84,7 @@ def name(self) -> str:
 
     @property
     def timestamps(self) -> np.ndarray:
+        self._ensure_open()
         if self._timestamps is None:
             self._timestamps = self._load_timestamps()
         return self._timestamps
@@ -128,6 +131,7 @@ def __getitem__(self, key):
         raise TypeError(f"Unsupported key type: {type(key)}")
 
     def _query_time_range(self, start_time: int, end_time: int) -> Tuple[np.ndarray, np.ndarray]:
+        self._ensure_open()
         time_parts = []
         value_parts = []
         for reader, device_id, field_idx in self._series_refs: