# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#

import numpy as np
import pandas as pd
import pytest

from tsfile.dataset import dataframe as dataframe_module
from tsfile import ColumnCategory, ColumnSchema, TSDataType, TableSchema, TsFileTableWriter
from tsfile import AlignedTimeseries, Timeseries, TsFileDataFrame
from tsfile.dataset.formatting import format_timestamp
from tsfile.dataset.metadata import MetadataCatalog, build_series_path, resolve_series_path
from tsfile.dataset.reader import TsFileSeriesReader, _build_exact_tag_filter


def _write_weather_file(path, start):
    schema = TableSchema(
        "weather",
        [
            ColumnSchema("device", TSDataType.STRING, ColumnCategory.TAG),
            ColumnSchema("temperature", TSDataType.DOUBLE, ColumnCategory.FIELD),
            ColumnSchema("humidity", TSDataType.DOUBLE, ColumnCategory.FIELD),
        ],
    )
    df = pd.DataFrame(
        {
            "time": [start, start + 1, start + 2],
            "device": ["device_a", "device_a", "device_a"],
            "temperature": [20.0, 21.5, 23.0],
            "humidity": [50.0, 52.0, 55.0],
        }
    )
    with TsFileTableWriter(str(path), schema) as writer:
        writer.write_dataframe(df)


def _write_weather_rows_file(path, rows):
    schema = TableSchema(
        "weather",
        [
            ColumnSchema("device", TSDataType.STRING, ColumnCategory.TAG),
            ColumnSchema("temperature", TSDataType.DOUBLE, ColumnCategory.FIELD),
            ColumnSchema("humidity", TSDataType.DOUBLE, ColumnCategory.FIELD),
        ],
    )
    df = pd.DataFrame(rows)
    with TsFileTableWriter(str(path), schema) as writer:
        writer.write_dataframe(df)


def _write_empty_weather_file(path):
    schema = TableSchema(
        "weather",
        [
            ColumnSchema("device", TSDataType.STRING, ColumnCategory.TAG),
            ColumnSchema("temperature", TSDataType.DOUBLE, ColumnCategory.FIELD),
            ColumnSchema("humidity", TSDataType.DOUBLE, ColumnCategory.FIELD),
        ],
    )
    with TsFileTableWriter(str(path), schema):
        pass


def _write_numeric_and_text_file(path):
    schema = TableSchema(
        "weather",
        [
            ColumnSchema("device", TSDataType.STRING, ColumnCategory.TAG),
            ColumnSchema("temperature", TSDataType.DOUBLE, ColumnCategory.FIELD),
            ColumnSchema("status", TSDataType.STRING, ColumnCategory.FIELD),
        ],
    )
    df = pd.DataFrame(
        {
            "time": [0, 1, 2],
            "device": ["device_a", "device_a", "device_a"],
            "temperature": [20.0, np.nan, 23.5],
            "status": ["ok", "warn", "ok"],
        }
    )
    with TsFileTableWriter(str(path), schema) as writer:
        writer.write_dataframe(df)


def _write_partial_numeric_rows_file(path):
    schema = TableSchema(
        "weather",
        [
            ColumnSchema("device", TSDataType.STRING, ColumnCategory.TAG),
            ColumnSchema("temperature", TSDataType.DOUBLE, ColumnCategory.FIELD),
            ColumnSchema("humidity", TSDataType.DOUBLE, ColumnCategory.FIELD),
        ],
    )
    df = pd.DataFrame(
        {
            "time": [0, 1],
            "device": ["device_a", "device_a"],
            "temperature": [np.nan, 21.0],
            "humidity": [50.0, 51.0],
        }
    )
    with TsFileTableWriter(str(path), schema) as writer:
        writer.write_dataframe(df)


