Source code for deephaven_enterprise.checkpoint_control

#
# Copyright (c) 2016-2026 Deephaven Data Labs and Patent Pending
#
"""Examine and repair table checkpoint records from a Core+ Python worker.

Checkpoint records (``table.size`` files) describe the state of a table location on disk. This module reports on -
and optionally repairs - locations that are not directories, are empty directories, or whose checkpoint record is
missing or out of date. It is the scripting equivalent of ``dhconfig checkpoint repair``.

Nothing is modified unless ``force=True`` is passed; by default problems are only reported.

Simple usage::

    from deephaven_enterprise.checkpoint_control import repair_table

    result = repair_table("MyNamespace", "MyTable")
    print(result)

    if result.summary_status is not RepairStatus.OK:
        result = repair_table("MyNamespace", "MyTable", force=True)
        print(result)

Bulk usage::

    from deephaven_enterprise.checkpoint_control import repair_namespace_set, repair_all

    print(repair_namespace_set("User", force=True))
    print(repair_all())
"""

from enum import Enum
from functools import cached_property
from typing import Any, Iterable, Optional

import jpy
from deephaven import DHError
from deephaven._wrapper import JObjectWrapper
from deephaven.jcompat import j_collection_to_list

_JCheckpointControl: Any = jpy.get_type("io.deephaven.configuration.CheckpointControl")
_JRepairOptions: Any = jpy.get_type(
    "io.deephaven.configuration.CheckpointControl$RepairOptions"
)
_JCheckpointRepairResult: Any = jpy.get_type(
    "io.deephaven.configuration.CheckpointRepairResult"
)
_JLocationResult: Any = jpy.get_type(
    "io.deephaven.configuration.CheckpointRepairResult$LocationResult"
)
_JStatus: Any = jpy.get_type("io.deephaven.configuration.CheckpointRepairResult$Status")
_JIssue: Any = jpy.get_type("io.deephaven.configuration.CheckpointRepairResult$Issue")




class RepairStatus(Enum):
    """Outcome for a single table location. Also used as the summary across all reported locations, in which case it
    is the worst status reported."""



    OK = _JStatus.OK

    """No problem was found, or (as a summary) no problem was found anywhere."""



    FLAGGED = _JStatus.FLAGGED

    """A problem was found but not acted on, because ``force`` was not set."""



    REPAIRED = _JStatus.REPAIRED

    """A problem was found and repaired."""



    FAILED = _JStatus.FAILED

    """A problem was found, and the attempt to repair it failed."""

    def __repr__(self) -> str:
        return self.name





class RepairIssue(Enum):
    """The kind of problem found at a table location."""



    DEFINITION_UNAVAILABLE = _JIssue.DEFINITION_UNAVAILABLE

    """The table's schema could not be read, so none of its locations could be inspected."""



    NOT_A_DIRECTORY = _JIssue.NOT_A_DIRECTORY

    """The location exists but is not a directory; it should be deleted."""



    EMPTY = _JIssue.EMPTY

    """The location is an empty directory; it should be deleted."""



    MISSING_CHECKPOINT_RECORD = _JIssue.MISSING_CHECKPOINT_RECORD

    """The location has no checkpoint record; one should be generated."""



    OUTDATED_CHECKPOINT_RECORD = _JIssue.OUTDATED_CHECKPOINT_RECORD

    """The location's checkpoint record predates the current version; it should be upgraded."""

    def __repr__(self) -> str:
        return self.name





class LocationResult(JObjectWrapper):
    """The problem found at a single table location, and what was done about it."""

    j_object_type = _JLocationResult

    def __init__(self, j_location_result: jpy.JType):
        self.j_location_result = j_location_result

    @property
    def j_object(self) -> jpy.JType:
        return self.j_location_result

    @cached_property


    def namespace(self) -> str:
        """The namespace of the table that owns the location."""
        return self.j_location_result.namespace()


    @cached_property


    def table_name(self) -> str:
        """The name of the table that owns the location."""
        return self.j_location_result.tableName()


    @cached_property


    def path(self) -> Optional[str]:
        """The absolute path of the location, or ``None`` when the whole table could not be inspected."""
        return self.j_location_result.path()


    @cached_property


    def issue(self) -> RepairIssue:
        """The :class:`RepairIssue` that was found."""
        return RepairIssue(self.j_location_result.issue())


    @cached_property


    def status(self) -> RepairStatus:
        """The :class:`RepairStatus` describing what was done about the problem."""
        return RepairStatus(self.j_location_result.status())


    @cached_property


    def message(self) -> str:
        """A human-readable description of the problem and the action taken."""
        return self.j_location_result.message()


    def __str__(self) -> str:
        return self.j_location_result.toString()





