feat: add validate_fragments for distributed fragment validation

daft_lance/__init__.py

@@ -14,6 +14,7 @@
     read_lance,
     update_columns,
     update_columns_df,
+    validate_fragments,
     write_lance,
 )
 from .lance_update_column import UpdateColumnsResult
@@ -31,5 +32,6 @@
     "take_blobs",
     "update_columns",
     "update_columns_df",
+    "validate_fragments",
     "write_lance",
 ]

daft_lance/_lance.py

@@ -15,6 +15,7 @@
 
 from .lance_compaction import compact_files_internal
 from .lance_data_sink import LanceDataSink
+from .lance_fragment_validation import validate_fragments_internal
 from .lance_merge_column import merge_columns_from_df, merge_columns_internal
 from .lance_scalar_index import create_scalar_index_internal, optimize_indices_internal
 from .lance_scan import LanceScanOperator
@@ -1098,6 +1099,92 @@ def optimize_indices(
     )
 
 
+@PublicAPI
+def validate_fragments(
+    uri: str | pathlib.Path | None = None,
+    io_config: IOConfig | None = None,
+    *,
+    table_id: list[str] | None = None,
+    namespace_impl: str | None = None,
+    namespace_properties: dict[str, str] | None = None,
+    storage_options: dict[str, Any] | None = None,
+    version: int | str | None = None,
+    asof: str | None = None,
+    block_size: int | None = None,
+    index_cache_size: int | None = None,
+    metadata_cache_size_bytes: int | None = None,
+    fragment_ids: Sequence[int] | None = None,
+    concurrency: int | None = None,
+) -> DataFrame:
+    """Validate selected Lance fragments without scanning table rows.
+
+    The driver resolves one dataset version and enumerates fragment metadata;
+    Daft workers reopen that pinned snapshot and call Lance's native
+    ``LanceFragment.validate()``, which checks fragment file and metadata
+    consistency, including deletion vectors. The operation is read-only: it
+    never commits, repairs, or changes the dataset version.
+
+    Args:
+        uri: The URI of the Lance table (supports remote URLs to object stores such as `s3://` or `gs://`)
+        io_config: A custom IOConfig to use when accessing Lance data. Defaults to None.
+        table_id: Table identifier within the namespace, e.g. ["catalog", "schema", "table"].
+            Mutually exclusive with ``uri``.
+        namespace_impl: Lance Namespace implementation, e.g. "dir" or "rest".
+        namespace_properties: Properties for connecting to the namespace, e.g.
+            {"root": "/data"} for "dir" or {"uri": "http://host:port"} for "rest".
+        storage_options: Extra options for storage connection.
+        version: If specified, validate a specific version of the Lance dataset.
+        asof: If specified, validate the latest version created on or earlier than the given value.
+        block_size: Block size in bytes. Provide a hint for the size of the minimal I/O request.
+        index_cache_size: Index cache size.
+        metadata_cache_size_bytes: Size of the metadata cache in bytes.
+        fragment_ids: Optional subset of fragment IDs to validate. Every worker
+            reopens the same pinned version, so validation only covers the
+            listed fragments. Duplicates are ignored; unknown IDs and an empty
+            list raise ``ValueError``. When None, every fragment in the pinned
+            manifest is validated.
+        concurrency: Maximum number of concurrent validation tasks. Defaults
+            to Daft's runner default. Must be a positive integer.
+
+    Returns:
+        DataFrame: One row per selected fragment with the columns
+            ``version`` (Int64), ``fragment_id`` (Int64), ``valid`` (Boolean),
+            and ``error`` (String). ``error`` is null for valid fragments;
+            otherwise it holds the validation exception. A table-open or
+            authentication failure on a worker is reported with a
+            ``failed to open the dataset at version ...`` prefix so it can be
+            distinguished from detected corruption.
+
+    Raises:
+        ValueError: If ``fragment_ids`` contains IDs that are not in the
+            pinned version, or if ``concurrency`` is not positive.
+
+    Note:
+        ``LanceFragment.validate`` requires PyLance >= 11; daft-lance already
+        requires a newer PyLance (>= 12).
+    """
+    dataset_handle = construct_lance_dataset_handle(
+        uri,
+        storage_options=storage_options,
+        io_config=io_config,
+        namespace_impl=namespace_impl,
+        namespace_properties=namespace_properties,
+        table_id=table_id,
+        version=version,
+        asof=asof,
+        block_size=block_size,
+        index_cache_size=index_cache_size,
+        metadata_cache_size_bytes=metadata_cache_size_bytes,
+    )
+
+    return validate_fragments_internal(
+        dataset_handle.dataset,
+        dataset_handle.worker_open_context(),
+        fragment_ids=fragment_ids,
+        concurrency=concurrency,
+    )
+
+
 @PublicAPI
 def compact_files(
     uri: str | pathlib.Path | None = None,

daft_lance/lance_fragment_validation.py

@@ -0,0 +1,119 @@
+from __future__ import annotations
+
+import logging
+from collections.abc import Sequence
+from typing import TYPE_CHECKING
+
+import lance
+import pyarrow as pa
+
+import daft
+
+if TYPE_CHECKING:
+    from daft_lance.namespace import DatasetOpenContext
+
+logger = logging.getLogger(__name__)
+
+#: One row per selected fragment; ``error`` is null when the fragment is valid.
+EMPTY_REPORT = pa.table(
+    {
+        "version": pa.array([], type=pa.int64()),
+        "fragment_id": pa.array([], type=pa.int64()),
+        "valid": pa.array([], type=pa.bool_()),
+        "error": pa.array([], type=pa.string()),
+    }
+)
+
+
+class FragmentValidationUDF:
+    """Validate Lance fragments on a Daft worker against a pinned snapshot.
+
+    The dataset is opened once per worker instance, not per fragment, so the
+    pinned-manifest read is amortised across the fragments that worker owns.
+    """
+
+    def __init__(self, open_context: DatasetOpenContext, version: int) -> None:
+        self.open_context = open_context
+        self.version = version
+        self._lance_ds: lance.LanceDataset | None = None
+
+    def _dataset(self) -> lance.LanceDataset:
+        if self._lance_ds is None:
+            self._lance_ds = self.open_context.open_pinned()
+        return self._lance_ds
+
+    def __call__(self, fragment_id: int) -> str | None:
+        try:
+            dataset = self._dataset()
+        except Exception as exc:
+            # A table-open/authentication failure is not evidence of corruption,
+            # so report it with a distinct prefix instead of a validation error.
+            return f"failed to open the dataset at version {self.version}: {type(exc).__name__}: {exc}"
+
+        try:
+            fragment = dataset.get_fragment(int(fragment_id))
+            if fragment is None:
+                return f"fragment {fragment_id} is not present in pinned version {self.version}"
+            fragment.validate()
+        except Exception as exc:
+            return f"{type(exc).__name__}: {exc}"
+        return None
+
+
+def _selected_fragment_ids(
+    available: list[int],
+    fragment_ids: Sequence[int] | None,
+    version: int,
+) -> list[int]:
+    if fragment_ids is None:
+        return available
+
+    requested = list(dict.fromkeys(fragment_ids))
+    if not requested:
+        raise ValueError("fragment_ids must not be empty; pass None to validate every fragment")
+    available_set = set(available)
+    unknown = [fragment_id for fragment_id in requested if fragment_id not in available_set]
+    if unknown:
+        raise ValueError(
+            f"Fragment IDs {unknown} do not exist in dataset version {version}. "
+            f"Available fragment IDs: {sorted(available_set)}."
+        )
+    return requested
+
+
+def validate_fragments_internal(
+    lance_ds: lance.LanceDataset,
+    open_context: DatasetOpenContext,
+    *,
+    fragment_ids: Sequence[int] | None = None,
+    concurrency: int | None = None,
+) -> daft.DataFrame:
+    """Validate selected fragments of the pinned snapshot ``open_context`` describes.
+
+    ``lance_ds`` is the driver's live dataset and is only used to enumerate
+    fragment metadata and resolve the version; ``open_context`` is what workers
+    reopen from.
+    """
+    if concurrency is not None and concurrency <= 0:
+        raise ValueError(f"concurrency must be positive, got {concurrency}")
+
+    version = lance_ds.version
+    # ``get_fragments`` reads fragment metadata from the pinned manifest; it
+    # does not scan rows, so fragments with no live rows are still listed.
+    available = [fragment.fragment_id for fragment in lance_ds.get_fragments()]
+    selected = _selected_fragment_ids(available, fragment_ids, version)
+    logger.info("Validating %d of %d fragments at dataset version %d", len(selected), len(available), version)
+
+    if not selected:
+        return daft.from_arrow(EMPTY_REPORT)
+
+    df = daft.from_pylist([{"fragment_id": fragment_id} for fragment_id in selected])
+    validate_fragment = daft.cls(FragmentValidationUDF, max_concurrency=concurrency)
+    validator = validate_fragment(open_context, version)
+    df = df.select(df["fragment_id"], validator(df["fragment_id"]).alias("error"))
+    return df.select(
+        daft.lit(version).alias("version"),
+        df["fragment_id"],
+        df["error"].is_null().alias("valid"),
+        df["error"],
+    )