def _write_weather_with_extra_field_file(path, start):
    schema = TableSchema(
        "weather",
        [
            ColumnSchema("device", TSDataType.STRING, ColumnCategory.TAG),
            ColumnSchema("temperature", TSDataType.DOUBLE, ColumnCategory.FIELD),
            ColumnSchema("humidity", TSDataType.DOUBLE, ColumnCategory.FIELD),
            ColumnSchema("pressure", TSDataType.DOUBLE, ColumnCategory.FIELD),
        ],
    )
    df = pd.DataFrame(
        {
            "time": [start, start + 1],
            "device": ["device_a", "device_a"],
            "temperature": [20.0, 21.0],
            "humidity": [50.0, 51.0],
            "pressure": [1000.0, 1001.0],
        }
    )
    with TsFileTableWriter(str(path), schema) as writer:
        writer.write_dataframe(df)


def _write_multi_tag_file(path):
    schema = TableSchema(
        "weather",
        [
            ColumnSchema("city", TSDataType.STRING, ColumnCategory.TAG),
            ColumnSchema("device", TSDataType.STRING, ColumnCategory.TAG),
            ColumnSchema("temperature", TSDataType.DOUBLE, ColumnCategory.FIELD),
            ColumnSchema("humidity", TSDataType.DOUBLE, ColumnCategory.FIELD),
            ColumnSchema("status", TSDataType.STRING, ColumnCategory.FIELD),
        ],
    )
    df = pd.DataFrame(
        {
            "time": [0, 1, 0, 1],
            "city": ["beijing", "beijing", "shanghai", "shanghai"],
            "device": ["device_a", "device_a", "device_b", "device_b"],
            "temperature": [20.0, 21.0, 24.0, 25.0],
            "humidity": [50.0, 51.0, 60.0, 61.0],
            "status": ["ok", "ok", "warn", "warn"],
        }
    )
    with TsFileTableWriter(str(path), schema) as writer:
        writer.write_dataframe(df)


def _write_special_tag_file(path):
    schema = TableSchema(
        "weather",
        [
            ColumnSchema("city", TSDataType.STRING, ColumnCategory.TAG),
            ColumnSchema("device", TSDataType.STRING, ColumnCategory.TAG),
            ColumnSchema("temperature", TSDataType.DOUBLE, ColumnCategory.FIELD),
        ],
    )
    df = pd.DataFrame(
        {
            "time": [0, 1],
            "city": ["bei.jing", "bei.jing"],
            "device": [r"dev\1", r"dev\1"],
            "temperature": [20.0, 21.0],
        }
    )
    with TsFileTableWriter(str(path), schema) as writer:
        writer.write_dataframe(df)


def test_dataset_top_level_imports():
    assert TsFileDataFrame.__module__ == "tsfile.dataset.dataframe"
    assert Timeseries.__module__ == "tsfile.dataset.timeseries"
    assert AlignedTimeseries.__module__ == "tsfile.dataset.timeseries"


def test_format_timestamp_preserves_millisecond_precision():
    assert "." not in format_timestamp(1000)
    assert format_timestamp(1).endswith(".001")


def test_dataset_basic_access_patterns(tmp_path, capsys):
    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:
        assert len(tsdf) == 2

        first = tsdf[0]
        assert isinstance(first, Timeseries)
        assert first.name in tsdf.list_timeseries()
        assert len(first) == 6
        assert first[0] == 20.0
        assert first[-1] == 23.0
        assert "Timeseries(" in repr(first)

        by_name = tsdf[first.name]
        assert isinstance(by_name, Timeseries)
        assert by_name.name == first.name

        subset = tsdf[:1]
        assert isinstance(subset, TsFileDataFrame)
        assert len(subset) == 1

        selected = tsdf[[0, 1]]
        assert isinstance(selected, TsFileDataFrame)
        assert len(selected) == 2

        aligned = tsdf.loc[0:5, [0, 1]]
        assert isinstance(aligned, AlignedTimeseries)
        assert aligned.shape == (6, 2)

        aligned_negative = tsdf.loc[0:5, [-1]]
        assert isinstance(aligned_negative, AlignedTimeseries)
        assert aligned_negative.shape == (6, 1)

        assert list(tsdf["field"]) == ["temperature", "humidity"]

        assert "TsFileDataFrame(2 time series, 2 files)" in repr(tsdf)
        aligned.show(2)
        assert "AlignedTimeseries(6 rows, 2 series)" in capsys.readouterr().out


