Merge branch 'main' into spill-pr1

Author: lxy-9602

include/paimon/append/append_compact_coordinator.h

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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 <map>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "paimon/result.h"
+
+namespace paimon {
+
+class CommitMessage;
+class FileSystem;
+class MemoryPool;
+
+/// Compact coordinator for append-only unaware-bucket tables.
+///
+/// This coordinator scans the latest snapshot for small files, groups them by partition,
+/// and generates compaction tasks using a bin-packing algorithm. It then synchronously
+/// executes all tasks and returns the resulting commit messages.
+///
+/// @note This implementation does not support deletion vectors or streaming mode.
+///       It only scans the current latest snapshot (batch mode).
+class PAIMON_EXPORT AppendCompactCoordinator {
+ public:
+    AppendCompactCoordinator() = delete;
+    ~AppendCompactCoordinator() = delete;
+    /// Run the compaction coordinator.
+    ///
+    /// Scans the latest snapshot for small files across the specified partitions,
+    /// generates compact tasks via bin-packing, executes them synchronously,
+    /// and returns the resulting commit messages.
+    ///
+    /// @param table_path The root path of the table.
+    /// @param options User-defined options (will be merged with schema options).
+    /// @param partitions Partition filters; each element is a partition spec as key-value pairs.
+    ///                   Empty vector means all partitions.
+    /// @param file_system The file system to use. If nullptr, will be created from options.
+    /// @param pool The memory pool to use. If nullptr, will use default pool.
+    /// @return Result containing a vector of commit messages from compaction tasks.
+    static Result<std::vector<std::shared_ptr<CommitMessage>>> Run(
+        const std::string& table_path, const std::map<std::string, std::string>& options,
+        const std::vector<std::map<std::string, std::string>>& partitions,
+        const std::shared_ptr<FileSystem>& file_system, const std::shared_ptr<MemoryPool>& pool);
+};
+
+}  // namespace paimon

include/paimon/catalog/catalog.h

@@ -26,6 +26,7 @@
 
 #include "paimon/catalog/identifier.h"
 #include "paimon/result.h"
+#include "paimon/snapshot/snapshot_info.h"
 #include "paimon/status.h"
 #include "paimon/type_fwd.h"
 #include "paimon/visibility.h"
@@ -188,6 +189,15 @@ class PAIMON_EXPORT Catalog {
     /// @param identifier The identifier (database and table name) of the table to load.
     /// @return A result containing table schema if the table exists, or an error status on failure.
     virtual Result<std::shared_ptr<Schema>> LoadTableSchema(const Identifier& identifier) const = 0;
+
+    /// Lists all snapshots of the specified table, ordered by snapshot id.
+    ///
+    /// @param identifier The identifier (database and table name) of the table.
+    /// @param branch Branch name; empty string means the main branch.
+    /// @return A result containing a vector of SnapshotInfo ordered by
+    ///         snapshot id ascending, or an error status.
+    virtual Result<std::vector<SnapshotInfo>> ListSnapshots(
+        const Identifier& identifier, const std::string& branch = "") const = 0;
 };
 
 }  // namespace paimon

include/paimon/defs.h

@@ -434,24 +434,6 @@ struct PAIMON_EXPORT Options {
     /// "lookup.cache-max-disk-size" - Max disk size for lookup cache, you can use this option
     /// to limit the use of local disks. Default value is unlimited (INT64_MAX).
     static const char LOOKUP_CACHE_MAX_DISK_SIZE[];
-    /// "btree-index.compression" - The compression algorithm to use for BTreeIndex.
-    /// Default value is "none".
-    static const char BTREE_INDEX_COMPRESSION[];
-    /// "btree-index.compression-level" - The compression level of the compression algorithm.
-    /// Default value is 1.
-    static const char BTREE_INDEX_COMPRESSION_LEVEL[];
-    /// "btree-index.block-size" - The block size to use for BTreeIndex.
-    /// Default value is 64 KB.
-    static const char BTREE_INDEX_BLOCK_SIZE[];
-    /// "btree-index.cache-size" - The cache size to use for BTreeIndex.
-    /// Default value is 128 MB.
-    static const char BTREE_INDEX_CACHE_SIZE[];
-    /// "btree-index.high-priority-pool-ratio" - The high priority pool ratio to use for BTreeIndex.
-    /// Default value is 0.1.
-    static const char BTREE_INDEX_HIGH_PRIORITY_POOL_RATIO[];
-    /// "btree-index.records-per-range" - The expected number of records per BTree Index File.
-    /// Default value is 1000000.
-    static const char BTREE_INDEX_RECORDS_PER_RANGE[];
 };
 
 static constexpr int64_t BATCH_WRITE_COMMIT_IDENTIFIER = std::numeric_limits<int64_t>::max();

