include/paimon/file_store_commit.h

/*
 * 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.
 */

#pragma once

#include <cstdint>
#include <map>
#include <memory>
#include <optional>
#include <string>
#include <vector>

#include "paimon/defs.h"
#include "paimon/executor.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/metrics.h"
#include "paimon/realtime/realtime_commit_progress.h"
#include "paimon/result.h"
#include "paimon/status.h"
#include "paimon/type_fwd.h"
#include "paimon/visibility.h"

namespace paimon {
class CommitContext;
class CommitMessage;

/// Interface for commit operations in a file store.
///
/// The `FileStoreCommit` class provides interfaces for committing changes, expiring old snapshots,
/// dropping partitions, and retrieving commit metrics.
///
/// @note Direct file-system commits support append-only and primary-key tables on non-object-store
/// paths. `CommitContextBuilder::WithCatalog()` also supports object stores when the catalog
/// manages versions. `UseRESTCatalogCommit()` only prepares a request for the caller to send.
class PAIMON_EXPORT FileStoreCommit {
 public:
    /// Create an instance of `FileStoreCommit`.
    ///
    /// @param context A unique pointer to the `CommitContext` used for commit operations.
    ///
    /// @return A Result containing a unique pointer to the `FileStoreCommit` instance.
    static Result<std::unique_ptr<FileStoreCommit>> Create(std::unique_ptr<CommitContext> context);

    virtual ~FileStoreCommit() = default;

    /// Commit changes to the file store.
    ///
    /// @param commit_messages A vector of commit messages to be committed.
    /// @param commit_identifier An optional identifier for the commit operation. Default is
    /// `BATCH_WRITE_COMMIT_IDENTIFIER`.
    /// @param watermark An optional event-time watermark used to indicate the progress of data
    ///     processing. Default is std::nullopt.
    /// @return Status indicating the success or failure of the commit operation.
    virtual Status Commit(const std::vector<std::shared_ptr<CommitMessage>>& commit_messages,
                          int64_t commit_identifier = BATCH_WRITE_COMMIT_IDENTIFIER,
                          std::optional<int64_t> watermark = std::nullopt) = 0;

    /// Commit sealed real-time segments and persist their partition-bucket offset progress.
    ///
    /// Entries for each partition-bucket must be ordered and non-overlapping after the offset
    /// recorded by the latest committed snapshot. Input entries may be unordered; this method
    /// orders them by partition, bucket, and offset before validation. Offset gaps are allowed.
    /// The resulting snapshot atomically publishes the data files and the updated offset map.
    ///
    /// For each partition-bucket, the upstream coordinator must include the complete prefix of
    /// prepared-but-uncommitted entries through the requested progress. Because offsets may be
    /// sparse, this method cannot distinguish a valid offset gap from an omitted prepared entry.
    /// Omitting an earlier entry may advance committed progress past unpublished files and allow
    /// their real-time data to be reclaimed.
    ///
    /// Snapshot conflicts are retried internally using the configured commit retry limit, timeout,
    /// and backoff. Each attempt reloads the latest snapshot and rebases both file changes and
    /// offset progress. A retry succeeds idempotently when both the identifier and all requested
    /// ranges are already committed. Inconsistent identifiers, overlapping offset progress, and
    /// file or index conflicts fail without further retry.
    ///
    /// An error is terminal for the writer state which produced `realtime_commits`. The caller must
    /// discard its `RealtimeContext` and `FileStoreWrite`, load the current latest snapshot's
    /// durable offsets, recreate both objects, and replay input from those exclusive offsets.
    /// Conflicts from `CommitContextBuilder::WithCatalog()` are retried internally. With
    /// `UseRESTCatalogCommit()`, the caller sends the request and handles server conflicts.
    /// Concurrent rollback or partition deletion must be fenced by the upstream coordinator.
    /// Configure catalog writers with `WriteContextBuilder::WithCatalog()` too.
    ///
    /// In deduplicate mode each commit attempt resolves logical deleted offsets against the
    /// latest snapshot and merges them with its deletion vectors. Keep the complete progress
    /// when retrying; serializing only CommitMessage discards logical deletions.
    ///
    /// @param realtime_commits Commit messages and left-closed, right-open offset ranges to
    /// commit.
    /// @param commit_identifier Identifier of the streaming commit operation.
    /// @param watermark Optional event-time watermark.
    /// @return The id of the latest snapshot containing the committed progress. On retry, this may
    /// be a snapshot produced by a later commit and is suitable for refreshing a real-time context.
    virtual Result<int64_t> CommitWithProgress(
        const std::vector<RealtimeCommitProgress>& realtime_commits, int64_t commit_identifier,
        std::optional<int64_t> watermark) = 0;