def test_dataset_loc_aligns_timestamp_union_and_preserves_requested_order(tmp_path):
    path = tmp_path / "weather_sparse.tsfile"
    _write_weather_rows_file(
        path,
        {
            "time": [0, 1, 2],
            "device": ["device_a", "device_a", "device_a"],
            "temperature": [10.0, np.nan, 30.0],
            "humidity": [np.nan, 200.0, 300.0],
        },
    )

    with TsFileDataFrame(str(path), show_progress=False) as tsdf:
        aligned = tsdf.loc[
            0:2,
            [
                "weather.device_a.humidity",
                "weather.device_a.temperature",
            ],
        ]

        assert isinstance(aligned, AlignedTimeseries)
        assert aligned.series_names == [
            "weather.device_a.humidity",
            "weather.device_a.temperature",
        ]
        np.testing.assert_array_equal(aligned.timestamps, np.array([0, 1, 2], dtype=np.int64))
        assert aligned.shape == (3, 2)
        assert np.isnan(aligned.values[0, 0])
        assert aligned.values[0, 1] == 10.0
        assert aligned.values[1, 0] == 200.0
        assert np.isnan(aligned.values[1, 1])
        assert aligned.values[2, 0] == 300.0
        assert aligned.values[2, 1] == 30.0


def test_dataset_loc_supports_single_timestamp_and_mixed_series_specifiers(tmp_path):
    path = tmp_path / "weather.tsfile"
    _write_weather_file(path, 0)

    with TsFileDataFrame(str(path), show_progress=False) as tsdf:
        aligned = tsdf.loc[1, [0, "weather.device_a.humidity"]]

        assert isinstance(aligned, AlignedTimeseries)
        assert aligned.series_names == [
            "weather.device_a.temperature",
            "weather.device_a.humidity",
        ]
        np.testing.assert_array_equal(aligned.timestamps, np.array([1], dtype=np.int64))
        np.testing.assert_array_equal(aligned.values, np.array([[21.5, 52.0]]))


def test_dataset_loc_supports_open_ended_ranges_and_negative_series_index(tmp_path):
    path = tmp_path / "weather.tsfile"
    _write_weather_file(path, 100)

    with TsFileDataFrame(str(path), show_progress=False) as tsdf:
        aligned = tsdf.loc[:101, [-1]]

        assert isinstance(aligned, AlignedTimeseries)
        assert aligned.series_names == ["weather.device_a.humidity"]
        np.testing.assert_array_equal(aligned.timestamps, np.array([100, 101], dtype=np.int64))
        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):
    path = tmp_path / "weather_sparse_range.tsfile"
    _write_weather_rows_file(
        path,
        {
            "time": [0, 1, 2, 3],
            "device": ["device_a", "device_a", "device_a", "device_a"],
            "temperature": [10.0, np.nan, np.nan, 40.0],
            "humidity": [np.nan, 20.0, np.nan, 50.0],
        },
    )

    with TsFileDataFrame(str(path), show_progress=False) as tsdf:
        aligned = tsdf.loc[
            1:2,
            [
                "weather.device_a.temperature",
                "weather.device_a.humidity",
            ],
        ]

        assert isinstance(aligned, AlignedTimeseries)
        np.testing.assert_array_equal(aligned.timestamps, np.array([1, 2], dtype=np.int64))
        assert aligned.shape == (2, 2)
        assert np.isnan(aligned.values[0, 0])
        assert aligned.values[0, 1] == 20.0
        assert np.isnan(aligned.values[1, 0])
        assert np.isnan(aligned.values[1, 1])