include/paimon/file_index/file_index_reader.h

@@ -62,12 +62,6 @@ class PAIMON_EXPORT FileIndexReader : public FunctionVisitor<std::shared_ptr<Fil
     Result<std::shared_ptr<FileIndexResult>> VisitContains(const Literal& literal) override;
 
     Result<std::shared_ptr<FileIndexResult>> VisitLike(const Literal& literal) override;
-
-    Result<std::shared_ptr<FileIndexResult>> VisitAnd(
-        const std::vector<Result<std::shared_ptr<FileIndexResult>>>& children) override;
-
-    Result<std::shared_ptr<FileIndexResult>> VisitOr(
-        const std::vector<Result<std::shared_ptr<FileIndexResult>>>& children) override;
 };
 
 }  // namespace paimon

include/paimon/global_index/global_index_reader.h

@@ -48,20 +48,6 @@ class PAIMON_EXPORT GlobalIndexReader : public FunctionVisitor<std::shared_ptr<G
     virtual Result<std::shared_ptr<GlobalIndexResult>> VisitFullTextSearch(
         const std::shared_ptr<FullTextSearch>& full_text_search) = 0;
 
-    /// VisitAnd performs logical AND across multiple child results.
-    /// Default implementation returns "not supported" error.
-    Result<std::shared_ptr<GlobalIndexResult>> VisitAnd(
-        const std::vector<Result<std::shared_ptr<GlobalIndexResult>>>& children) override {
-        return Status::NotImplemented("AND operations not supported by this index type");
-    }
-
-    /// VisitOr performs logical OR across multiple child results.
-    /// Default implementation returns "not supported" error.
-    Result<std::shared_ptr<GlobalIndexResult>> VisitOr(
-        const std::vector<Result<std::shared_ptr<GlobalIndexResult>>>& children) override {
-        return Status::NotImplemented("OR operations not supported by this index type");
-    }
-
     /// @return true if the reader is thread-safe; false otherwise.
     virtual bool IsThreadSafe() const = 0;
 

include/paimon/predicate/function_visitor.h

@@ -74,41 +74,5 @@ class PAIMON_EXPORT FunctionVisitor {
 
     /// Evaluates whether string values like the given string.
     virtual Result<T> VisitLike(const Literal& literal) = 0;
-
-    /// Evaluates the BETWEEN predicate with the given lower and upper bounds.
-    virtual Result<T> VisitBetween(const Literal& from, const Literal& to) {
-        // Default implementation: BETWEEN is equivalent to >= AND <=
-        auto lower_result = VisitGreaterOrEqual(from);
-        if (!lower_result.ok()) {
-            return lower_result.status();
-        }
-        auto upper_result = VisitLessOrEqual(to);
-        if (!upper_result.ok()) {
-            return upper_result.status();
-        }
-        return VisitAnd({std::move(lower_result).value(), std::move(upper_result).value()});
-    }
-
-    /// Evaluates the NOT BETWEEN predicate with the given lower and upper bounds.
-    virtual Result<T> VisitNotBetween(const Literal& from, const Literal& to) {
-        // Default implementation: NOT BETWEEN is equivalent to < OR >
-        auto lower_result = VisitLessThan(from);
-        if (!lower_result.ok()) {
-            return lower_result.status();
-        }
-        auto upper_result = VisitGreaterThan(to);
-        if (!upper_result.ok()) {
-            return upper_result.status();
-        }
-        return VisitOr({std::move(lower_result).value(), std::move(upper_result).value()});
-    }
-
-    // ----------------- Compound functions ------------------------
-
-    /// Evaluates the AND predicate across multiple child results.
-    virtual Result<T> VisitAnd(const std::vector<Result<T>>& children) = 0;
-
-    /// Evaluates the OR predicate across multiple child results.
-    virtual Result<T> VisitOr(const std::vector<Result<T>>& children) = 0;
 };
 }  // namespace paimon

