Merge branch 'alibaba:main' into fix-ci

Author: lucasfang

cmake_modules/ThirdpartyToolchain.cmake

@@ -548,6 +548,10 @@ macro(build_avro)
 
     get_target_property(AVRO_ZSTD_INCLUDE_DIR zstd INTERFACE_INCLUDE_DIRECTORIES)
     get_filename_component(AVRO_ZSTD_ROOT "${AVRO_ZSTD_INCLUDE_DIR}" DIRECTORY)
+
+    get_target_property(AVRO_ZLIB_INCLUDE_DIR zlib INTERFACE_INCLUDE_DIRECTORIES)
+    get_filename_component(AVRO_ZLIB_ROOT "${AVRO_ZLIB_INCLUDE_DIR}" DIRECTORY)
+
     set(AVRO_CMAKE_CXX_FLAGS "${EP_CXX_FLAGS} -Wno-error")
     set(AVRO_CMAKE_C_FLAGS "${EP_C_FLAGS} -Wno-error")
 
@@ -558,7 +562,7 @@ macro(build_avro)
         "-DCMAKE_C_FLAGS=${AVRO_CMAKE_C_FLAGS}"
         "-DAVRO_BUILD_TESTS=OFF"
         "-DAVRO_BUILD_EXECUTABLES=OFF"