def test_dataset_loc_single_timestamp_with_nulls_keeps_exact_time_window(tmp_path):
    path = tmp_path / "weather_sparse_point.tsfile"
    _write_weather_rows_file(
        path,
        {
            "time": [0, 1, 2],
            "device": ["device_a", "device_a", "device_a"],
            "temperature": [10.0, np.nan, 30.0],
            "humidity": [np.nan, 20.0, 40.0],
        },
    )

    with TsFileDataFrame(str(path), show_progress=False) as tsdf:
        aligned = tsdf.loc[
            1,
            [
                "weather.device_a.temperature",
                "weather.device_a.humidity",
            ],
        ]

        assert isinstance(aligned, AlignedTimeseries)
        np.testing.assert_array_equal(aligned.timestamps, np.array([1], dtype=np.int64))
        assert aligned.shape == (1, 2)
        assert np.isnan(aligned.values[0, 0])
        assert aligned.values[0, 1] == 20.0


def test_dataset_repr_only_builds_preview_rows(tmp_path, monkeypatch):
    path = tmp_path / "weather.tsfile"
    _write_weather_file(path, 0)

    with TsFileDataFrame(str(path), show_progress=False) as tsdf:
        tsdf._index.series_refs_ordered = [(0, 0)] * 1000

        built_rows = []

        def fake_build_series_info(series_ref):
            built_rows.append(series_ref)
            return {
                "table_name": "weather",
                "field": "temperature",
                "tag_columns": ("device",),
                "tag_values": {"device": "device_a"},
                "min_time": 0,
                "max_time": 2,
                "count": 3,
            }

        def fail_build_series_name(_series_ref):
            raise AssertionError("__repr__ should not build full series names for preview output")

        monkeypatch.setattr(tsdf, "_build_series_info", fake_build_series_info)
        monkeypatch.setattr(tsdf, "_build_series_name", fail_build_series_name)

        rendered = repr(tsdf)
        assert "TsFileDataFrame(1000 time series, 1 files)" in rendered
        assert "..." in rendered
        assert len(built_rows) == 20


def test_dataset_exposes_only_numeric_fields_and_keeps_nan(tmp_path):
    path = tmp_path / "numeric_and_text.tsfile"
    _write_numeric_and_text_file(path)

    with TsFileDataFrame(str(path), show_progress=False) as tsdf:
        assert tsdf.list_timeseries() == ["weather.device_a.temperature"]

        series = tsdf[0]
        assert series.name == "weather.device_a.temperature"
        assert len(series) == 3
        assert series.stats == {"start_time": 0, "end_time": 2, "count": 3}
        assert np.isnan(series[1])
        np.testing.assert_array_equal(series.timestamps, np.array([0, 1, 2], dtype=np.int64))
        sliced = series[:]
        assert sliced.shape == (3,)
        assert np.isnan(sliced[1])
        assert sliced[2] == 23.5
        assert series[1:1].shape == (0,)


def test_dataset_timeseries_supports_negative_step_slices(tmp_path):
    path = tmp_path / "weather.tsfile"
    _write_weather_file(path, 0)

    with TsFileDataFrame(str(path), show_progress=False) 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):
    path = tmp_path / "partial_numeric_rows.tsfile"
    _write_partial_numeric_rows_file(path)

    with TsFileDataFrame(str(path), show_progress=False) as tsdf:
        assert tsdf.list_timeseries() == [
            "weather.device_a.temperature",
            "weather.device_a.humidity",
        ]

        assert list(tsdf["count"]) == [2, 2]
        assert list(tsdf["start_time"]) == [0, 0]
        assert list(tsdf["end_time"]) == [1, 1]


def test_dataset_rejects_duplicate_timestamps_across_shards(tmp_path):
    path1 = tmp_path / "part1.tsfile"
    path2 = tmp_path / "part2.tsfile"
    _write_weather_file(path1, 0)
    _write_weather_file(path2, 2)

    with TsFileDataFrame([str(path1), str(path2)], show_progress=False) as tsdf:
        series = tsdf["weather.device_a.temperature"]
        with pytest.raises(ValueError, match="Duplicate timestamp"):
            _ = series.timestamps


