fsspeckit.core.maintenance API Reference

maintenance

Backend-neutral maintenance layer for parquet dataset operations.

This module provides shared functionality for dataset discovery, statistics, and grouping algorithms used by both DuckDB and PyArrow maintenance operations. It serves as the authoritative implementation for maintenance planning, ensuring consistent behavior across different backends.

Key responsibilities: 1. Dataset discovery and file-level statistics 2. Compaction grouping algorithms with streaming execution 3. Optimization planning with z-order validation 4. Canonical statistics structures 5. Partition filtering and edge case handling

Architecture: - Functions accept both dict format (legacy) and FileInfo objects for backward compatibility - All planning functions return structured results with canonical MaintenanceStats - Backend implementations delegate to this core for consistent behavior - Streaming design avoids materializing entire datasets in memory

Core components: - FileInfo: Canonical file information with validation - MaintenanceStats: Canonical statistics structure across backends - CompactionGroup: Logical grouping of files for processing - collect_dataset_stats: Dataset discovery with partition filtering - plan_compaction_groups: Shared compaction planning algorithm - plan_optimize_groups: Shared optimization planning with z-order validation

Usage: Backend functions should delegate to this module rather than implementing their own discovery and planning logic. This ensures that DuckDB and PyArrow produce identical grouping decisions and statistics structures.

Classes

fsspeckit.core.maintenance.DatasetMaintenanceCoordinator

DatasetMaintenanceCoordinator(
    backend: str | MaintenanceBackend,
)

Direct coordinator that creates immutable, backend-pinned maintenance plans.