    /// Filter out all `std::vector<CommitMessage>` which have been committed and commit the
    /// remaining ones.
    ///
    /// Compared to commit, this method will first check if a commit_identifier has been
    /// committed, so this method might be slower. A common usage of this method is to retry the
    /// commit process after a failure.
    ///
    /// @param commit_identifier_and_messages A map containing all {@link CommitMessage}s in
    ///     question. The key is the commit_identifier.
    ///
    /// @param watermark An optional event-time watermark used to indicate the progress of data
    ///     processing. Default is std::nullopt.
    /// @return Number of `std::vector<CommitMessage>` committed.
    virtual Result<int32_t> FilterAndCommit(
        const std::map<int64_t, std::vector<std::shared_ptr<CommitMessage>>>&
            commit_identifier_and_messages,
        std::optional<int64_t> watermark = std::nullopt) = 0;

    /// Overwrite from manifest committable and partition.
    ///
    /// @param partition A single partition maps each partition key to a partition value. Depending
    ///     on the user-defined statement, the partition might not include all partition keys. Also
    ///     note that this partition does not necessarily equal to the partitions of the newly added
    ///     key-values. This is just the partition to be cleaned up.
    /// @param commit_messages Description of the commit messages.
    /// @param commit_identifier Unique identifier.
    /// @param watermark An optional event-time watermark used to indicate the progress of data
    ///     processing. Default is std::nullopt.
    /// @return Result of the operation.
    /// @note A full-table overwrite clears all committed real-time progress. A partition
    ///     overwrite removes progress only for matching partitions. In either case, active
    ///     real-time writers and their `RealtimeContext` instances must be recreated before
    ///     further real-time operations.
    virtual Status Overwrite(const std::map<std::string, std::string>& partition,
                             const std::vector<std::shared_ptr<CommitMessage>>& commit_messages,
                             int64_t commit_identifier,
                             std::optional<int64_t> watermark = std::nullopt) = 0;

    /// This is a temporary interface for internal use. It will be removed in a future version.
    /// Please do not rely on it for long-term use.
    ///
    /// @param partition Description of the partition.
    /// @param commit_messages Description of the commit messages.
    /// @param commit_identifier Unique identifier.
    /// @param watermark An optional event-time watermark used to indicate the progress of data
    ///     processing. Default is std::nullopt.
    /// @return Result of the operation.
    /// @note A full-table overwrite clears all committed real-time progress. A partition
    ///     overwrite removes progress only for matching partitions. In either case, active
    ///     real-time writers and their `RealtimeContext` instances must be recreated before
    ///     further real-time operations.
    virtual Result<int32_t> FilterAndOverwrite(
        const std::map<std::string, std::string>& partition,
        const std::vector<std::shared_ptr<CommitMessage>>& commit_messages,
        int64_t commit_identifier, std::optional<int64_t> watermark = std::nullopt) = 0;

    /// Returns the request from the latest commit attempt, including failed attempts. Each attempt
    /// clears the request of the one before it, so an attempt that failed before building one, as
    /// one naming a branch a catalog cannot address does, leaves this returning an error.
    /// Catalog commits send requests automatically; `UseRESTCatalogCommit()` only prepares them.
    /// @note Temporary interface for internal use, will be removed in the future.
    /// @return JSON with `tableId`, `baseSnapshotUuid`, `snapshot` and `statistics`.
    /// Unset `tableId` and an absent or legacy base snapshot UUID are serialized as null.
    virtual Result<std::string> GetLastCommitTableRequest() = 0;