def test_dataset_overlap_position_access_avoids_full_timestamp_materialization(tmp_path, monkeypatch):
    path1 = tmp_path / "part1.tsfile"
    path2 = tmp_path / "part2.tsfile"
    _write_weather_rows_file(
        path1,
        {
            "time": [0, 2, 4],
            "device": ["device_a", "device_a", "device_a"],
            "temperature": [10.0, 30.0, 50.0],
            "humidity": [100.0, 300.0, 500.0],
        },
    )
    _write_weather_rows_file(
        path2,
        {
            "time": [1, 3, 5],
            "device": ["device_a", "device_a", "device_a"],
            "temperature": [20.0, 40.0, 60.0],
            "humidity": [200.0, 400.0, 600.0],
        },
    )

    def fail_merge(*_args, **_kwargs):
        raise AssertionError("full timestamp merge should not run for overlap position reads")

    monkeypatch.setattr(dataframe_module, "_merge_field_timestamps", fail_merge)

    with TsFileDataFrame([str(path1), str(path2)], show_progress=False) as tsdf:
        series = tsdf["weather.device_a.temperature"]
        assert series[0] == 10.0
        assert series[1] == 20.0
        assert series[4] == 50.0
        np.testing.assert_array_equal(series[1:5], np.array([20.0, 30.0, 40.0, 50.0]))


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"
    _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)


def test_dataset_skips_empty_tsfile_shards(tmp_path):
    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) as tsdf:
        assert tsdf.list_timeseries() == [
            "weather.device_a.temperature",
            "weather.device_a.humidity",
        ]


def test_reader_allows_empty_tsfile(tmp_path):
    path = tmp_path / "empty.tsfile"
    _write_empty_weather_file(path)

    reader = TsFileSeriesReader(str(path), show_progress=False)
    try:
        assert reader.series_paths == []
        assert reader.catalog.series_count == 0
    finally:
        reader.close()


def test_dataset_multi_tag_metadata_discovery(tmp_path):
    path = tmp_path / "multi_tag.tsfile"
    _write_multi_tag_file(path)

    with TsFileDataFrame(str(path), show_progress=False) as tsdf:
        assert tsdf.list_timeseries() == [
            "weather.beijing.device_a.temperature",
            "weather.beijing.device_a.humidity",
            "weather.shanghai.device_b.temperature",
            "weather.shanghai.device_b.humidity",
        ]

        summary = pd.DataFrame(
            {
                "series_path": tsdf.list_timeseries(),
                "table": tsdf["table"],
                "city": tsdf["city"],
                "device": tsdf["device"],
                "field": tsdf["field"],
                "start_time": tsdf["start_time"],
                "end_time": tsdf["end_time"],
                "count": tsdf["count"],
            }
        ).sort_values(["city", "device", "field"]).reset_index(drop=True)
        assert list(summary.columns) == [
            "series_path",
            "table",
            "city",
            "device",
            "field",
            "start_time",
            "end_time",
            "count",
        ]
        assert list(summary["city"]) == ["beijing", "beijing", "shanghai", "shanghai"]
        assert list(summary["device"]) == ["device_a", "device_a", "device_b", "device_b"]
        assert list(summary["field"]) == ["humidity", "temperature", "humidity", "temperature"]
        assert list(summary["count"]) == [2, 2, 2, 2]


def test_dataset_series_paths_escape_special_tag_values(tmp_path):
    path = tmp_path / "special_tag.tsfile"
    _write_special_tag_file(path)

    with TsFileDataFrame(str(path), show_progress=False) as tsdf:
        expected_path = r"weather.bei\.jing.dev\\1.temperature"
        assert tsdf.list_timeseries() == [expected_path]

        series = tsdf[expected_path]
        assert isinstance(series, Timeseries)
        assert series.name == expected_path
        assert list(tsdf["city"]) == ["bei.jing"]
        assert list(tsdf["device"]) == [r"dev\1"]


def test_reader_series_paths_escape_special_tag_values(tmp_path):
    path = tmp_path / "special_tag.tsfile"
    _write_special_tag_file(path)

    reader = TsFileSeriesReader(str(path), show_progress=False)
    try:
        expected_path = r"weather.bei\.jing.dev\\1.temperature"
        assert reader.series_paths == [expected_path]
        info = reader.get_series_info(expected_path)
        assert info["tag_values"] == {"city": "bei.jing", "device": r"dev\1"}
    finally:
        reader.close()


