You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tsfile.apache.org by ge...@apache.org on 2026/09/09 01:56:54 UTC
(tsfile) 03/03: feat(python): add trusted index and data attestations
This is an automated email from the ASF dual-hosted git repository.
RkGrit pushed a commit to branch gewu/timer4-batched-window-read-clean
in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit 94649f3e4a1d0f1245cf0616ac24c3853db2769a
Author: Gewu <89...@users.noreply.github.com>
AuthorDate: Tue Sep 8 15:31:37 2026 +0800
feat(python): add trusted index and data attestations
---
python/tests/test_dataset_index.py | 202 +++++++++++++++++++++++++
python/tests/test_index_identity.py | 213 ++++++++++++++++++++++++++
python/tsfile/dataset/__init__.py | 27 +++-
python/tsfile/dataset/dataframe.py | 64 ++++++--
python/tsfile/dataset/index_identity.py | 254 ++++++++++++++++++++++++++++++++
python/tsfile/dataset/runtime.py | 142 +++++++++++++++++-
6 files changed, 890 insertions(+), 12 deletions(-)
diff --git a/python/tests/test_dataset_index.py b/python/tests/test_dataset_index.py
index 9b3eab077..684ff2f3c 100644
--- a/python/tests/test_dataset_index.py
+++ b/python/tests/test_dataset_index.py
@@ -24,6 +24,8 @@ import pandas as pd
import pyarrow as pa
import pytest
+import tsfile.dataset.dataframe as dataframe_module
+import tsfile.dataset.index_identity as identity_module
import tsfile.dataset.index as index_module
import tsfile.dataset.runtime as runtime_module
from tsfile import (
@@ -1746,3 +1748,203 @@ def test_prefork_timeseries_fails_fast_in_child_without_harming_parent(tmp_path)
inherited_query_lease.close()
assert parent_runtime._query_leases == 0
series.close()
+
+
+def _build_runtime_attestations(paths):
+ canonical_paths = [os.path.realpath(os.fspath(path)) for path in paths]
+ with TsFileDataFrame(
+ canonical_paths,
+ show_progress=False,
+ use_index=True,
+ ) as dataframe:
+ index_path = dataframe._runtime.index.path
+
+ index_attestation = identity_module.attest_index(index_path)
+ with MappedDatasetIndex(index_path, trust_index=True) as mapped_index:
+ data_attestations = tuple(
+ identity_module.attest_data_file(
+ file_id,
+ mapped_index.string(record[0]),
+ existing_index_fingerprint=f"{record[3]:016x}",
+ )
+ for file_id in range(mapped_index.count(TSFILE_RECORD))
+ for record in (mapped_index.record(TSFILE_RECORD, file_id),)
+ )
+ return index_path, index_attestation, data_attestations
+
+
+def test_attested_trusted_dataframe_uses_manifest_without_directory_scan(
+ tmp_path, monkeypatch
+):
+ paths = [tmp_path / "part-0.tsfile", tmp_path / "part-1.tsfile"]
+ _write_runtime_file(paths[0], 0)
+ _write_runtime_file(paths[1], 10)
+ _, index_attestation, data_attestations = _build_runtime_attestations(paths)
+
+ def unexpected_scan(_paths):
+ raise AssertionError("attested workers must not expand or scan dataset paths")
+
+ monkeypatch.setattr(dataframe_module, "_expand_paths", unexpected_scan)
+ with TsFileDataFrame(
+ str(tmp_path),
+ show_progress=False,
+ trust_index=True,
+ index_attestation=index_attestation,
+ data_file_attestations=data_attestations,
+ ) as dataframe:
+ runtime = dataframe._runtime
+ assert tuple(dataframe._paths) == tuple(
+ item.canonical_path for item in data_attestations
+ )
+ assert runtime.data_manifest_identity_sha256 == (
+ identity_module.data_manifest_identity(
+ data_attestations,
+ require_sorted=True,
+ )
+ )
+ assert tuple(runtime.file_generation_tokens) == (0, 1)
+ assert runtime.prepared._mapped_index_identity == int.from_bytes(
+ bytes.fromhex(index_attestation.index_identity_sha256)[:8],
+ byteorder="little",
+ )
+ np.testing.assert_array_equal(
+ dataframe[0][:],
+ np.array([0.0, 1.0, 10.0, 11.0]),
+ )
+
+
+def test_attested_dataframe_requires_explicit_trust_and_complete_pair(tmp_path):
+ path = tmp_path / "part.tsfile"
+ _write_runtime_file(path, 0)
+ _, index_attestation, data_attestations = _build_runtime_attestations((path,))
+
+ with pytest.raises(ValueError, match="trust_index=True"):
+ TsFileDataFrame(
+ str(tmp_path),
+ show_progress=False,
+ use_index=True,
+ index_attestation=index_attestation,
+ data_file_attestations=data_attestations,
+ )
+ with pytest.raises(ValueError, match="provided together"):
+ TsFileDataFrame(
+ str(tmp_path),
+ show_progress=False,
+ trust_index=True,
+ index_attestation=index_attestation,
+ )
+ with pytest.raises(ValueError, match="provided together"):
+ TsFileDataFrame(
+ str(tmp_path),
+ show_progress=False,
+ trust_index=True,
+ data_file_attestations=data_attestations,
+ )
+
+
+def test_attested_dataframe_rejects_incomplete_order_and_unbound_fingerprint(tmp_path):
+ paths = [tmp_path / "part-0.tsfile", tmp_path / "part-1.tsfile"]
+ _write_runtime_file(paths[0], 0)
+ _write_runtime_file(paths[1], 10)
+ _, index_attestation, data_attestations = _build_runtime_attestations(paths)
+
+ with pytest.raises(ValueError, match="file_id=0..N-1 in order"):
+ TsFileDataFrame(
+ str(tmp_path),
+ show_progress=False,
+ trust_index=True,
+ index_attestation=index_attestation,
+ data_file_attestations=data_attestations[:1],
+ )
+ with pytest.raises(ValueError, match="file_id=0..N-1 in order"):
+ TsFileDataFrame(
+ str(tmp_path),
+ show_progress=False,
+ trust_index=True,
+ index_attestation=index_attestation,
+ data_file_attestations=tuple(reversed(data_attestations)),
+ )
+
+ first = data_attestations[0]
+ unbound = identity_module.DataFileAttestation(
+ file_id=first.file_id,
+ canonical_path=first.canonical_path,
+ st_dev=first.st_dev,
+ st_ino=first.st_ino,
+ st_size=first.st_size,
+ st_mtime_ns=first.st_mtime_ns,
+ )
+ with pytest.raises(
+ identity_module.AttestationMismatchError,
+ match="fingerprint",
+ ):
+ TsFileDataFrame(
+ str(tmp_path),
+ show_progress=False,
+ trust_index=True,
+ index_attestation=index_attestation,
+ data_file_attestations=(unbound, data_attestations[1]),
+ )
+
+
+def test_attested_index_mismatch_fails_before_mapping(tmp_path, monkeypatch):
+ path = tmp_path / "part.tsfile"
+ _write_runtime_file(path, 0)
+ index_path, index_attestation, data_attestations = _build_runtime_attestations(
+ (path,)
+ )
+ with open(index_path, "ab") as stream:
+ stream.write(b"x")
+
+ mapped_calls = 0
+
+ def forbidden_mapping(*_args, **_kwargs):
+ nonlocal mapped_calls
+ mapped_calls += 1
+ raise AssertionError("a mismatched index must not be mapped")
+
+ monkeypatch.setattr(runtime_module, "MappedDatasetIndex", forbidden_mapping)
+ with pytest.raises(
+ identity_module.AttestationMismatchError,
+ match="index stat",
+ ):
+ TsFileDataFrame(
+ str(tmp_path),
+ show_progress=False,
+ trust_index=True,
+ index_attestation=index_attestation,
+ data_file_attestations=data_attestations,
+ )
+ assert mapped_calls == 0
+
+
+def test_attested_data_mismatch_fails_before_native_reader_open(tmp_path, monkeypatch):
+ path = tmp_path / "part.tsfile"
+ _write_runtime_file(path, 0)
+ _, index_attestation, data_attestations = _build_runtime_attestations((path,))
+
+ replacement = tmp_path / "replacement.tsfile"
+ replacement.write_bytes(b"X" * path.stat().st_size)
+ os.replace(replacement, path)
+
+ native_open_calls = 0
+
+ def forbidden_reader(*_args, **_kwargs):
+ nonlocal native_open_calls
+ native_open_calls += 1
+ raise AssertionError("a mismatched data file must not reach the native reader")
+
+ monkeypatch.setattr(runtime_module, "TsFileReaderPy", forbidden_reader)
+ with TsFileDataFrame(
+ str(tmp_path),
+ show_progress=False,
+ trust_index=True,
+ index_attestation=index_attestation,
+ data_file_attestations=data_attestations,
+ ) as dataframe:
+ with pytest.raises(
+ identity_module.AttestationMismatchError,
+ match="data-file stat",
+ ):
+ dataframe[0][:]
+ assert native_open_calls == 0
diff --git a/python/tests/test_index_identity.py b/python/tests/test_index_identity.py
new file mode 100644
index 000000000..ea3cd0ef4
--- /dev/null
+++ b/python/tests/test_index_identity.py
@@ -0,0 +1,213 @@
+# 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 dataclasses
+import hashlib
+import os
+from pathlib import Path
+import runpy
+from types import SimpleNamespace
+
+import pytest
+
+import tsfile.dataset.index_identity as identity
+
+
+def test_index_identity_module_imports_with_python39_dataclass_api(monkeypatch):
+ """The public dataset package supports Python 3.9, which has no slots= API."""
+
+ real_dataclass = dataclasses.dataclass
+
+ def python39_dataclass(*args, **kwargs):
+ if "slots" in kwargs:
+ raise TypeError("dataclass() got an unexpected keyword argument 'slots'")
+ return real_dataclass(*args, **kwargs)
+
+ monkeypatch.setattr(dataclasses, "dataclass", python39_dataclass)
+ runpy.run_path(identity.__file__, run_name="_tsfile_index_identity_py39_probe")
+
+
+def test_index_attestation_hashes_exact_bytes_and_worker_validation_is_metadata_only(
+ tmp_path, monkeypatch
+):
+ first = tmp_path / "first.tsidx"
+ first.write_bytes(b"final-index\x00bytes")
+ attestation = identity.attest_index(first, chunk_size=3)
+ assert (
+ attestation.index_identity_sha256
+ == hashlib.sha256(first.read_bytes()).hexdigest()
+ )
+
+ relocated = tmp_path / "relocated.tsidx"
+ relocated.write_bytes(first.read_bytes())
+ relocated_attestation = identity.attest_index(relocated)
+ assert relocated_attestation.index_identity_sha256 == (
+ attestation.index_identity_sha256
+ )
+ relocated.write_bytes(b"Final-index\x00bytes")
+ assert identity.attest_index(relocated).index_identity_sha256 != (
+ attestation.index_identity_sha256
+ )
+
+ hash_calls = 0
+
+ def forbidden_hash(*_args, **_kwargs):
+ nonlocal hash_calls
+ hash_calls += 1
+ raise AssertionError("worker validation must not hash index contents")
+
+ monkeypatch.setattr(identity, "attest_index", forbidden_hash)
+ for _ in range(32):
+ identity.validate_index_attestation(first, attestation)
+ assert hash_calls == 0
+
+ first.write_bytes(first.read_bytes() + b"x")
+ with pytest.raises(identity.AttestationMismatchError, match="index stat"):
+ identity.validate_index_attestation(first, attestation)
+
+
+def test_index_attestation_fails_if_open_file_changes_while_hashing(
+ tmp_path, monkeypatch
+):
+ path = tmp_path / "changing.tsidx"
+ path.write_bytes(b"stable-index-bytes")
+ real_fstat = identity.os.fstat
+ fstat_calls = 0
+
+ def changing_fstat(file_descriptor):
+ nonlocal fstat_calls
+ stat_result = real_fstat(file_descriptor)
+ fstat_calls += 1
+ if fstat_calls == 1:
+ return stat_result
+ return SimpleNamespace(
+ st_dev=stat_result.st_dev,
+ st_ino=stat_result.st_ino,
+ st_size=stat_result.st_size,
+ st_mtime_ns=stat_result.st_mtime_ns + 1,
+ )
+
+ monkeypatch.setattr(identity.os, "fstat", changing_fstat)
+ with pytest.raises(
+ identity.AttestationMismatchError, match="changed while hashing"
+ ):
+ identity.attest_index(path, chunk_size=3)
+ assert fstat_calls == 2
+
+
+def test_manifest_and_generation_are_ordered_versioned_and_fail_closed(tmp_path):
+ paths = [tmp_path / "a.tsfile", tmp_path / "b.tsfile"]
+ for ordinal, path in enumerate(paths):
+ path.write_bytes(bytes([ordinal]) * (ordinal + 1))
+ files = tuple(identity.attest_data_file(i, path) for i, path in enumerate(paths))
+
+ manifest = identity.data_manifest_identity(files)
+ assert len(manifest) == 64 and manifest == manifest.lower()
+ assert identity.data_manifest_identity(tuple(reversed(files))) != manifest
+ with pytest.raises(ValueError, match="strictly increasing"):
+ identity.data_manifest_identity(tuple(reversed(files)), require_sorted=True)
+
+ token = identity.file_generation_token("11" * 32, manifest, files[0])
+ assert token == identity.file_generation_token("11" * 32, manifest, files[0])
+ assert token != identity.file_generation_token("11" * 32, manifest, files[1])
+ identity.validate_data_file_attestation(paths[0], files[0])
+ replacement = tmp_path / "replacement"
+ replacement.write_bytes(b"replacement")
+ os.replace(replacement, paths[0])
+ with pytest.raises(identity.AttestationMismatchError, match="data-file stat"):
+ identity.validate_data_file_attestation(paths[0], files[0])
+
+
+@pytest.mark.parametrize(
+ "fingerprint",
+ ["", "0" * 15, "0" * 17, "ABCDEF0123456789", "0123456789abcdeg"],
+)
+def test_data_file_attestation_rejects_noncanonical_index_fingerprint(
+ tmp_path, fingerprint
+):
+ path = tmp_path / "data.tsfile"
+ path.write_bytes(b"data")
+ with pytest.raises(ValueError, match="16-character lowercase hex"):
+ identity.attest_data_file(0, path, existing_index_fingerprint=fingerprint)
+
+
+def test_data_file_fingerprint_has_unique_canonical_identity(tmp_path):
+ path = tmp_path / "data.tsfile"
+ path.write_bytes(b"data")
+ without_fingerprint = identity.attest_data_file(0, path)
+ with_fingerprint = identity.attest_data_file(
+ 0, path, existing_index_fingerprint="0123456789abcdef"
+ )
+ manifest_without = identity.data_manifest_identity((without_fingerprint,))
+ manifest_with = identity.data_manifest_identity((with_fingerprint,))
+ assert manifest_without != manifest_with
+ assert identity.file_generation_token(
+ "11" * 32, manifest_without, without_fingerprint
+ ) != identity.file_generation_token("11" * 32, manifest_with, with_fingerprint)
+
+
+def test_data_file_attestation_stats_the_opened_object(tmp_path, monkeypatch):
+ path = tmp_path / "data.tsfile"
+ path.write_bytes(b"data")
+ real_fstat = identity.os.fstat
+ fstat_calls = 0
+
+ def recording_fstat(file_descriptor):
+ nonlocal fstat_calls
+ fstat_calls += 1
+ return real_fstat(file_descriptor)
+
+ monkeypatch.setattr(identity.os, "fstat", recording_fstat)
+ attestation = identity.attest_data_file(7, path)
+ assert attestation.file_id == 7
+ assert attestation.st_size == 4
+ assert fstat_calls == 1
+
+
+@pytest.mark.parametrize("value", ["", "A" * 64, "0" * 63, "gg" * 32])
+def test_generation_token_digest_validation_is_strict(value):
+ with pytest.raises(ValueError, match="lowercase SHA-256"):
+ identity.file_generation_token(value, "44" * 32, object())
+
+
+def test_attestations_are_immutable_and_canonical_paths_are_absolute(tmp_path):
+ path = tmp_path / "index.tsidx"
+ path.write_bytes(b"index")
+ attestation = identity.attest_index(path)
+ assert Path(attestation.canonical_index_path).is_absolute()
+ with pytest.raises((AttributeError, TypeError)):
+ attestation.index_size = 1
+
+
+def test_dataset_identity_public_api_is_generic_and_explicit():
+ import tsfile.dataset as dataset_api
+
+ expected = (
+ "AttestationMismatchError",
+ "DataFileAttestation",
+ "IndexAttestation",
+ "attest_data_file",
+ "attest_index",
+ "data_manifest_identity",
+ "file_generation_token",
+ "validate_data_file_attestation",
+ "validate_index_attestation",
+ )
+ for name in expected:
+ assert getattr(dataset_api, name) is getattr(identity, name)
+ assert name in dataset_api.__all__
+ assert "read_plan_identity" not in dataset_api.__all__
diff --git a/python/tsfile/dataset/__init__.py b/python/tsfile/dataset/__init__.py
index 15c20d540..bbdcb6866 100644
--- a/python/tsfile/dataset/__init__.py
+++ b/python/tsfile/dataset/__init__.py
@@ -19,7 +19,32 @@
"""Dataset-style TsFile accessors."""
from .dataframe import TsFileDataFrame
+from .index_identity import (
+ AttestationMismatchError,
+ DataFileAttestation,
+ IndexAttestation,
+ attest_data_file,
+ attest_index,
+ data_manifest_identity,
+ file_generation_token,
+ validate_data_file_attestation,
+ validate_index_attestation,
+)
from .metadata import SeriesPath
from .timeseries import AlignedTimeseries, Timeseries
-__all__ = ["TsFileDataFrame", "Timeseries", "AlignedTimeseries", "SeriesPath"]
+__all__ = [
+ "AlignedTimeseries",
+ "AttestationMismatchError",
+ "DataFileAttestation",
+ "IndexAttestation",
+ "SeriesPath",
+ "Timeseries",
+ "TsFileDataFrame",
+ "attest_data_file",
+ "attest_index",
+ "data_manifest_identity",
+ "file_generation_token",
+ "validate_data_file_attestation",
+ "validate_index_attestation",
+]
diff --git a/python/tsfile/dataset/dataframe.py b/python/tsfile/dataset/dataframe.py
index c81990777..8d4bc1c66 100644
--- a/python/tsfile/dataset/dataframe.py
+++ b/python/tsfile/dataset/dataframe.py
@@ -25,7 +25,7 @@ import heapq
import os
import sys
from types import SimpleNamespace
-from typing import Dict, List, Optional, Tuple, Union
+from typing import TYPE_CHECKING, Dict, List, Optional, Sequence, Tuple, Union
import numpy as np
@@ -43,6 +43,9 @@ from .metadata import (
from .merge import build_aligned_matrix, merge_time_value_parts, merge_timestamp_parts
from .timeseries import AlignedTimeseries, Timeseries
+if TYPE_CHECKING:
+ from .index_identity import DataFileAttestation, IndexAttestation
+
DeviceKey = Tuple[str, tuple]
SeriesRefKey = Tuple[int, int]
SeriesRef = Tuple[object, int, int]
@@ -811,16 +814,41 @@ class TsFileDataFrame:
show_progress: bool = True,
use_index: bool = False,
trust_index: Optional[bool] = None,
+ index_attestation: Optional["IndexAttestation"] = None,
+ data_file_attestations: Optional[Sequence["DataFileAttestation"]] = None,
):
if not isinstance(use_index, bool):
raise TypeError("use_index must be a bool")
self._trust_index = _resolve_trust_index(trust_index)
- self._paths = _expand_paths(paths)
+ has_index_attestation = index_attestation is not None
+ has_data_attestations = data_file_attestations is not None
+ if has_index_attestation != has_data_attestations:
+ raise ValueError(
+ "index_attestation and data_file_attestations must be provided together"
+ )
+ if has_index_attestation and not self._trust_index:
+ raise ValueError("dataset attestations require trust_index=True")
+
+ self._index_attestation = index_attestation
+ self._data_file_attestations = tuple(data_file_attestations or ())
+ if has_index_attestation:
+ from .index_identity import DataFileAttestation, IndexAttestation
+
+ if not isinstance(index_attestation, IndexAttestation):
+ raise TypeError("index_attestation must be an IndexAttestation")
+ if any(
+ not isinstance(item, DataFileAttestation)
+ for item in self._data_file_attestations
+ ):
+ raise TypeError(
+ "data_file_attestations must contain DataFileAttestation values"
+ )
+ self._paths = [item.canonical_path for item in self._data_file_attestations]
+ else:
+ self._paths = _expand_paths(paths)
self._show_progress = show_progress
- # A trusted index is meaningful only when the persistent index path is
- # used. Make the safe, explicit fast-path convenient for callers by
- # enabling it automatically instead of requiring two flags.
- self._use_index = use_index or self._trust_index
+ # Trusted and attested indexes always use the persistent mmap path.
+ self._use_index = use_index or self._trust_index or has_index_attestation
self._readers: Dict[str, object] = {}
self._index = _DataFrameCatalog()
self._is_view = False
@@ -843,6 +871,8 @@ class TsFileDataFrame:
obj._show_progress = parent._show_progress
obj._use_index = parent._use_index
obj._trust_index = parent._trust_index
+ obj._index_attestation = parent._index_attestation
+ obj._data_file_attestations = parent._data_file_attestations
obj._readers = parent._readers
subset_refs = list(series_refs)
obj._index = SimpleNamespace(
@@ -943,11 +973,18 @@ class TsFileDataFrame:
)
from .runtime import DatasetRuntime
- index_path = index_path_for(self._paths)
+ index_path = (
+ self._index_attestation.canonical_index_path
+ if self._index_attestation is not None
+ else index_path_for(self._paths)
+ )
if self._trust_index:
if not os.path.isfile(index_path):
+ index_kind = (
+ "Attested" if self._index_attestation is not None else "Trusted"
+ )
raise FileNotFoundError(
- f"Trusted Dataset Index not found: {index_path}"
+ f"{index_kind} Dataset Index not found: {index_path}"
)
elif not index_matches_paths(index_path, self._paths):
lock_path = index_path + ".lock"
@@ -968,7 +1005,16 @@ class TsFileDataFrame:
reader.close()
self._readers.clear()
- self._runtime = DatasetRuntime(index_path, trust_index=self._trust_index)
+ self._runtime = DatasetRuntime(
+ index_path,
+ trust_index=self._trust_index,
+ index_attestation=self._index_attestation,
+ data_file_attestations=(
+ self._data_file_attestations
+ if self._index_attestation is not None
+ else None
+ ),
+ )
self._runtime_lease = self._runtime.lease()
self._index = self._runtime.catalog
if len(self._index.series) == 0:
diff --git a/python/tsfile/dataset/index_identity.py b/python/tsfile/dataset/index_identity.py
new file mode 100644
index 000000000..dbb687ba3
--- /dev/null
+++ b/python/tsfile/dataset/index_identity.py
@@ -0,0 +1,254 @@
+# 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.
+
+"""Content and generation identities for trusted dataset indexes.
+
+Expensive content hashing belongs to the parent attestation pass. Workers use
+the validation helpers, which compare canonical paths and stat metadata only.
+The encodings below are explicitly versioned and length-prefixed so they do not
+depend on Python object identity, mapping order, locale, or pickle.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import hashlib
+import os
+import struct
+from typing import BinaryIO, Iterable
+
+
+_INDEX_SCHEMA_VERSION = 1
+_DATA_MANIFEST_DOMAIN = b"tsfile-data-manifest-v1\0"
+_FILE_GENERATION_DOMAIN = b"tsfile-file-generation-v1\0"
+_DIGEST_HEX_LENGTH = 64
+_INDEX_FINGERPRINT_HEX_LENGTH = 16
+_DEFAULT_CHUNK_SIZE = 8 * 1024 * 1024
+
+
+class AttestationMismatchError(RuntimeError):
+ """Raised before reading when trusted metadata no longer matches."""
+
+
+@dataclass(frozen=True)
+class IndexAttestation:
+ schema_version: int
+ canonical_index_path: str
+ index_identity_sha256: str
+ index_size: int
+ st_dev: int
+ st_ino: int
+ st_size: int
+ st_mtime_ns: int
+
+
+@dataclass(frozen=True)
+class DataFileAttestation:
+ file_id: int
+ canonical_path: str
+ st_dev: int
+ st_ino: int
+ st_size: int
+ st_mtime_ns: int
+ existing_index_fingerprint: str | None = None
+
+
+def _canonical_path(path: os.PathLike[str] | str) -> str:
+ return os.path.realpath(os.path.abspath(os.fspath(path)))
+
+
+def _stat_tuple(stat_result: os.stat_result) -> tuple[int, int, int, int]:
+ return (
+ stat_result.st_dev,
+ stat_result.st_ino,
+ stat_result.st_size,
+ stat_result.st_mtime_ns,
+ )
+
+
+def _digest_bytes(value: str) -> bytes:
+ if (
+ len(value) != _DIGEST_HEX_LENGTH
+ or value != value.lower()
+ or any(character not in "0123456789abcdef" for character in value)
+ ):
+ raise ValueError("identity must be a lowercase SHA-256 hex digest")
+ return bytes.fromhex(value)
+
+
+def _optional_index_fingerprint_bytes(value: str | None) -> bytes:
+ if value is None:
+ return b""
+ if (
+ not isinstance(value, str)
+ or len(value) != _INDEX_FINGERPRINT_HEX_LENGTH
+ or value != value.lower()
+ or any(character not in "0123456789abcdef" for character in value)
+ ):
+ raise ValueError(
+ "existing_index_fingerprint must be a 16-character lowercase hex value"
+ )
+ return value.encode("ascii")
+
+
+def _u64(value: int, field: str) -> bytes:
+ if not isinstance(value, int) or isinstance(value, bool) or not 0 <= value < 2**64:
+ raise ValueError(f"{field} must fit uint64")
+ return struct.pack("<Q", value)
+
+
+def _field(value: bytes) -> bytes:
+ return _u64(len(value), "field length") + value
+
+
+def _update_stream(digest: "hashlib._Hash", stream: BinaryIO, chunk_size: int) -> None:
+ while chunk := stream.read(chunk_size):
+ digest.update(chunk)
+
+
+def attest_index(
+ path: os.PathLike[str] | str, *, chunk_size: int = _DEFAULT_CHUNK_SIZE
+) -> IndexAttestation:
+ """Hash finalized index bytes once in the trusted parent attestation pass."""
+ if chunk_size <= 0:
+ raise ValueError("chunk_size must be positive")
+ canonical_path = _canonical_path(path)
+ digest = hashlib.sha256()
+ with open(canonical_path, "rb") as stream:
+ before_hash = os.fstat(stream.fileno())
+ _update_stream(digest, stream, chunk_size)
+ after_hash = os.fstat(stream.fileno())
+ if _stat_tuple(before_hash) != _stat_tuple(after_hash):
+ raise AttestationMismatchError("index changed while hashing")
+ return IndexAttestation(
+ schema_version=_INDEX_SCHEMA_VERSION,
+ canonical_index_path=canonical_path,
+ index_identity_sha256=digest.hexdigest(),
+ index_size=after_hash.st_size,
+ st_dev=after_hash.st_dev,
+ st_ino=after_hash.st_ino,
+ st_size=after_hash.st_size,
+ st_mtime_ns=after_hash.st_mtime_ns,
+ )
+
+
+def validate_index_attestation(
+ path: os.PathLike[str] | str, attestation: IndexAttestation
+) -> None:
+ """Validate with one metadata lookup; never scan or re-hash index bytes."""
+ canonical_path = _canonical_path(path)
+ if attestation.schema_version != _INDEX_SCHEMA_VERSION:
+ raise AttestationMismatchError("unsupported index attestation schema")
+ _digest_bytes(attestation.index_identity_sha256)
+ if canonical_path != attestation.canonical_index_path:
+ raise AttestationMismatchError("index path does not match attestation")
+ stat_result = os.stat(canonical_path)
+ expected = (
+ attestation.st_dev,
+ attestation.st_ino,
+ attestation.st_size,
+ attestation.st_mtime_ns,
+ )
+ if (
+ attestation.index_size != attestation.st_size
+ or _stat_tuple(stat_result) != expected
+ ):
+ raise AttestationMismatchError("index stat does not match attestation")
+
+
+def attest_data_file(
+ file_id: int,
+ path: os.PathLike[str] | str,
+ *,
+ existing_index_fingerprint: str | None = None,
+) -> DataFileAttestation:
+ canonical_path = _canonical_path(path)
+ _u64(file_id, "file_id")
+ _optional_index_fingerprint_bytes(existing_index_fingerprint)
+ with open(canonical_path, "rb") as stream:
+ stat_result = os.fstat(stream.fileno())
+ return DataFileAttestation(
+ file_id=file_id,
+ canonical_path=canonical_path,
+ st_dev=stat_result.st_dev,
+ st_ino=stat_result.st_ino,
+ st_size=stat_result.st_size,
+ st_mtime_ns=stat_result.st_mtime_ns,
+ existing_index_fingerprint=existing_index_fingerprint,
+ )
+
+
+def validate_data_file_attestation(
+ path: os.PathLike[str] | str, attestation: DataFileAttestation
+) -> None:
+ canonical_path = _canonical_path(path)
+ if canonical_path != attestation.canonical_path:
+ raise AttestationMismatchError("data-file path does not match attestation")
+ stat_result = os.stat(canonical_path)
+ expected = (
+ attestation.st_dev,
+ attestation.st_ino,
+ attestation.st_size,
+ attestation.st_mtime_ns,
+ )
+ if _stat_tuple(stat_result) != expected:
+ raise AttestationMismatchError("data-file stat does not match attestation")
+
+
+def _canonical_data_file(attestation: DataFileAttestation) -> bytes:
+ encoded_fingerprint = _optional_index_fingerprint_bytes(
+ attestation.existing_index_fingerprint
+ )
+ return b"".join(
+ (
+ _field(attestation.canonical_path.encode("utf-8")),
+ _u64(attestation.st_dev, "st_dev"),
+ _u64(attestation.st_ino, "st_ino"),
+ _u64(attestation.st_size, "st_size"),
+ _u64(attestation.st_mtime_ns, "st_mtime_ns"),
+ _field(encoded_fingerprint),
+ )
+ )
+
+
+def data_manifest_identity(
+ attestations: Iterable[DataFileAttestation], *, require_sorted: bool = False
+) -> str:
+ files = tuple(attestations)
+ if require_sorted and any(
+ left.file_id >= right.file_id for left, right in zip(files, files[1:])
+ ):
+ raise ValueError("data-file IDs must be strictly increasing")
+ digest = hashlib.sha256(_DATA_MANIFEST_DOMAIN)
+ digest.update(_u64(len(files), "file count"))
+ for attestation in files:
+ digest.update(_u64(attestation.file_id, "file_id"))
+ digest.update(_field(_canonical_data_file(attestation)))
+ return digest.hexdigest()
+
+
+def file_generation_token(
+ index_identity_sha256: str,
+ data_manifest_identity_sha256: str,
+ attestation: DataFileAttestation,
+) -> str:
+ digest = hashlib.sha256(_FILE_GENERATION_DOMAIN)
+ digest.update(_digest_bytes(index_identity_sha256))
+ digest.update(_digest_bytes(data_manifest_identity_sha256))
+ digest.update(_u64(attestation.file_id, "file_id"))
+ digest.update(_field(_canonical_data_file(attestation)))
+ return digest.hexdigest()
diff --git a/python/tsfile/dataset/runtime.py b/python/tsfile/dataset/runtime.py
index 7d62f66c3..44560c3cc 100644
--- a/python/tsfile/dataset/runtime.py
+++ b/python/tsfile/dataset/runtime.py
@@ -47,6 +47,15 @@ from .index import (
MappedDatasetIndex,
file_fingerprint,
)
+from .index_identity import (
+ AttestationMismatchError,
+ DataFileAttestation,
+ IndexAttestation,
+ data_manifest_identity,
+ file_generation_token,
+ validate_data_file_attestation,
+ validate_index_attestation,
+)
from .metadata import (
MODEL_TABLE,
MODEL_TREE,
@@ -59,6 +68,10 @@ from .merge import build_aligned_matrix
_SERIES_DESCRIPTOR_CACHE_SIZE = 4096
+def _digest_u64(value: str) -> int:
+ return int.from_bytes(bytes.fromhex(value)[:8], byteorder="little")
+
+
def _configured_non_negative_int(name, default):
raw = os.environ.get(name)
if raw is None:
@@ -240,6 +253,7 @@ class _ReaderSession:
expected_size: int,
fingerprint: int,
validate_generation: bool = True,
+ data_file_attestation: Optional[DataFileAttestation] = None,
):
self.file_id = file_id
self.path = path
@@ -248,6 +262,12 @@ class _ReaderSession:
self._validate_generation_enabled = validate_generation
if validate_generation:
self._validate_generation()
+ if data_file_attestation is not None:
+ if data_file_attestation.file_id != file_id:
+ raise AttestationMismatchError(
+ "data-file attestation has the wrong file_id"
+ )
+ validate_data_file_attestation(path, data_file_attestation)
self.reader = TsFileReaderPy(path)
self.active_uses = 0
self._closed = False
@@ -278,10 +298,12 @@ class ReaderSessionPool:
index: MappedDatasetIndex,
max_open_files: int,
validate_generation: bool = True,
+ data_file_attestations: Optional[Mapping[int, DataFileAttestation]] = None,
):
self._index = index
self.max_open_files = max(1, int(max_open_files))
self._validate_generation = bool(validate_generation)
+ self._data_file_attestations = dict(data_file_attestations or {})
self._sessions: "OrderedDict[int, _ReaderSession]" = OrderedDict()
self._condition = threading.Condition()
self._closed = False
@@ -294,6 +316,7 @@ class ReaderSessionPool:
record[2],
record[3],
validate_generation=self._validate_generation,
+ data_file_attestation=self._data_file_attestations.get(file_id),
)
@contextlib.contextmanager
@@ -378,9 +401,15 @@ class PreparedSeriesCache:
index: MappedDatasetIndex,
validate_references: bool = True,
max_entries: Optional[int] = None,
+ mapped_index_identity: Optional[int] = None,
+ file_generation_tokens: Optional[Mapping[int, str]] = None,
):
self._index = index
self._validate_references = bool(validate_references)
+ self._mapped_index_identity = (
+ id(index) if mapped_index_identity is None else int(mapped_index_identity)
+ )
+ self._file_generation_tokens = dict(file_generation_tokens or {})
maximum = _SERIES_DESCRIPTOR_CACHE_SIZE if max_entries is None else max_entries
if int(maximum) < 0:
raise ValueError("max_entries must be non-negative")
@@ -398,7 +427,7 @@ class PreparedSeriesCache:
if self._validate_references and device_span[1] != file_id:
raise ValueError("series locator points at another TsFile")
return (
- id(self._index),
+ self._mapped_index_identity,
file_id,
file_record[2],
file_record[3],
@@ -464,7 +493,16 @@ class PreparedSeriesCache:
@contextlib.contextmanager
def acquire(self, file_id, locator_id, reader, time_owner=None):
- key = (id(self._index), file_id, locator_id)
+ generation = self._file_generation_tokens.get(file_id)
+ if generation is None:
+ file_record = self._index.record(TSFILE_RECORD, file_id)
+ generation = (
+ self._mapped_index_identity,
+ file_id,
+ file_record[2],
+ file_record[3],
+ )
+ key = (generation, locator_id)
entry = None
owner_entry = None
prepared_time_owner = time_owner
@@ -585,8 +623,27 @@ class DatasetRuntime:
query_workers: Optional[int] = None,
query_parallel_min_rows: Optional[int] = None,
trust_index: bool = False,
+ index_attestation: Optional[IndexAttestation] = None,
+ data_file_attestations: Optional[Sequence[DataFileAttestation]] = None,
):
self.trust_index = bool(trust_index)
+ has_index_attestation = index_attestation is not None
+ has_data_attestations = data_file_attestations is not None
+ if has_index_attestation != has_data_attestations:
+ raise ValueError(
+ "index_attestation and data_file_attestations must be provided together"
+ )
+ if has_index_attestation and not self.trust_index:
+ raise ValueError("attestations require trust_index=True")
+ if has_index_attestation and not isinstance(
+ index_attestation, IndexAttestation
+ ):
+ raise TypeError("index_attestation must be an IndexAttestation")
+ self.index_attestation = index_attestation
+ self.data_file_attestations = tuple(data_file_attestations or ())
+ self.data_file_attestations_by_id = {}
+ self.data_manifest_identity_sha256 = None
+ self.file_generation_tokens = {}
maximum = (
int(os.environ.get("TSFILE_DATAFRAME_MAX_OPEN_FILES", "16"))
if max_open_files is None
@@ -627,7 +684,21 @@ class DatasetRuntime:
self._torn_down = False
try:
+ if self.index_attestation is not None:
+ validate_index_attestation(self._path, self.index_attestation)
self.index = MappedDatasetIndex(self._path, trust_index=self.trust_index)
+ if self.index_attestation is not None:
+ expected_index_identity = (
+ self.index_attestation.st_dev,
+ self.index_attestation.st_ino,
+ self.index_attestation.st_size,
+ self.index_attestation.st_mtime_ns,
+ )
+ if self.index.identity != expected_index_identity:
+ raise AttestationMismatchError(
+ "mapped index does not match its attestation"
+ )
+ self._configure_attestations()
self._query_executor = (
ThreadPoolExecutor(
max_workers=self.query_workers,
@@ -640,11 +711,18 @@ class DatasetRuntime:
self.index,
self.max_open_files,
validate_generation=not self.trust_index,
+ data_file_attestations=self.data_file_attestations_by_id,
)
self.prepared = PreparedSeriesCache(
self.index,
validate_references=not self.trust_index,
max_entries=prepared_max_entries,
+ mapped_index_identity=(
+ _digest_u64(self.index_attestation.index_identity_sha256)
+ if self.index_attestation is not None
+ else None
+ ),
+ file_generation_tokens=self.file_generation_tokens,
)
self.catalog = MappedDataFrameCatalog(self)
except BaseException:
@@ -654,6 +732,60 @@ class DatasetRuntime:
pass
raise
+ def _configure_attestations(self):
+ if self.index_attestation is None:
+ return
+ if any(
+ not isinstance(attestation, DataFileAttestation)
+ for attestation in self.data_file_attestations
+ ):
+ raise TypeError(
+ "data_file_attestations must contain DataFileAttestation values"
+ )
+ expected_file_ids = tuple(range(self.index.count(TSFILE_RECORD)))
+ actual_file_ids = tuple(
+ attestation.file_id for attestation in self.data_file_attestations
+ )
+ if actual_file_ids != expected_file_ids:
+ raise ValueError(
+ "data_file_attestations must cover file_id=0..N-1 in order"
+ )
+
+ for attestation in self.data_file_attestations:
+ record = self.index.record(TSFILE_RECORD, attestation.file_id)
+ indexed_path = os.path.realpath(
+ os.path.abspath(self.index.string(record[0]))
+ )
+ if (
+ indexed_path != attestation.canonical_path
+ or record[2] != attestation.st_size
+ ):
+ raise AttestationMismatchError(
+ "data-file attestation does not match the Dataset Index"
+ )
+ expected_fingerprint = f"{record[3]:016x}"
+ if attestation.existing_index_fingerprint != expected_fingerprint:
+ raise AttestationMismatchError(
+ "data-file fingerprint does not match the Dataset Index"
+ )
+
+ self.data_file_attestations_by_id = {
+ attestation.file_id: attestation
+ for attestation in self.data_file_attestations
+ }
+ self.data_manifest_identity_sha256 = data_manifest_identity(
+ self.data_file_attestations,
+ require_sorted=True,
+ )
+ self.file_generation_tokens = {
+ attestation.file_id: file_generation_token(
+ self.index_attestation.index_identity_sha256,
+ self.data_manifest_identity_sha256,
+ attestation,
+ )
+ for attestation in self.data_file_attestations
+ }
+
def _close_resources(self):
if self._resources_closed:
return
@@ -681,6 +813,12 @@ class DatasetRuntime:
query_workers=self.query_workers,
query_parallel_min_rows=self.query_parallel_min_rows,
trust_index=self.trust_index,
+ index_attestation=self.index_attestation,
+ data_file_attestations=(
+ self.data_file_attestations
+ if self.index_attestation is not None
+ else None
+ ),
)
def _assert_current_process(self):