    /// Expire old snapshot in the file store.
    ///
    /// Protects files referenced by retained snapshots, including files restored by rollback.
    /// Catalog commits require the current snapshot and retained history to be published to the
    /// file system before deletion. An unpublished or mismatched current snapshot skips expiration;
    /// catalog or retained-metadata read errors propagate without deleting files.
    /// Retained snapshots with index manifests return `NotImplemented` before deletion.
    /// @note Coordinate rollback and expiration so they do not run concurrently.
    /// @note Returns `NotImplemented` before any snapshot is read or any file is deleted when this
    ///       commit is on a branch other than main, or the table path holds such a branch under
    ///       `branch/branch-<name>`: branches share the table's data files, while expiration only
    ///       reads the retained snapshots of its own branch. A branch held only by a catalog, or
    ///       created while expiration runs, is not found; do not expire a table with such a
    ///       branch, and serialize branch creation and expiration.
    /// @return Result<int32_t> indicating the number of expired items or an error status.
    virtual Result<int32_t> Expire() = 0;

    /// Drop specified partitions from the file store.
    ///
    /// @param partitions A vector of partitions to be dropped.
    /// @param commit_identifier An identifier for the commit operation.
    /// @return Status indicating the success or failure of the drop partition operation.
    /// @note A partition drop removes committed real-time progress only for matching partitions.
    ///     Active real-time writers and their `RealtimeContext` instances must be recreated before
    ///     further real-time operations.
    virtual Status DropPartition(const std::vector<std::map<std::string, std::string>>& partitions,
                                 int64_t commit_identifier) = 0;

    /// Truncate the whole table by overwriting all partitions with empty data. The generated
    /// snapshot has commit kind OVERWRITE.
    ///
    /// @param commit_identifier An identifier for the commit operation.
    /// @return Status indicating the success or failure of the truncate operation.
    /// @note Truncation clears all committed real-time progress. Active real-time writers and
    ///     their `RealtimeContext` instances must be recreated before further real-time
    ///     operations.
    virtual Status TruncateTable(int64_t commit_identifier) = 0;

    /// Abort an unsuccessful commit. The data and index files described by the given commit
    /// messages will be deleted on a best-effort basis (delete failures are ignored).
    ///
    /// @param commit_messages A vector of commit messages whose files should be cleaned up.
    /// @return Status indicating the success or failure of the abort operation.
    virtual Status Abort(const std::vector<std::shared_ptr<CommitMessage>>& commit_messages) = 0;

    /// Roll back to the target snapshot and materialize it as the latest snapshot.
    ///
    /// Reads the surviving files of both the current latest snapshot and the target
    /// snapshot, then commits an OVERWRITE snapshot whose visible state equals the target.
    ///
    /// @param target_snapshot_id The snapshot id to roll back to.
    /// @return Result<bool>; true if the atomic commit succeeded. Returns an error status if
    ///     there is no latest snapshot or the target snapshot does not exist.
    /// @note Rollback restores the real-time progress recorded by the target snapshot. Active
    ///     real-time writers and their `RealtimeContext` instances must be recreated before
    ///     further real-time operations. Coordinate rollback with expiration so the target's
    ///     files cannot be deleted while the rollback is being prepared or committed.
    virtual Result<bool> RollbackToAsLatest(int64_t target_snapshot_id) = 0;

    /// Configure row-id conflict checking from a specific snapshot id.
    ///
    /// If set to a snapshot id, commit conflict detection will additionally validate row-id
    /// conflicts against snapshots after that id. Passing std::nullopt disables this behavior.
    ///
    /// @param row_id_check_from_snapshot Snapshot id to start row-id conflict checks from, or
    ///     std::nullopt to disable.
    /// @return Current commit object for chaining.
    virtual FileStoreCommit& RowIdCheckConflict(
        std::optional<int64_t> row_id_check_from_snapshot) = 0;

    /// Retrieve metrics related to commit operations.
    ///
    /// @return A shared pointer to a `Metrics` object containing commit metrics.
    virtual std::shared_ptr<Metrics> GetCommitMetrics() const = 0;
};

}  // namespace paimon