def test_reader_catalog_shares_device_metadata_and_resolves_paths(tmp_path):
    path = tmp_path / "weather.tsfile"
    _write_weather_file(path, 100)

    reader = TsFileSeriesReader(str(path), show_progress=False)
    try:
        assert reader.series_paths == [
            "weather.device_a.temperature",
            "weather.device_a.humidity",
        ]
        assert len(reader.catalog.table_entries) == 1
        assert len(reader.catalog.device_entries) == 1
        assert reader.catalog.series_count == 2

        by_path = reader.get_series_info("weather.device_a.temperature")
        by_ref = reader.get_series_info_by_ref(0, 0)
        assert by_ref == by_path
        assert by_ref["tag_values"] == {"device": "device_a"}
        ts_arr, values = reader.read_series_by_ref(0, 0, 100, 102)
        np.testing.assert_array_equal(ts_arr, np.array([100, 101, 102]))
        np.testing.assert_array_equal(values, np.array([20.0, 21.5, 23.0]))
    finally:
        reader.close()


def test_reader_read_series_by_row_retries_across_native_row_query_boundaries():
    class _FakeResultSet:
        def __init__(self, rows):
            self._rows = rows
            self._index = -1

        def __enter__(self):
            return self

        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]

    class _FakeNativeReader:
        def __init__(self, timestamps, values, boundary):
            self._timestamps = timestamps
            self._values = values
            self._boundary = boundary

        def query_table_by_row(self, table_name, column_names, offset=0, limit=-1, tag_filter=None):
            assert table_name == "pvf"
            assert column_names == ["totalcloudcover"]
            assert tag_filter is None
            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
            # 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)

    reader = object.__new__(TsFileSeriesReader)
    reader._reader = _FakeNativeReader(np.arange(30, dtype=np.int64), np.arange(30, dtype=np.float64), boundary=10)
    reader._catalog = MetadataCatalog()
    table_id = reader._catalog.add_table("pvf", (), (), ("totalcloudcover",))
    device_id = reader._catalog.add_device(table_id, (), 0, 29)

    ts_arr, values = reader.read_series_by_row(device_id, 0, 5, 12)
    np.testing.assert_array_equal(ts_arr, np.arange(5, 17, dtype=np.int64))
    np.testing.assert_array_equal(values, np.arange(5, 17, dtype=np.float64))


def test_series_path_resolution_allows_prefix_tag_values():
    catalog = MetadataCatalog()
    table_id = catalog.add_table(
        "weather",
        ("site", "device", "region"),
        (TSDataType.STRING, TSDataType.STRING, TSDataType.STRING),
        ("temperature",),
    )
    device_id = catalog.add_device(table_id, ("site_a", "device_a"), 0, 1)
    catalog.series_stats_by_ref[(device_id, 0)] = {
        "length": 1,
        "min_time": 0,
        "max_time": 0,
        "timeline_length": 1,
        "timeline_min_time": 0,
        "timeline_max_time": 0,
    }

    series_path = build_series_path(catalog, device_id, 0)
    assert series_path == "weather.site_a.device_a.temperature"
    assert resolve_series_path(catalog, series_path) == (table_id, device_id, 0)


def test_series_path_resolution_allows_missing_trailing_tag_value():
    catalog = MetadataCatalog()
    table_id = catalog.add_table(
        "weather",
        ("device",),
        (TSDataType.STRING,),
        ("temperature",),
    )
    device_id = catalog.add_device(table_id, (), 0, 1)
    catalog.series_stats_by_ref[(device_id, 0)] = {
        "length": 1,
        "min_time": 0,
        "max_time": 0,
        "timeline_length": 1,
        "timeline_min_time": 0,
        "timeline_max_time": 0,
    }

    series_path = build_series_path(catalog, device_id, 0)
    assert series_path == "weather.temperature"
    assert resolve_series_path(catalog, series_path) == (table_id, device_id, 0)