include/paimon/predicate/literal.h

@@ -106,7 +106,7 @@ class PAIMON_EXPORT Literal {
 
 namespace std {
 template <>
-struct hash<paimon::Literal> {
+struct hash<::paimon::Literal> {
     size_t operator()(const paimon::Literal& literal) const {
         return literal.HashCode();
     }

include/paimon/snapshot/snapshot_info.h

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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 <optional>
+#include <string>
+
+#include "paimon/visibility.h"
+
+namespace paimon {
+
+/// Plain snapshot metadata returned by Catalog::ListSnapshots().
+struct PAIMON_EXPORT SnapshotInfo {
+    /// Commit kind exposed through the public Catalog API.
+    enum class PAIMON_EXPORT CommitKind : int8_t {
+        APPEND,
+        COMPACT,
+        OVERWRITE,
+        ANALYZE,
+        UNKNOWN,
+    };
+
+    /// Convert a CommitKind to its canonical string representation.
+    static std::string CommitKindToString(CommitKind kind);
+
+    int64_t snapshot_id;
+    int64_t schema_id;
+    std::string commit_user;
+    CommitKind commit_kind;
+    int64_t time_millis;
+    std::optional<int64_t> total_record_count;
+    std::optional<int64_t> delta_record_count;
+    std::optional<int64_t> watermark;
+};
+
+}  // namespace paimon

src/paimon/CMakeLists.txt

@@ -151,6 +151,9 @@ set(PAIMON_CORE_SRCS
     core/mergetree/spill_writer.cpp
     core/append/append_only_writer.cpp
     core/append/bucketed_append_compact_manager.cpp
+    core/bucket/bucket_select_converter.cpp
+    core/append/append_compact_task.cpp
+    core/append/append_compact_coordinator.cpp
     core/bucket/hive_bucket_function.cpp
     core/bucket/mod_bucket_function.cpp
     core/bucket/bucket_id_calculator.cpp
@@ -284,6 +287,7 @@ set(PAIMON_CORE_SRCS
     core/schema/schema_validation.cpp
     core/schema/table_schema.cpp
     core/snapshot.cpp
+    core/snapshot_info.cpp
     core/stats/simple_stats_collector.cpp
     core/stats/simple_stats_converter.cpp
     core/stats/simple_stats.cpp
@@ -412,8 +416,7 @@ if(PAIMON_BUILD_TESTS)
                     common/global_index/bitmap/bitmap_global_index_test.cpp
                     common/global_index/btree/btree_index_meta_test.cpp
                     common/global_index/btree/btree_file_footer_test.cpp
-                    common/global_index/btree/btree_global_indexer_test.cpp
-                    common/global_index/btree/btree_global_index_writer_test.cpp
+                    common/global_index/btree/key_serializer_test.cpp
                     common/global_index/btree/btree_global_index_integration_test.cpp
                     common/global_index/btree/btree_compatibility_test.cpp
                     common/global_index/rangebitmap/range_bitmap_global_index_test.cpp
@@ -527,7 +530,9 @@ if(PAIMON_BUILD_TESTS)
     add_paimon_test(core_test
                     SOURCES
                     core/append/append_only_writer_test.cpp
+                    core/append/append_compact_coordinator_test.cpp
                     core/append/bucketed_append_compact_manager_test.cpp
+                    core/bucket/bucket_select_converter_test.cpp
                     core/bucket/default_bucket_function_test.cpp
                     core/bucket/hive_bucket_function_test.cpp
                     core/bucket/bucket_id_calculator_test.cpp

src/paimon/common/data/binary_row.cpp

@@ -18,6 +18,7 @@
 
 #include <cstdint>
 
+#include "fmt/format.h"
 #include "paimon/common/data/binary_data_read_utils.h"
 #include "paimon/common/memory/memory_segment.h"
 #include "paimon/common/memory/memory_segment_utils.h"
@@ -300,4 +301,8 @@ int32_t BinaryRow::HashCode() const {
     return MemorySegmentUtils::HashByWords({segment_}, offset_, size_in_bytes_, nullptr);
 }
 
+std::string BinaryRow::ToString() const {
+    return fmt::format("BinaryRow@{:#x}", static_cast<uint32_t>(HashCode()));
+}
+
 }  // namespace paimon