feat(realtime): improve append table lifecycle and query support (#213)

Author: lxy-9602

include/paimon/api.h

@@ -33,6 +33,7 @@
 #include "paimon/record_batch.h"             // IWYU pragma: export
 #include "paimon/result.h"                   // IWYU pragma: export
 #include "paimon/scan_context.h"             // IWYU pragma: export
+#include "paimon/statistics_mode.h"          // IWYU pragma: export
 #include "paimon/status.h"                   // IWYU pragma: export
 #include "paimon/table/source/table_read.h"  // IWYU pragma: export
 #include "paimon/table/source/table_scan.h"  // IWYU pragma: export

include/paimon/defs.h

@@ -562,10 +562,18 @@ struct PAIMON_EXPORT Options {
     /// "scan.timestamp" can be used as an alternative string input for the same mode.
     static const char SCAN_TIMESTAMP_MILLIS[];
 
+    /// "realtime.enabled" - Whether real-time write, commit, and read operations are enabled.
+    /// Default value is "false".
+    static const char REALTIME_ENABLED[];
+
     /// "realtime.read-view-ttl" - Lifetime of a real-time memory view pinned by scan planning
     /// before reader creation. Default value is "5 min".
     static const char REALTIME_READ_VIEW_TTL[];
 
+    /// "realtime.store.stats-mode" - Statistics collected by the default real-time store.
+    /// Supported values are "none" and "full". Default value is "none".
+    static const char REALTIME_STORE_STATS_MODE[];
+
     /// "scan.timestamp" - Optional timestamp string used in case of "from-timestamp" scan mode,
     /// as an alternative to "scan.timestamp-millis".
     /// It will be automatically converted to timestamp in unix milliseconds, using local time zone.

include/paimon/file_store_commit.h

@@ -79,11 +79,17 @@ class PAIMON_EXPORT FileStoreCommit {
     /// orders them by partition, bucket, and offset before validating continuity. The resulting
     /// snapshot atomically publishes the data files and the updated offset map.
     ///
+    /// If this method returns an error, the caller may retry with the same arguments. Each call
+    /// reloads the latest committed state. As in `FilterAndCommit`, a retry's identifier is
+    /// considered committed when it is not newer than the latest identifier for `commit_user`.
+    /// The requested offset ranges must also be covered by the latest committed progress.
+    ///
     /// @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 final snapshot produced by this commit.
+    /// @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;
@@ -117,6 +123,10 @@ class PAIMON_EXPORT FileStoreCommit {
     /// @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,
@@ -131,6 +141,10 @@ class PAIMON_EXPORT FileStoreCommit {
     /// @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,
@@ -157,6 +171,9 @@ class PAIMON_EXPORT FileStoreCommit {
     /// @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;
 
@@ -165,6 +182,9 @@ class PAIMON_EXPORT FileStoreCommit {
     ///
     /// @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
@@ -182,6 +202,9 @@ class PAIMON_EXPORT FileStoreCommit {
     /// @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.
     virtual Result<bool> RollbackToAsLatest(int64_t target_snapshot_id) = 0;
 
     /// Configure row-id conflict checking from a specific snapshot id.

include/paimon/file_store_write.h

@@ -107,6 +107,12 @@ class PAIMON_EXPORT FileStoreWrite {
     ///
     /// The writer loads the snapshot's partition-bucket offsets and releases sealed memory that is
     /// fully covered by disk. Calling this method on a non-real-time writer returns an error.
+    /// If the snapshot overwrites table contents or moves committed progress backwards, such as
+    /// after a partition drop, overwrite, or rollback, this method returns an error and the caller
+    /// must recreate the `RealtimeContext` and writer. These operations are not fenced against an
+    /// active writer and do not clear its process-local state automatically. The caller must
+    /// coordinate them with active writers; skipping the resetting snapshot and continuing to use
+    /// an old context is unsupported.
     virtual Status RefreshCommittedSnapshot(int64_t snapshot_id);
 
     virtual std::shared_ptr<Metrics> GetMetrics() const = 0;

include/paimon/realtime/arrow_realtime_store_factory.h

@@ -28,7 +28,7 @@ class PAIMON_EXPORT ArrowRealtimeStoreFactory : public RealtimeStoreFactory {
  public:
     /// Creates an Arrow-backed store for one partition and bucket.
     Result<std::shared_ptr<RealtimeStore>> Create(
-        std::unique_ptr<::ArrowSchema> write_schema,
+        std::unique_ptr<::ArrowSchema> write_schema, StatisticsMode statistics_mode,
         const std::map<std::string, std::string>& options,
         const std::shared_ptr<MemoryPool>& memory_pool) override;
 };

include/paimon/realtime/realtime_context.h

@@ -73,6 +73,11 @@ using RealtimeOffsetMap = std::map<RealtimePartitionBucket, int64_t>;
 /// reads. `RealtimeContext` itself is not a customization interface and must not be implemented by
 /// applications. Customize real-time storage and retrieval through `RealtimeStoreFactory` and
 /// `RealtimeStore` instead.
+///
+/// A context is valid only for one uninterrupted committed-progress history. Overwrite, truncate,
+/// partition drop, and rollback operations do not automatically clear process-local real-time
+/// state. Applications must coordinate these operations with active real-time writers and recreate
+/// the `RealtimeContext` and writers before continuing.
 class PAIMON_EXPORT RealtimeContext {
  public:
     /// Creates a context backed by Paimon's default in-memory Arrow `RealtimeStore`.

include/paimon/realtime/realtime_store.h

@@ -31,6 +31,7 @@
 #include "paimon/realtime/offset_range.h"
 #include "paimon/record_batch.h"
 #include "paimon/result.h"
+#include "paimon/statistics_mode.h"
 #include "paimon/visibility.h"
 
 struct ArrowSchema;
@@ -156,13 +157,14 @@ class PAIMON_EXPORT RealtimeStoreFactory {
  public:
     virtual ~RealtimeStoreFactory() = default;
 
-    /// Creates a store configured with the supplied schema, options, and memory pool.
+    /// Creates a store configured with the supplied schema, statistics, options, and memory pool.
     /// @param write_schema Complete table write schema whose ownership is transferred to the
     /// factory. The factory may consume it or retain it in the created store.
+    /// @param statistics_mode Framework-parsed statistics collection mode.
     /// @param options Effective table options available to the store.
     /// @param memory_pool Memory pool provided by the write context.
     virtual Result<std::shared_ptr<RealtimeStore>> Create(
-        std::unique_ptr<::ArrowSchema> write_schema,
+        std::unique_ptr<::ArrowSchema> write_schema, StatisticsMode statistics_mode,
         const std::map<std::string, std::string>& options,
         const std::shared_ptr<MemoryPool>& memory_pool) = 0;
 };

include/paimon/scan_context.h

@@ -170,7 +170,7 @@ class PAIMON_EXPORT ScanContextBuilder {
     ScanContextBuilder& SetGlobalIndexResult(
         const std::shared_ptr<GlobalIndexResult>& global_index_result);
 
-    /// Enables process-local union reads with the memory indexers owned by `realtime_context`.
+    /// Enables process-local union reads with the real-time stores owned by `realtime_context`.
     ScanContextBuilder& WithRealtimeContext(
         const std::shared_ptr<RealtimeContext>& realtime_context);
 

include/paimon/statistics_mode.h

@@ -0,0 +1,32 @@
+/*
+ * 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
+
+namespace paimon {
+
+/// Controls the amount of statistics collected for metadata pruning.
+enum class StatisticsMode {
+    /// Do not collect statistics.
+    NONE,
+    /// Collect statistics for all supported fields.
+    FULL,
+};
+
+}  // namespace paimon

include/paimon/write_context.h

@@ -218,7 +218,7 @@ class PAIMON_EXPORT WriteContextBuilder {
     WriteContextBuilder& WithFileSystem(const std::shared_ptr<FileSystem>& file_system);
 
     /// Enables the real-time write path with the provided shared context.
-    /// @param realtime_context Non-null context that owns the real-time indexers.
+    /// @param realtime_context Non-null context that owns the real-time stores.
     /// @return Reference to this builder for method chaining.
     WriteContextBuilder& WithRealtimeContext(
         const std::shared_ptr<RealtimeContext>& realtime_context);