def test_series_path_resolution_uses_named_tags_for_sparse_non_prefix_values():
    catalog = MetadataCatalog()
    table_id = catalog.add_table(
        "weather",
        ("city", "device", "region"),
        (TSDataType.STRING, TSDataType.STRING, TSDataType.STRING),
        ("temperature",),
    )
    device_id = catalog.add_device(table_id, (None, "device_a", None), 0, 1)
    catalog.series_stats_by_ref[(device_id, 0)] = {
        "length": 1,
        "min_time": 0,
        "max_time": 0,
        "timeline_length": 1,
        "timeline_min_time": 0,
        "timeline_max_time": 0,
    }

    series_path = build_series_path(catalog, device_id, 0)
    assert series_path == "weather.device_a.temperature"
    assert resolve_series_path(catalog, series_path) == (table_id, device_id, 0)


def test_reader_metadata_tag_values_trim_trailing_none():
    class _Group:
        segments = ("weather", "device_a", None, None)

    assert TsFileSeriesReader._metadata_tag_values(_Group(), 3) == ("device_a",)
    assert TsFileSeriesReader._metadata_tag_values(_Group(), 1) == ("device_a",)


def test_exact_tag_filter_rejects_none_tag_values():
    with pytest.raises(NotImplementedError, match="IS NULL / IS NOT NULL"):
        _build_exact_tag_filter({"device": None})
    with pytest.raises(NotImplementedError, match="IS NULL / IS NOT NULL"):
        _build_exact_tag_filter({"city": "beijing", "device": None})


def test_reader_exact_match_with_none_tag_values_fails_fast():
    class _FakeNativeReader:
        def query_table(self, *args, **kwargs):
            raise AssertionError("query should not be issued when None-tag exact matching is unsupported")

        def query_table_by_row(self, *args, **kwargs):
            raise AssertionError("row query should not be issued when None-tag exact matching is unsupported")

    reader = object.__new__(TsFileSeriesReader)
    reader._reader = _FakeNativeReader()
    reader._catalog = MetadataCatalog()
    table_id = reader._catalog.add_table(
        "weather",
        ("city", "device", "region"),
        (TSDataType.STRING, TSDataType.STRING, TSDataType.STRING),
        ("temperature",),
    )
    device_id = reader._catalog.add_device(table_id, (None, "device_a", "north"), 0, 1)
    reader._catalog.series_stats_by_ref[(device_id, 0)] = {
        "length": 2,
        "min_time": 0,
        "max_time": 1,
        "timeline_length": 2,
        "timeline_min_time": 0,
        "timeline_max_time": 1,
    }

    with pytest.raises(NotImplementedError, match="IS NULL / IS NOT NULL"):
        reader.read_series_by_ref(device_id, 0, 0, 1)
    with pytest.raises(NotImplementedError, match="IS NULL / IS NOT NULL"):
        reader.read_series_by_row(device_id, 0, 0, 2)


def test_dataframe_resolves_named_sparse_tag_series_path():
    tsdf = object.__new__(TsFileDataFrame)
    tsdf._index = dataframe_module._LogicalIndex()
    tsdf._index.table_entries["weather"] = dataframe_module.TableEntry(
        table_name="weather",
        tag_columns=("city", "device", "region"),
        tag_types=(TSDataType.STRING, TSDataType.STRING, TSDataType.STRING),
        field_columns=("temperature",),
    )
    device_key = ("weather", (None, "device_a"))
    tsdf._index.device_order = [device_key]
    tsdf._index.device_index_by_key = {device_key: 0}
    tsdf._index.tables_with_sparse_tag_values = {"weather"}
    tsdf._index.sparse_device_indices_by_compressed_path = {("weather", ("device_a",)): [0]}
    tsdf._index.device_refs = [[]]
    tsdf._index.series_refs_ordered = [(0, 0)]
    tsdf._index.series_ref_set = {(0, 0)}
    tsdf._index.series_ref_map = {(0, 0): []}

    assert tsdf.list_timeseries() == ["weather.device_a.temperature"]
    assert tsdf._resolve_series_name("weather.device_a.temperature") == (0, 0)