Planning is lock-free and does not modify dataset files. Execution implements the full staged-rename lifecycle for atomic_local :class:CompactionPlan (flat local datasets, issue #37) and the staged-copy lifecycle for best_effort_object_store :class:CompactionPlan (issue #39). Other plan types are left as typed seams with descriptive NotImplementedError messages.

Source code in src/fsspeckit/core/maintenance.py

def __init__(self, backend: str | MaintenanceBackend) -> None:
    self.backend = _coerce_backend(backend)

Functions

fsspeckit.core.maintenance.DatasetMaintenanceCoordinator.execute

execute(
    plan: MaintenancePlan,
    filesystem: AbstractFileSystem | None = None,
    lock_timeout_s: float = 30.0,
    lock_retry_interval_s: float = 0.05,
) -> MaintenanceResult

Execute an accepted maintenance plan and return a typed result.

Implemented paths:

• atomic_local :class:CompactionPlan — full staged-rename lifecycle with advisory locking and partition-subtree preservation.
• atomic_local :class:PartitionLocalDeduplicationPlan — partition-local staged-rename deduplication with rollback.
• atomic_local :class:GlobalRepartitionDeduplicationPlan — global cross-partition staged-rename deduplication with rollback (#42).
• best_effort_object_store :class:CompactionPlan — staged-copy lifecycle with per-key validation, source-drift revalidation, and recovery artifact reporting. filesystem is required for this path.
• best_effort_object_store :class:PartitionLocalDeduplicationPlan — staged-copy deduplication with per-key validation and source-drift revalidation.
• best_effort_object_store :class:GlobalRepartitionDeduplicationPlan — staged-copy global repartitioning deduplication.
• atomic_local :class:CoordinatedOptimizationPlan — optional partition-local deduplication followed by compaction through one staged, validated, locked, and rollback-capable publication.
• best_effort_object_store :class:CoordinatedOptimizationPlan — optional partition-local deduplication followed by compaction, reported as separate phases. filesystem is required for this path.

Parameters:

Name	Type	Description	Default
plan	MaintenancePlan	An accepted plan created by one of the plan_* methods.	required
filesystem	AbstractFileSystem | None	Required for best_effort_object_store plans. Ignored for atomic_local plans.	None
lock_timeout_s	float	Maximum seconds to wait for the publication lock (atomic_local only).	30.0
lock_retry_interval_s	float	Sleep between lock-acquisition retries (atomic_local only).	0.05

Returns:

Name	Type	Description
A	MaintenanceResult	class:MaintenanceResult (or :class:BestEffortCompactionResult
	MaintenanceResult	subclass) carrying per-phase outcomes, validation, publication,
	MaintenanceResult	recovery artifacts, and actual metrics.

Raises:

Type	Description
NotImplementedError	For plan types or guarantee levels not yet implemented in this release.
ValueError	When filesystem is not provided for a best_effort_object_store plan.

Source code in src/fsspeckit/core/maintenance.py

def execute(
    self,
    plan: MaintenancePlan,
    filesystem: AbstractFileSystem | None = None,
    lock_timeout_s: float = 30.0,
    lock_retry_interval_s: float = 0.05,
) -> MaintenanceResult:
    """Execute an accepted maintenance plan and return a typed result.

    Implemented paths:

    - ``atomic_local`` :class:`CompactionPlan` — full staged-rename
      lifecycle with advisory locking and partition-subtree preservation.
    - ``atomic_local`` :class:`PartitionLocalDeduplicationPlan` —
      partition-local staged-rename deduplication with rollback.
    - ``atomic_local`` :class:`GlobalRepartitionDeduplicationPlan` —
      global cross-partition staged-rename deduplication with rollback
      (#42).
    - ``best_effort_object_store`` :class:`CompactionPlan` — staged-copy
      lifecycle with per-key validation, source-drift revalidation, and
      recovery artifact reporting.  *filesystem* is **required** for this
      path.
    - ``best_effort_object_store`` :class:`PartitionLocalDeduplicationPlan`
      — staged-copy deduplication with per-key validation and source-drift
      revalidation.
    - ``best_effort_object_store`` :class:`GlobalRepartitionDeduplicationPlan`
      — staged-copy global repartitioning deduplication.
    - ``atomic_local`` :class:`CoordinatedOptimizationPlan` — optional
      partition-local deduplication followed by compaction through one
      staged, validated, locked, and rollback-capable publication.
    - ``best_effort_object_store`` :class:`CoordinatedOptimizationPlan`
      — optional partition-local deduplication followed by compaction,
      reported as separate phases.  *filesystem* is **required** for this
      path.

    Args:
        plan: An accepted plan created by one of the ``plan_*`` methods.
        filesystem: Required for ``best_effort_object_store`` plans.
            Ignored for ``atomic_local`` plans.
        lock_timeout_s: Maximum seconds to wait for the publication lock
            (``atomic_local`` only).
        lock_retry_interval_s: Sleep between lock-acquisition retries
            (``atomic_local`` only).

    Returns:
        A :class:`MaintenanceResult` (or :class:`BestEffortCompactionResult`
        subclass) carrying per-phase outcomes, validation, publication,
        recovery artifacts, and actual metrics.

    Raises:
        NotImplementedError: For plan types or guarantee levels not yet
            implemented in this release.
        ValueError: When *filesystem* is not provided for a
            ``best_effort_object_store`` plan.
    """
    if plan.guarantee_level == GuaranteeLevel.ATOMIC_LOCAL and isinstance(
        plan, PartitionLocalDeduplicationPlan
    ):
        return _execute_atomic_local_partition_local_deduplication(
            plan,
            lock_timeout_s=lock_timeout_s,
            lock_retry_interval_s=lock_retry_interval_s,
        )

    if plan.guarantee_level == GuaranteeLevel.ATOMIC_LOCAL and isinstance(
        plan, GlobalRepartitionDeduplicationPlan
    ):
        return _execute_atomic_local_global_repartition_deduplication(
            plan,
            lock_timeout_s=lock_timeout_s,
            lock_retry_interval_s=lock_retry_interval_s,
        )

    if plan.guarantee_level == GuaranteeLevel.ATOMIC_LOCAL and isinstance(
        plan, CoordinatedOptimizationPlan
    ):
        return _execute_atomic_local_coordinated_optimization(
            plan,
            lock_timeout_s=lock_timeout_s,
            lock_retry_interval_s=lock_retry_interval_s,
        )

    if plan.guarantee_level == GuaranteeLevel.ATOMIC_LOCAL and isinstance(
        plan, CompactionPlan
    ):
        return _execute_atomic_local_compaction(
            plan,
            lock_timeout_s=lock_timeout_s,
            lock_retry_interval_s=lock_retry_interval_s,
        )

    # ---------------------------------------------------------- #
    # best_effort_object_store compaction (#39)
    # ---------------------------------------------------------- #
    if (
        plan.guarantee_level == GuaranteeLevel.BEST_EFFORT_OBJECT_STORE
        and isinstance(plan, CompactionPlan)
    ):
        if filesystem is None:
            # Keep a NotImplementedError seam that callers can detect until
            # they update to pass the filesystem argument.  The "#39" token
            # lets existing tests and tooling identify the seam.
            raise NotImplementedError(
                "best_effort_object_store execute() requires a filesystem "
                "argument (pass the fsspec filesystem used to create the plan). "
                "See issue #39."
            )
        return _execute_best_effort_compaction(plan, filesystem)

    if (
        plan.guarantee_level == GuaranteeLevel.BEST_EFFORT_OBJECT_STORE
        and isinstance(plan, PartitionLocalDeduplicationPlan)
    ):
        if filesystem is None:
            raise ValueError(
                "best_effort_object_store deduplication requires the filesystem "
                "used to create the plan."
            )
        return _execute_best_effort_partition_local_deduplication(plan, filesystem)

    if (
        plan.guarantee_level == GuaranteeLevel.BEST_EFFORT_OBJECT_STORE
        and isinstance(plan, GlobalRepartitionDeduplicationPlan)
    ):
        if filesystem is None:
            raise ValueError(
                "best_effort_object_store deduplication requires the filesystem "
                "used to create the plan."
            )
        return _execute_best_effort_global_repartition_deduplication(
            plan, filesystem
        )

    # ---------------------------------------------------------- #
    # best_effort_object_store coordinated optimization (#45)
    # ---------------------------------------------------------- #
    if (
        plan.guarantee_level == GuaranteeLevel.BEST_EFFORT_OBJECT_STORE
        and isinstance(plan, CoordinatedOptimizationPlan)
    ):
        if filesystem is None:
            raise ValueError(
                "best_effort_object_store optimization requires the filesystem "
                "used to create the plan."
            )
        return _execute_best_effort_coordinated_optimization(plan, filesystem)

    # ---------------------------------------------------------- #
    # Seams for downstream issues — raise descriptive errors
    # ---------------------------------------------------------- #

    raise NotImplementedError(
        f"execute() is not implemented for plan type {type(plan).__name__!r} "
        f"with guarantee_level={plan.guarantee_level!r}."
    )

fsspeckit.core.maintenance.DatasetMaintenanceCoordinator.plan_compaction

plan_compaction(
    dataset_path: str,
    filesystem: AbstractFileSystem | None = None,
    target_mb_per_file: int | None = None,
    target_rows_per_file: int | None = None,
    partition_filter: list[str] | None = None,
    validation_level: ValidationLevel | str | None = None,
    codec: str | None = None,
) -> CompactionPlan

Create an immutable compaction plan without modifying files.

Source code in src/fsspeckit/core/maintenance.py

def plan_compaction(
    self,
    dataset_path: str,
    filesystem: AbstractFileSystem | None = None,
    target_mb_per_file: int | None = None,
    target_rows_per_file: int | None = None,
    partition_filter: list[str] | None = None,
    validation_level: ValidationLevel | str | None = None,
    codec: str | None = None,
) -> CompactionPlan:
    """Create an immutable compaction plan without modifying files."""
    if target_rows_per_file is not None and target_rows_per_file <= 0:
        raise ValueError("target_rows_per_file must be > 0")
    (
        fs,
        file_stats,
        snapshot,
        scope,
        schema_outcome,
        schema,
        selected_codec,
        target_byte_size,
        validation,
    ) = self._common_plan_inputs(
        MaintenanceOperation.COMPACTION,
        dataset_path,
        filesystem,
        partition_filter,
        target_mb_per_file,
        validation_level,
        codec,
    )
    # Partition-local planning (#38) is scoped to the atomic_local path,
    # whose per-subtree publisher preserves partition tuples. The
    # best_effort_object_store publisher (#39) flattens outputs and keeps
    # its own planning policy until its partition-subtree work lands.
    guarantee_level = _classify_guarantee(fs)
    if guarantee_level == GuaranteeLevel.ATOMIC_LOCAL:
        compaction_groups = _plan_partition_local_compaction_groups(
            file_stats,
            snapshot.dataset_path,
            target_mb_per_file,
            target_rows_per_file,
        )
    else:
        compaction_groups = tuple(
            plan_compaction_groups(
                file_stats, target_mb_per_file, target_rows_per_file
            )["groups"]
        )
    return CompactionPlan(
        source_snapshot=snapshot,
        selected_backend=self.backend,
        guarantee_level=guarantee_level,
        partition_scope=scope,
        schema_outcome=schema_outcome,
        selected_codec=selected_codec,
        max_rows_per_file=target_rows_per_file,
        target_byte_size=target_byte_size,
        validation_level=validation,
        schema=schema,
        compaction_groups=compaction_groups,
    )

fsspeckit.core.maintenance.DatasetMaintenanceCoordinator.plan_coordinated_optimization

plan_coordinated_optimization(
    dataset_path: str,
    filesystem: AbstractFileSystem | None = None,
    dedup_key_columns: list[str] | None = None,
    dedup_order_by: list[str] | None = None,
    target_mb_per_file: int | None = None,
    target_rows_per_file: int | None = None,
    partition_filter: list[str] | None = None,
    validation_level: ValidationLevel | str | None = None,
    codec: str | None = None,
) -> CoordinatedOptimizationPlan

Create an immutable coordinated optimization plan.

Coordinated optimization is defined as optional deduplication followed by compaction. No z-ordering, sorting, or implicit repartitioning is planned.

Source code in src/fsspeckit/core/maintenance.py

def plan_coordinated_optimization(
    self,
    dataset_path: str,
    filesystem: AbstractFileSystem | None = None,
    dedup_key_columns: list[str] | None = None,
    dedup_order_by: list[str] | None = None,
    target_mb_per_file: int | None = None,
    target_rows_per_file: int | None = None,
    partition_filter: list[str] | None = None,
    validation_level: ValidationLevel | str | None = None,
    codec: str | None = None,
) -> CoordinatedOptimizationPlan:
    """Create an immutable coordinated optimization plan.

    Coordinated optimization is defined as optional deduplication followed
    by compaction. No z-ordering, sorting, or implicit repartitioning is
    planned.
    """
    if target_rows_per_file is not None and target_rows_per_file <= 0:
        raise ValueError("target_rows_per_file must be > 0")
    (
        fs,
        file_stats,
        snapshot,
        scope,
        schema_outcome,
        schema,
        selected_codec,
        target_byte_size,
        validation,
    ) = self._common_plan_inputs(
        MaintenanceOperation.COORDINATED_OPTIMIZATION,
        dataset_path,
        filesystem,
        partition_filter,
        target_mb_per_file,
        validation_level,
        codec,
    )
    if dedup_key_columns is not None:
        normalized_key_columns, normalized_dedup_order_by = (
            validate_deduplication_inputs(dedup_key_columns, dedup_order_by)
        )
        groups = _plan_partition_local_deduplication_groups(
            file_stats, snapshot.dataset_path
        )
    else:
        normalized_key_columns = None
        normalized_dedup_order_by = None
        groups = _plan_partition_local_compaction_groups(
            file_stats,
            snapshot.dataset_path,
            target_mb_per_file,
            target_rows_per_file,
        )
    return CoordinatedOptimizationPlan(
        source_snapshot=snapshot,
        selected_backend=self.backend,
        guarantee_level=_classify_guarantee(fs),
        partition_scope=scope,
        schema_outcome=schema_outcome,
        selected_codec=selected_codec,
        max_rows_per_file=target_rows_per_file,
        target_byte_size=target_byte_size,
        validation_level=validation,
        schema=schema,
        dedup_key_columns=(
            tuple(normalized_key_columns) if normalized_key_columns else None
        ),
        dedup_order_by=(
            tuple(normalized_dedup_order_by) if normalized_dedup_order_by else None
        ),
        optimization_groups=groups,
    )

fsspeckit.core.maintenance.DatasetMaintenanceCoordinator.plan_global_repartition_deduplication

plan_global_repartition_deduplication(
    dataset_path: str,
    partition_columns: list[str],
    filesystem: AbstractFileSystem | None = None,
    key_columns: list[str] | None = None,
    dedup_order_by: list[str] | None = None,
    target_mb_per_file: int | None = None,
    target_rows_per_file: int | None = None,
    validation_level: ValidationLevel | str | None = None,
    codec: str | None = None,
) -> GlobalRepartitionDeduplicationPlan

Create an immutable global-repartitioning deduplication plan.

Source code in src/fsspeckit/core/maintenance.py

def plan_global_repartition_deduplication(
    self,
    dataset_path: str,
    partition_columns: list[str],
    filesystem: AbstractFileSystem | None = None,
    key_columns: list[str] | None = None,
    dedup_order_by: list[str] | None = None,
    target_mb_per_file: int | None = None,
    target_rows_per_file: int | None = None,
    validation_level: ValidationLevel | str | None = None,
    codec: str | None = None,
) -> GlobalRepartitionDeduplicationPlan:
    """Create an immutable global-repartitioning deduplication plan."""
    if target_rows_per_file is not None and target_rows_per_file <= 0:
        raise ValueError("target_rows_per_file must be > 0")
    if not partition_columns:
        raise ValueError("partition_columns must be a non-empty list")
    if any(
        not isinstance(column, str) or not column for column in partition_columns
    ):
        raise ValueError("partition_columns must contain non-empty strings")
    if len(set(partition_columns)) != len(partition_columns):
        raise ValueError("partition_columns must not contain duplicates")
    (
        fs,
        file_stats,
        snapshot,
        scope,
        schema_outcome,
        schema,
        selected_codec,
        target_byte_size,
        validation,
    ) = _prepare_plan_inputs(
        MaintenanceOperation.GLOBAL_REPARTITION_DEDUPLICATION,
        dataset_path,
        filesystem,
        None,
        partition_columns,
        target_mb_per_file,
        validation_level,
        codec,
    )
    if schema is None or any(
        column not in schema.names for column in partition_columns
    ):
        raise ValueError(
            "partition_columns must name columns present in the source schema"
        )
    normalized_key_columns, normalized_dedup_order_by = (
        validate_deduplication_inputs(key_columns, dedup_order_by)
    )
    # Global deduplication always includes every source file.  Output
    # chunking is applied after one complete-dataset winner selection.
    global_group = CompactionGroup(
        files=tuple(
            FileInfo(
                path=file_stat["path"],
                size_bytes=file_stat["size_bytes"],
                num_rows=file_stat["num_rows"],
            )
            for file_stat in file_stats
        )
    )
    groups = (global_group,)
    return GlobalRepartitionDeduplicationPlan(
        source_snapshot=snapshot,
        selected_backend=self.backend,
        guarantee_level=_classify_guarantee(fs),
        partition_scope=scope,
        schema_outcome=schema_outcome,
        selected_codec=selected_codec,
        max_rows_per_file=target_rows_per_file,
        target_byte_size=target_byte_size,
        validation_level=validation,
        schema=schema,
        partition_columns=tuple(partition_columns),
        dedup_key_columns=(
            tuple(normalized_key_columns) if normalized_key_columns else None
        ),
        dedup_order_by=(
            tuple(normalized_dedup_order_by) if normalized_dedup_order_by else None
        ),
        dedup_groups=groups,
    )

fsspeckit.core.maintenance.DatasetMaintenanceCoordinator.plan_partition_local_deduplication

plan_partition_local_deduplication(
    dataset_path: str,
    filesystem: AbstractFileSystem | None = None,
    key_columns: list[str] | None = None,
    dedup_order_by: list[str] | None = None,
    target_mb_per_file: int | None = None,
    target_rows_per_file: int | None = None,
    partition_filter: list[str] | None = None,
    validation_level: ValidationLevel | str | None = None,
    codec: str | None = None,
) -> PartitionLocalDeduplicationPlan

Create an immutable partition-local deduplication plan.

Source code in src/fsspeckit/core/maintenance.py

def plan_partition_local_deduplication(
    self,
    dataset_path: str,
    filesystem: AbstractFileSystem | None = None,
    key_columns: list[str] | None = None,
    dedup_order_by: list[str] | None = None,
    target_mb_per_file: int | None = None,
    target_rows_per_file: int | None = None,
    partition_filter: list[str] | None = None,
    validation_level: ValidationLevel | str | None = None,
    codec: str | None = None,
) -> PartitionLocalDeduplicationPlan:
    """Create an immutable partition-local deduplication plan."""
    if target_rows_per_file is not None and target_rows_per_file <= 0:
        raise ValueError("target_rows_per_file must be > 0")
    (
        fs,
        file_stats,
        snapshot,
        scope,
        schema_outcome,
        schema,
        selected_codec,
        target_byte_size,
        validation,
    ) = self._common_plan_inputs(
        MaintenanceOperation.PARTITION_LOCAL_DEDUPLICATION,
        dataset_path,
        filesystem,
        partition_filter,
        target_mb_per_file,
        validation_level,
        codec,
    )
    normalized_key_columns, normalized_dedup_order_by = (
        validate_deduplication_inputs(key_columns, dedup_order_by)
    )
    groups = _plan_partition_local_deduplication_groups(
        file_stats, snapshot.dataset_path
    )
    return PartitionLocalDeduplicationPlan(
        source_snapshot=snapshot,
        selected_backend=self.backend,
        guarantee_level=_classify_guarantee(fs),
        partition_scope=scope,
        schema_outcome=schema_outcome,
        selected_codec=selected_codec,
        max_rows_per_file=target_rows_per_file,
        target_byte_size=target_byte_size,
        validation_level=validation,
        schema=schema,
        dedup_key_columns=(
            tuple(normalized_key_columns) if normalized_key_columns else None
        ),
        dedup_order_by=(
            tuple(normalized_dedup_order_by) if normalized_dedup_order_by else None
        ),
        dedup_groups=groups,
    )

fsspeckit.core.maintenance.MaintenancePlan dataclass

MaintenancePlan(
    operation: MaintenanceOperation,
    source_snapshot: SourceSnapshot,
    selected_backend: str,
    guarantee_level: GuaranteeLevel,
    partition_scope: PartitionScope,
    schema_outcome: SchemaOutcome,
    selected_codec: str,
    max_rows_per_file: int | None,
    target_byte_size: int | None,
    validation_level: ValidationLevel,
    schema: Schema | None = None,
)

Base immutable maintenance plan.

Attributes:

Name	Type	Description
operation	MaintenanceOperation	The maintenance operation this plan represents.
source_snapshot	SourceSnapshot	Snapshot of the source dataset at plan time.
selected_backend	str	Backend pinned to this plan.
guarantee_level	GuaranteeLevel	Automatic guarantee classification.
partition_scope	PartitionScope	Partition scope of the operation.
schema_outcome	SchemaOutcome	Lossless schema-reconciliation outcome.
selected_codec	str	Selected compression codec.
max_rows_per_file	int | None	Hard upper bound on output rows per file.
target_byte_size	int | None	Advisory target for output bytes per file.
validation_level	ValidationLevel	Validation level requested for the operation.
schema	Schema | None	Optional captured source schema.

fsspeckit.core.maintenance.CompactionPlan dataclass

CompactionPlan(
    operation: MaintenanceOperation,
    source_snapshot: SourceSnapshot,
    selected_backend: str,
    guarantee_level: GuaranteeLevel,
    partition_scope: PartitionScope,
    schema_outcome: SchemaOutcome,
    selected_codec: str,
    max_rows_per_file: int | None,
    target_byte_size: int | None,
    validation_level: ValidationLevel,
    schema: Schema | None = None,
    compaction_groups: tuple[CompactionGroup, ...] = (),
)

Bases: MaintenancePlan

Immutable plan for a compaction operation.

fsspeckit.core.maintenance.PartitionLocalDeduplicationPlan dataclass

PartitionLocalDeduplicationPlan(
    operation: MaintenanceOperation,
    source_snapshot: SourceSnapshot,
    selected_backend: str,
    guarantee_level: GuaranteeLevel,
    partition_scope: PartitionScope,
    schema_outcome: SchemaOutcome,
    selected_codec: str,
    max_rows_per_file: int | None,
    target_byte_size: int | None,
    validation_level: ValidationLevel,
    schema: Schema | None = None,
    dedup_key_columns: tuple[str, ...] | None = None,
    dedup_order_by: tuple[str, ...] | None = None,
    dedup_groups: tuple[CompactionGroup, ...] = (),
)

Bases: MaintenancePlan

Immutable plan for partition-local deduplication.

fsspeckit.core.maintenance.GlobalRepartitionDeduplicationPlan dataclass

GlobalRepartitionDeduplicationPlan(
    operation: MaintenanceOperation,
    source_snapshot: SourceSnapshot,
    selected_backend: str,
    guarantee_level: GuaranteeLevel,
    partition_scope: PartitionScope,
    schema_outcome: SchemaOutcome,
    selected_codec: str,
    max_rows_per_file: int | None,
    target_byte_size: int | None,
    validation_level: ValidationLevel,
    schema: Schema | None = None,
    partition_columns: tuple[str, ...] = (),
    dedup_key_columns: tuple[str, ...] | None = None,
    dedup_order_by: tuple[str, ...] | None = None,
    dedup_groups: tuple[CompactionGroup, ...] = (),
)

Bases: MaintenancePlan

Immutable plan for global-repartitioning deduplication.

fsspeckit.core.maintenance.CoordinatedOptimizationPlan dataclass

CoordinatedOptimizationPlan(
    operation: MaintenanceOperation,
    source_snapshot: SourceSnapshot,
    selected_backend: str,
    guarantee_level: GuaranteeLevel,
    partition_scope: PartitionScope,
    schema_outcome: SchemaOutcome,
    selected_codec: str,
    max_rows_per_file: int | None,
    target_byte_size: int | None,
    validation_level: ValidationLevel,
    schema: Schema | None = None,
    dedup_key_columns: tuple[str, ...] | None = None,
    dedup_order_by: tuple[str, ...] | None = None,
    optimization_groups: tuple[CompactionGroup, ...] = (),
)

Bases: MaintenancePlan

Immutable plan for coordinated optimization (optional dedup + compaction).

fsspeckit.core.maintenance.MaintenanceResult dataclass

MaintenanceResult(
    plan: MaintenancePlan,
    succeeded: bool,
    guarantee_level: GuaranteeLevel,
    phase_outcomes: tuple[PhaseOutcome, ...],
    validation: ValidationOutcome | None = None,
    publication: PublicationOutcome | None = None,
    recovery: RecoveryArtifacts | None = None,
    actual_metrics: ActualMetrics | None = None,
    error: str | None = None,
)

Typed result returned by DatasetMaintenanceCoordinator.execute().

Attributes:

Name	Type	Description
plan	MaintenancePlan	The plan that was executed.
succeeded	bool	True when the operation completed successfully end-to-end.
guarantee_level	GuaranteeLevel	Guarantee level that governed this execution.
phase_outcomes	tuple[PhaseOutcome, ...]	Ordered per-phase outcomes (stage, write, validate, lock, drift_check, publish, cleanup).
validation	ValidationOutcome | None	Staged-output validation outcome, populated after the write phase.
publication	PublicationOutcome | None	Atomic-rename publication outcome.
recovery	RecoveryArtifacts | None	Recovery artifact details; workspace_path is None after a successful cleanup.
actual_metrics	ActualMetrics | None	Row/file/byte counts of the published output. Only populated on success.
error	str | None	Top-level error summary, or None on success.

fsspeckit.core.maintenance.GuaranteeLevel

Bases: str, Enum

Publication guarantee levels for maintenance plans.

fsspeckit.core.maintenance.ValidationLevel

Bases: str, Enum

Validation levels for maintenance plans.

fsspeckit.core.maintenance.MaintenanceBackend

Bases: str, Enum

Supported maintenance backends.

Functions

fsspeckit.core.maintenance.collect_dataset_stats

collect_dataset_stats(
    path: str,
    filesystem: AbstractFileSystem | None = None,
    partition_filter: list[str] | None = None,
) -> dict[str, Any]

Collect file-level statistics for a parquet dataset.

This function walks the given dataset directory on the provided filesystem, discovers parquet files (recursively), and returns basic statistics.

Parameters:

Name	Type	Description	Default
path	str	Root directory of the parquet dataset.	required
filesystem	AbstractFileSystem | None	Optional fsspec filesystem. If omitted, a local "file" filesystem is used.	None
partition_filter	list[str] | None	Optional list of partition prefix filters (e.g. ["date=2025-11-04"]). Only files whose path relative to path starts with one of these prefixes are included.	None

Returns:

Type	Description
dict[str, Any]	Dict with keys:
dict[str, Any]	files: list of {"path", "size_bytes", "num_rows"} dicts
dict[str, Any]	total_bytes: sum of file sizes
dict[str, Any]	total_rows: sum of row counts

Raises:

Type	Description
FileNotFoundError	If the path does not exist or no parquet files match the optional partition filter.

Source code in src/fsspeckit/core/maintenance.py

def collect_dataset_stats(
    path: str,
    filesystem: AbstractFileSystem | None = None,
    partition_filter: list[str] | None = None,
) -> dict[str, Any]:
    """
    Collect file-level statistics for a parquet dataset.

    This function walks the given dataset directory on the provided filesystem,
    discovers parquet files (recursively), and returns basic statistics.

    Args:
        path: Root directory of the parquet dataset.
        filesystem: Optional fsspec filesystem. If omitted, a local "file"
            filesystem is used.
        partition_filter: Optional list of partition prefix filters
            (e.g. ["date=2025-11-04"]). Only files whose path relative to
            ``path`` starts with one of these prefixes are included.

    Returns:
        Dict with keys:
        - ``files``: list of ``{"path", "size_bytes", "num_rows"}`` dicts
        - ``total_bytes``: sum of file sizes
        - ``total_rows``: sum of row counts

    Raises:
        FileNotFoundError: If the path does not exist or no parquet files
            match the optional partition filter.
    """
    import pyarrow.parquet as pq

    fs = filesystem or fsspec_filesystem("file")

    if not fs.exists(path):
        raise FileNotFoundError(f"Dataset path '{path}' does not exist")

    root = Path(path)

    # Discover parquet files recursively via a manual stack walk so we can
    # respect partition_filter prefixes on the logical relative path.
    files: list[str] = []
    stack: list[str] = [path]
    while stack:
        current_dir = stack.pop()
        try:
            entries = fs.ls(current_dir, detail=False)
        except (OSError, PermissionError) as e:
            logger.warning("Failed to list directory '%s': %s", current_dir, e)
            continue

        for entry in entries:
            if entry.endswith(".parquet"):
                files.append(entry)
            else:
                try:
                    if fs.isdir(entry):
                        stack.append(entry)
                except (OSError, PermissionError) as e:
                    logger.warning(
                        "Failed to check if entry '%s' is a directory: %s", entry, e
                    )
                    continue

    if partition_filter:
        normalized_filters = [p.rstrip("/") for p in partition_filter]
        filtered_files: list[str] = []
        for filename in files:
            rel = Path(filename).relative_to(root).as_posix()
            if any(rel.startswith(prefix) for prefix in normalized_filters):
                filtered_files.append(filename)
        files = filtered_files

    if not files:
        raise FileNotFoundError(
            f"No parquet files found under '{path}' matching filter"
        )

    file_infos: list[dict[str, Any]] = []
    total_bytes = 0
    total_rows = 0

    for filename in files:
        size_bytes = 0
        try:
            info = fs.info(filename)
            if isinstance(info, dict):
                size_bytes = int(info.get("size", 0))
        except (OSError, PermissionError) as e:
            logger.warning("Failed to get file info for '%s': %s", filename, e)
            size_bytes = 0

        num_rows = 0
        try:
            with fs.open(filename, "rb") as fh:
                pf = pq.ParquetFile(fh)
                num_rows = pf.metadata.num_rows
        except (OSError, PermissionError, RuntimeError, ValueError) as e:
            # As a fallback, attempt a minimal table read to estimate rows.
            logger.debug(
                "Failed to read parquet metadata from '%s', trying fallback: %s",
                filename,
                e,
            )
            try:
                with fs.open(filename, "rb") as fh:
                    table = pq.read_table(fh)
                num_rows = table.num_rows
            except (OSError, PermissionError, RuntimeError, ValueError) as e:
                logger.debug("Fallback table read failed for '%s': %s", filename, e)
                num_rows = 0

        total_bytes += size_bytes
        total_rows += num_rows
        file_infos.append(
            {"path": filename, "size_bytes": size_bytes, "num_rows": num_rows}
        )

    return {"files": file_infos, "total_bytes": total_bytes, "total_rows": total_rows}