You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tsfile.apache.org by co...@apache.org on 2026/05/31 05:22:09 UTC
(tsfile) branch colin_tsfile_dataframe_init updated: read_series_by_row: pull TsBlocks via Arrow instead of row-by-row Python loop
This is an automated email from the ASF dual-hosted git repository.
ColinLeeo pushed a commit to branch colin_tsfile_dataframe_init
in repository https://gitbox.apache.org/repos/asf/tsfile.git
The following commit(s) were added to refs/heads/colin_tsfile_dataframe_init by this push:
new 011216880 read_series_by_row: pull TsBlocks via Arrow instead of row-by-row Python loop
011216880 is described below
commit 011216880984db1d08c9f445048570d403450b27
Author: ColinLee <sh...@163.com>
AuthorDate: Sun May 31 12:38:46 2026 +0800
read_series_by_row: pull TsBlocks via Arrow instead of row-by-row Python loop
`series[s:e]` on a TsFileDataFrame fans out to TsFileSeriesReader.
read_series_by_row, which was issuing query_table_by_row in
RETURN_ROW mode and then calling result_set.next() +
get_value_by_name() per row. Every iteration is a Python<->C
round-trip plus two list.append() calls, so a 10k-row slice costs
~30k transitions and dominates wall time -- it's actually slower
than the pre-optimization scalar path because the batch/parallel
work paid for by the snapshot only kicks in when batch_size>0,
which this call site never set.
Pass batch_size=65536 instead and drain the result set via
read_arrow_batch(), pushing each TsBlock straight into numpy via
the existing Arrow C-Data interface (mirrors what
_read_arrow / read_device_fields_by_time_range already does).
The outer while-remaining retry loop is kept so we can re-issue
from the advanced offset if the native call stops at an internal
block boundary before the requested window is filled.
The boundary-retry unit test is updated to mock read_arrow_batch
(returning a pyarrow.Table per call) and to assert batch_size>0
is actually plumbed through, so we don't silently regress to the
slow path again.
152 Python tests pass.
---
python/tests/test_tsfile_dataset.py | 51 ++++++++++++++++++++++---------------
python/tsfile/dataset/reader.py | 39 ++++++++++++++++------------
2 files changed, 54 insertions(+), 36 deletions(-)
diff --git a/python/tests/test_tsfile_dataset.py b/python/tests/test_tsfile_dataset.py
index f79a6d466..74ad86952 100644
--- a/python/tests/test_tsfile_dataset.py
+++ b/python/tests/test_tsfile_dataset.py
@@ -688,10 +688,19 @@ def test_reader_catalog_shares_device_metadata_and_resolves_paths(tmp_path):
def test_reader_read_series_by_row_retries_across_native_row_query_boundaries():
+ """read_series_by_row pulls TsBlocks via read_arrow_batch and must keep
+ re-issuing query_table_by_row when the underlying native call stops at
+ an internal block boundary before the caller's window is filled."""
+
+ import pyarrow as pa
+
class _FakeResultSet:
- def __init__(self, rows):
- self._rows = rows
- self._index = -1
+ def __init__(self, times, values):
+ self._batch = pa.table(
+ {"time": pa.array(times, type=pa.int64()),
+ "totalcloudcover": pa.array(values, type=pa.float64())}
+ )
+ self._delivered = False
def __enter__(self):
return self
@@ -699,12 +708,11 @@ def test_reader_read_series_by_row_retries_across_native_row_query_boundaries():
def __exit__(self, exc_type, exc_val, exc_tb):
return False
- def next(self):
- self._index += 1
- return self._index < len(self._rows)
-
- def get_value_by_name(self, name):
- return self._rows[self._index][name]
+ def read_arrow_batch(self):
+ if self._delivered or self._batch.num_rows == 0:
+ return None
+ self._delivered = True
+ return self._batch
class _FakeNativeReader:
def __init__(self, timestamps, values, boundary):
@@ -713,28 +721,31 @@ def test_reader_read_series_by_row_retries_across_native_row_query_boundaries():
self._boundary = boundary
def query_table_by_row(
- self, table_name, column_names, offset=0, limit=-1, tag_filter=None
+ self,
+ table_name,
+ column_names,
+ offset=0,
+ limit=-1,
+ tag_filter=None,
+ batch_size=0,
):
assert table_name == "pvf"
assert column_names == ["totalcloudcover"]
assert tag_filter is None
+ assert batch_size > 0, "row reads should use batch (Arrow) mode"
if limit < 0:
stop = len(self._timestamps)
else:
stop = min(offset + limit, len(self._timestamps))
- # Simulate the current native bug: one row query cannot cross the
- # next internal boundary, so callers must re-issue from the
+ # Simulate the native quirk where one query stops at the next
+ # internal block boundary; callers must re-issue from the
# advanced offset to complete a large logical window.
chunk_stop = min(stop, ((offset // self._boundary) + 1) * self._boundary)
- rows = [
- {
- "time": int(self._timestamps[idx]),
- "totalcloudcover": float(self._values[idx]),
- }
- for idx in range(offset, chunk_stop)
- ]
- return _FakeResultSet(rows)
+ return _FakeResultSet(
+ self._timestamps[offset:chunk_stop],
+ self._values[offset:chunk_stop],
+ )
reader = object.__new__(TsFileSeriesReader)
reader._reader = _FakeNativeReader(
diff --git a/python/tsfile/dataset/reader.py b/python/tsfile/dataset/reader.py
index ca435b78c..bcf977831 100644
--- a/python/tsfile/dataset/reader.py
+++ b/python/tsfile/dataset/reader.py
@@ -374,37 +374,44 @@ class TsFileSeriesReader:
tag_values = dict(zip(table_entry.tag_columns, device_entry.tag_values))
tag_filter = _build_exact_tag_filter(tag_values) if tag_values else None
- # Some native row-query paths stop at an internal block boundary even
- # when the requested window extends further. Re-issue from the advanced
- # offset until we fill the caller's logical row window or reach EOF.
+ # Pull whole TsBlocks via the Arrow C-Data interface instead of
+ # iterating row-by-row in Python. Each result_set.next() +
+ # get_value_by_name() pair would be a Python<->C round-trip per row
+ # and dominates wall time on long slices; read_arrow_batch() returns
+ # a column-oriented batch in one call and lands directly in numpy.
timestamp_parts = []
value_parts = []
remaining = limit
next_offset = offset
while remaining > 0:
- batch_timestamps = []
- batch_values = []
+ produced_this_call = 0
with self._reader.query_table_by_row(
table_entry.table_name,
[field_name],
offset=next_offset,
limit=remaining,
tag_filter=tag_filter,
+ batch_size=65536,
) as result_set:
- while result_set.next():
- batch_timestamps.append(result_set.get_value_by_name("time"))
- value = result_set.get_value_by_name(field_name)
- batch_values.append(np.nan if value is None else float(value))
-
- if not batch_timestamps:
+ while True:
+ arrow_table = result_set.read_arrow_batch()
+ if arrow_table is None:
+ break
+ if arrow_table.num_rows == 0:
+ continue
+ timestamp_parts.append(arrow_table.column("time").to_numpy())
+ raw_values = arrow_table.column(field_name).to_numpy(
+ zero_copy_only=False
+ )
+ value_parts.append(np.asarray(raw_values, dtype=np.float64))
+ produced_this_call += arrow_table.num_rows
+
+ if produced_this_call == 0:
break
- timestamp_parts.append(np.asarray(batch_timestamps, dtype=np.int64))
- value_parts.append(np.asarray(batch_values, dtype=np.float64))
- read_count = len(batch_timestamps)
- next_offset += read_count
- remaining -= read_count
+ next_offset += produced_this_call
+ remaining -= produced_this_call
if not timestamp_parts:
return np.array([], dtype=np.int64), np.array([], dtype=np.float64)