def test_dataframe_list_timeseries_filters_named_sparse_tag_prefix():
    tsdf = object.__new__(TsFileDataFrame)
    tsdf._index = dataframe_module._LogicalIndex()
    tsdf._index.table_entries["weather"] = dataframe_module.TableEntry(
        table_name="weather",
        tag_columns=("city", "device", "region"),
        tag_types=(TSDataType.STRING, TSDataType.STRING, TSDataType.STRING),
        field_columns=("temperature",),
    )
    tsdf._index.device_order = [
        ("weather", (None, "device_a")),
        ("weather", ("beijing", "device_b")),
    ]
    tsdf._index.device_index_by_key = {
        ("weather", (None, "device_a")): 0,
        ("weather", ("beijing", "device_b")): 1,
    }
    tsdf._index.tables_with_sparse_tag_values = {"weather"}
    tsdf._index.sparse_device_indices_by_compressed_path = {
        ("weather", ("device_a",)): [0],
        ("weather", ("beijing", "device_b")): [1],
    }
    tsdf._index.device_refs = [[], []]
    tsdf._index.series_refs_ordered = [(0, 0), (1, 0)]
    tsdf._index.series_ref_set = {(0, 0), (1, 0)}
    tsdf._index.series_ref_map = {(0, 0): [], (1, 0): []}

    assert tsdf.list_timeseries("weather.device_a") == ["weather.device_a.temperature"]


def test_dataframe_list_timeseries_prefix_can_skip_full_name_build(tmp_path, monkeypatch):
    path = tmp_path / "weather.tsfile"
    _write_weather_file(path, 0)

    with TsFileDataFrame(str(path), show_progress=False) as tsdf:
        tsdf._index.series_refs_ordered = [(0, 0)] * 1000

        def fail_build_series_name(_series_ref):
            raise AssertionError("list_timeseries(prefix) should not build full names for non-matching series")

        monkeypatch.setattr(tsdf, "_build_series_name", fail_build_series_name)
        assert tsdf.list_timeseries("pvf") == []


def test_series_path_resolution_reports_ambiguous_sparse_path():
    catalog = MetadataCatalog()
    table_id = catalog.add_table(
        "weather",
        ("city", "device"),
        (TSDataType.STRING, TSDataType.STRING),
        ("temperature",),
    )
    first_id = catalog.add_device(table_id, ("beijing", None), 0, 1)
    second_id = catalog.add_device(table_id, (None, "beijing"), 0, 1)
    for device_id in (first_id, second_id):
        catalog.series_stats_by_ref[(device_id, 0)] = {
            "length": 1,
            "min_time": 0,
            "max_time": 0,
            "timeline_length": 1,
            "timeline_min_time": 0,
            "timeline_max_time": 0,
        }

    assert build_series_path(catalog, first_id, 0) == "weather.beijing.temperature"
    assert build_series_path(catalog, second_id, 0) == "weather.beijing.temperature"
    with pytest.raises(ValueError, match="Ambiguous series path"):
        resolve_series_path(catalog, "weather.beijing.temperature")


def test_reader_show_progress_reports_start_immediately(tmp_path, capsys):
    path = tmp_path / "weather.tsfile"
    _write_weather_file(path, 0)

    reader = TsFileSeriesReader(str(path), show_progress=True)
    try:
        stderr = capsys.readouterr().err
        assert "Reading TsFile metadata: 0/1" in stderr
        assert "Reading TsFile metadata: 1 table(s), 2 series ... done" in stderr
    finally:
        reader.close()


def test_dataframe_parallel_show_progress_reports_start_immediately(tmp_path, capsys):
    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=True):
        pass

    stderr = capsys.readouterr().err
    assert "Loading TsFile shards: 0/2" in stderr
    assert "Loading TsFile shards: 2/2 (4 series) ... done" in stderr