class CheckpointRepairResult(JObjectWrapper):
    """Aggregate result of a checkpoint repair operation."""

    j_object_type = _JCheckpointRepairResult

    def __init__(self, j_checkpoint_repair_result: jpy.JType):
        self.j_checkpoint_repair_result = j_checkpoint_repair_result

    @property
    def j_object(self) -> jpy.JType:
        return self.j_checkpoint_repair_result

    @cached_property


    def summary_status(self) -> RepairStatus:
        """Summary :class:`RepairStatus` across all reported locations - the worst status reported, or
        :attr:`RepairStatus.OK` when no problems were found."""
        return RepairStatus(self.j_checkpoint_repair_result.getSummaryStatus())


    @cached_property


    def location_results(self) -> tuple[LocationResult, ...]:
        """The :class:`LocationResult` for every location that had a problem, in the order encountered. Locations
        without problems are not reported."""
        return tuple(
            LocationResult(j_location_result)
            for j_location_result in j_collection_to_list(
                self.j_checkpoint_repair_result.getLocationResults()
            )
        )


    @cached_property


    def tables_inspected(self) -> int:
        """The number of tables that were selected and inspected."""
        return self.j_checkpoint_repair_result.getTablesInspected()


    @cached_property


    def locations_inspected(self) -> int:
        """The number of table locations that were inspected, including those without problems."""
        return self.j_checkpoint_repair_result.getLocationsInspected()


    @cached_property


    def metadata_index_affected(self) -> bool:
        """``True`` if any reported location belongs to a System historical table, whose locations may be tracked by a
        metadata index. When ``True``,
        rebuild the index for the affected tables with ``dhctl metadata update`` once the locations are repaired."""
        return self.j_checkpoint_repair_result.isMetadataIndexAffected()


    @cached_property


    def force(self) -> bool:
        """``True`` if the operation was allowed to make changes."""
        return self.j_checkpoint_repair_result.isForce()


    def __str__(self) -> str:
        """Multi-line, human-readable summary mirroring the Java ``toString`` shape."""
        return self.j_checkpoint_repair_result.toString()





def repair(
    *,
    tables: Iterable[tuple[str, str]] = (),
    namespaces: Iterable[str] = (),
    namespace_sets: Iterable[str] = (),
    force: bool = False,
) -> CheckpointRepairResult:
    """Report on (and, with ``force=True``, repair) the checkpoint records of the selected tables.

    The three selections are unioned, and at least one of them must be non-empty.

    Args:
        tables: ``(namespace, table_name)`` pairs naming individual tables to inspect.
        namespaces: Namespaces whose every table should be inspected.
        namespace_sets: Namespace set names (``"System"`` or ``"User"``, case-insensitive) whose every namespace's
            every table should be inspected.
        force: If ``True``, perform the repairs. If ``False`` (the default), only report problems.

    Returns:
        The :class:`CheckpointRepairResult` reported by the underlying Java call.

    Raises:
        DHError: If nothing was selected, a selected namespace or table does not exist, or the underlying Java call
            raises.

    Example::

        result = repair(
            tables=[("MyNamespace", "MyTable")], namespaces=["OtherNamespace"], force=True
        )
        print(result)
    """
    try:
        j_builder = _JRepairOptions.builder().force(force)
        for namespace, table_name in tables:
            j_builder.addTable(namespace, table_name)
        for namespace in namespaces:
            j_builder.addNamespace(namespace)
        for namespace_set in namespace_sets:
            j_builder.addNamespaceSet(namespace_set)
        j_result = _JCheckpointControl.repair(j_builder.build())
    except Exception as e:
        raise DHError(e, "failed to repair checkpoint records") from e
    return CheckpointRepairResult(j_result)





def repair_table(
    namespace: str, table_name: str, *, force: bool = False
) -> CheckpointRepairResult:
    """Report on (and, with ``force=True``, repair) the checkpoint records of a single table.

    Args:
        namespace: The namespace of the table to inspect.
        table_name: The name of the table to inspect.
        force: If ``True``, perform the repairs. If ``False`` (the default), only report problems.

    Returns:
        The :class:`CheckpointRepairResult` reported by the underlying Java call.

    Raises:
        DHError: If the table does not exist, or the underlying Java call raises.

    Example::

        result = repair_table("MyNamespace", "MyTable", force=True)
        print(result)
    """
    return repair(tables=[(namespace, table_name)], force=force)





def repair_namespace(namespace: str, *, force: bool = False) -> CheckpointRepairResult:
    """Report on (and, with ``force=True``, repair) the checkpoint records of every table in a namespace.

    Args:
        namespace: The namespace whose tables should be inspected.
        force: If ``True``, perform the repairs. If ``False`` (the default), only report problems.

    Returns:
        The :class:`CheckpointRepairResult` reported by the underlying Java call.

    Raises:
        DHError: If the namespace does not exist or has no tables, or the underlying Java call raises.

    Example::

        result = repair_namespace("MyNamespace")
        print(result)
    """
    return repair(namespaces=[namespace], force=force)





def repair_namespace_set(
    namespace_set: str, *, force: bool = False
) -> CheckpointRepairResult:
    """Report on (and, with ``force=True``, repair) the checkpoint records of every table in a namespace set.

    Args:
        namespace_set: The namespace set name, ``"System"`` or ``"User"``, case-insensitive.
        force: If ``True``, perform the repairs. If ``False`` (the default), only report problems.

    Returns:
        The :class:`CheckpointRepairResult` reported by the underlying Java call.

    Raises:
        DHError: If the name does not match a namespace set, or the underlying Java call raises.

    Example::

        result = repair_namespace_set("User", force=True)
        print(result)
    """
    return repair(namespace_sets=[namespace_set], force=force)





def repair_all(*, force: bool = False) -> CheckpointRepairResult:
    """Report on (and, with ``force=True``, repair) the checkpoint records of every table in every namespace.

    Args:
        force: If ``True``, perform the repairs. If ``False`` (the default), only report problems.

    Returns:
        The :class:`CheckpointRepairResult` reported by the underlying Java call.

    Raises:
        DHError: If the underlying Java call raises.

    Example::

        result = repair_all()
        print(result)
    """
    return repair(namespace_sets=["System", "User"], force=force)