-        "-DZLIB_ROOT=${THIRDPARTY_ZLIB_ROOT}"
+        "-DZLIB_ROOT=${AVRO_ZLIB_ROOT}"
         "-Dzstd_ROOT=${AVRO_ZSTD_ROOT}"
         "-DSnappy_ROOT=${AVRO_SNAPPY_ROOT}")
     externalproject_add(avro_ep

include/paimon/defs.h

@@ -282,6 +282,8 @@ struct PAIMON_EXPORT Options {
     /// "blob-as-descriptor" - Read and write blob field using blob descriptor rather than blob
     /// bytes. Default value is "false".
     static const char BLOB_AS_DESCRIPTOR[];
+    /// "global-index.enabled" - Whether to enable global index for scan. Default value is "true".
+    static const char GLOBAL_INDEX_ENABLED[];
 };
 
 static constexpr int64_t BATCH_WRITE_COMMIT_IDENTIFIER = std::numeric_limits<int64_t>::max();

include/paimon/global_index/bitmap_global_index_result.h

@@ -78,9 +78,9 @@ class PAIMON_EXPORT BitmapGlobalIndexResult : public GlobalIndexResult {
     ///       bitmap is not actually required. **Not thread-safe**.
     Result<const RoaringBitmap64*> GetBitmap() const;
 
-    /// @return A shared pointer to a new BitmapGlobalIndexResult instance representing the given
-    ///         inclusive range [from, to].
-    static std::shared_ptr<BitmapGlobalIndexResult> FromRange(const Range& range);
+    /// Creates `BitmapGlobalIndexResult` for all row ids in the given ranges.
+    /// @note Overlapping or unsorted ranges are accepted.
+    static std::shared_ptr<BitmapGlobalIndexResult> FromRanges(const std::vector<Range>& ranges);
 
  private:
     mutable bool initialized_ = false;

include/paimon/global_index/global_index_result.h

@@ -23,6 +23,7 @@
 #include "paimon/memory/bytes.h"
 #include "paimon/memory/memory_pool.h"
 #include "paimon/result.h"
+#include "paimon/utils/range.h"
 #include "paimon/visibility.h"
 
 namespace paimon {
@@ -55,6 +56,9 @@ class PAIMON_EXPORT GlobalIndexResult : public std::enable_shared_from_this<Glob
     /// Creates a new iterator over the selected global row ids.
     virtual Result<std::unique_ptr<Iterator>> CreateIterator() const = 0;
 
+    /// Returns non-overlapping, sorted ranges covering all row ids in `GlobalIndexResult`.
+    Result<std::vector<Range>> ToRanges() const;
+
     /// Computes the logical AND (intersection) between current result and another.
     virtual Result<std::shared_ptr<GlobalIndexResult>> And(
         const std::shared_ptr<GlobalIndexResult>& other);

include/paimon/global_index/global_index_scan.h

@@ -37,7 +37,7 @@ class PAIMON_EXPORT GlobalIndexScan {
     ///
     /// @param table_path     Root directory of the table.
     /// @param snapshot_id    Optional snapshot ID to read from; if not provided, uses the latest.
-    /// @param partitions     Optional list of partition specs to restrict the scan scope.
+    /// @param partitions     Optional list of specific partitions to restrict the scan scope.
     ///                       Each map represents one partition (e.g., {"dt": "2024-06-01"}).
     ///                       If omitted, scans all partitions.
     /// @param options        Index-specific configuration.
@@ -53,6 +53,16 @@ class PAIMON_EXPORT GlobalIndexScan {
         const std::map<std::string, std::string>& options,
         const std::shared_ptr<FileSystem>& file_system, const std::shared_ptr<MemoryPool>& pool);
 
+    /// Creates a `GlobalIndexScan` instance for the specified table and context.
+    ///
+    /// @param partition_filters Optional specific partition predicates.
+    static Result<std::unique_ptr<GlobalIndexScan>> Create(
+        const std::string& root_path, const std::optional<int64_t>& snapshot_id,
+        const std::shared_ptr<Predicate>& partition_filters,
+        const std::map<std::string, std::string>& options,
+        const std::shared_ptr<FileSystem>& file_system,
+        const std::shared_ptr<MemoryPool>& memory_pool);
+
     virtual ~GlobalIndexScan() = default;
 
     /// Creates a scanner for the global index over the specified row ID range.

include/paimon/global_index/row_range_global_index_scanner.h

@@ -52,6 +52,18 @@ class PAIMON_EXPORT RowRangeGlobalIndexScanner {
     ///         format).
     virtual Result<std::shared_ptr<GlobalIndexReader>> CreateReader(
         const std::string& field_name, const std::string& index_type) const = 0;
+
+    /// Creates several `GlobalIndexReader`s for a specific field within this range.
+    ///
+    /// @param field_name  Name of the indexed column.
+    /// @return A `Result` that is:
+    ///         - Successful with several readers if the indexes exist and load correctly;
+    ///         - Successful with an empty vector if no index was built for the given field;
+    ///         - Error returns when loading fails (e.g., file corruption, I/O error, unsupported
+    ///         format) or the predicate method was incorrectly invoked (e.g., VisitTopK was invoked
+    ///         incorrectly).
+    virtual Result<std::vector<std::shared_ptr<GlobalIndexReader>>> CreateReaders(
+        const std::string& field_name) const = 0;
 };
 
 }  // namespace paimon

include/paimon/scan_context.h

@@ -23,12 +23,11 @@
 #include <string>
 #include <vector>
 
+#include "paimon/global_index/global_index_result.h"
 #include "paimon/predicate/predicate.h"
 #include "paimon/result.h"
 #include "paimon/type_fwd.h"
-#include "paimon/utils/range.h"
 #include "paimon/visibility.h"
-
 namespace paimon {
 class ScanContextBuilder;
 class ScanFilter;
@@ -45,6 +44,7 @@ class PAIMON_EXPORT ScanContext {
  public:
     ScanContext(const std::string& path, bool is_streaming_mode, std::optional<int32_t> limit,
                 const std::shared_ptr<ScanFilter>& scan_filter,
+                const std::shared_ptr<GlobalIndexResult>& global_index_result,
                 const std::shared_ptr<MemoryPool>& memory_pool,
                 const std::shared_ptr<Executor>& executor,
                 const std::map<std::string, std::string>& options);
@@ -77,12 +77,16 @@ class PAIMON_EXPORT ScanContext {
     std::shared_ptr<Executor> GetExecutor() const {
         return executor_;
     }
+    std::shared_ptr<GlobalIndexResult> GetGlobalIndexResult() const {
+        return global_index_result_;
+    }
 
  private:
     std::string path_;
     bool is_streaming_mode_;
     std::optional<int32_t> limit_;
     std::shared_ptr<ScanFilter> scan_filters_;
+    std::shared_ptr<GlobalIndexResult> global_index_result_;
     std::shared_ptr<MemoryPool> memory_pool_;
     std::shared_ptr<Executor> executor_;
     std::map<std::string, std::string> options_;
@@ -93,11 +97,10 @@ class PAIMON_EXPORT ScanFilter {
  public:
     ScanFilter(const std::shared_ptr<Predicate>& predicate,
                const std::vector<std::map<std::string, std::string>>& partition_filters,
-               const std::optional<int32_t>& bucket_filter, const std::vector<Range>& row_ranges)
+               const std::optional<int32_t>& bucket_filter)
         : predicates_(predicate),
           bucket_filter_(bucket_filter),
-          partition_filters_(partition_filters),
-          row_ranges_(row_ranges) {}
+          partition_filters_(partition_filters) {}
 
     std::shared_ptr<Predicate> GetPredicate() const {
         return predicates_;
@@ -109,15 +112,10 @@ class PAIMON_EXPORT ScanFilter {
         return partition_filters_;
     }
 
-    const std::vector<Range>& GetRowRanges() const {
-        return row_ranges_;
-    }
-
  private:
     std::shared_ptr<Predicate> predicates_;
     std::optional<int32_t> bucket_filter_;
     std::vector<std::map<std::string, std::string>> partition_filters_;
-    std::vector<Range> row_ranges_;
 };
 
 /// `ScanContextBuilder` used to build a `ScanContext`, has input validation.
@@ -138,10 +136,11 @@ class PAIMON_EXPORT ScanContextBuilder {
     /// Set a predicate for filtering data.
     ScanContextBuilder& SetPredicate(const std::shared_ptr<Predicate>& predicate);
 
-    /// Specify the row id ranges for scan. This is usually used to read specific rows in
-    /// data-evolution mode. File ranges that do not have any intersection with range_ids will be
-    /// filtered. If not set, all rows are returned
-    ScanContextBuilder& SetRowRanges(const std::vector<Range>& row_ranges);
+    /// Sets the result of a global index search (e.g., row ids (may with scores) from a distributed
+    /// index lookup). This is used to push down index-filtered row ids into the scan for efficient
+    /// data retrieval.
+    ScanContextBuilder& SetGlobalIndexResult(
+        const std::shared_ptr<GlobalIndexResult>& global_index_result);
     /// The options added or set in `ScanContextBuilder` have high priority and will be merged with
     /// the options in table schema.
     ScanContextBuilder& AddOption(const std::string& key, const std::string& value);

include/paimon/utils/range.h

@@ -46,6 +46,15 @@ struct PAIMON_EXPORT Range {
     /// Computes the set intersection of two collections of disjoint, sorted ranges.
     static std::vector<Range> And(const std::vector<Range>& left, const std::vector<Range>& right);
 
+    /// Excludes the given ranges from this range and returns the remaining ranges.
+    ///
+    /// For example, if this range is [0, 10000] and ranges to exclude are [1000, 2000], [3000,
+    /// 4000], [5000, 6000], then the result is [0, 999], [2001, 2999], [4001, 4999], [6001, 10000].
+    ///
+    /// @param ranges The ranges to exclude (can be unsorted and overlapping).
+    /// @return The remaining ranges after exclusion.
+    std::vector<Range> Exclude(const std::vector<Range>& ranges) const;
+
     bool operator==(const Range& other) const;
     bool operator<(const Range& other) const;
 

src/paimon/CMakeLists.txt

@@ -241,6 +241,7 @@ set(PAIMON_CORE_SRCS
     core/table/source/startup_mode.cpp
     core/table/source/table_read.cpp
     core/table/source/table_scan.cpp
+    core/table/source/data_evolution_batch_scan.cpp
     core/utils/field_mapping.cpp
     core/utils/fields_comparator.cpp
     core/utils/file_store_path_factory.cpp

src/paimon/common/defs.cpp

@@ -79,5 +79,5 @@ const char Options::ROW_TRACKING_ENABLED[] = "row-tracking.enabled";
 const char Options::DATA_EVOLUTION_ENABLED[] = "data-evolution.enabled";
 const char Options::PARTITION_GENERATE_LEGACY_NAME[] = "partition.legacy-name";
 const char Options::BLOB_AS_DESCRIPTOR[] = "blob-as-descriptor";
-
+const char Options::GLOBAL_INDEX_ENABLED[] = "global-index.enabled";
 }  // namespace paimon