Merge branch 'main' into codex/extend-map-column-allocator

Author: zjw1111

include/paimon/global_index/global_index_scan.h

@@ -97,6 +97,22 @@ class PAIMON_EXPORT GlobalIndexScan {
         const std::string& field_name,
         const std::optional<RowRangeIndex>& row_range_index) const = 0;
 
+    /// Creates a `GlobalIndexReader` for a specific field with a specific index type.
+    /// @param field_name       Name of the indexed column.
+    /// @param index_type       Name of index type.
+    /// @param row_range_index  Optional row range that limits the scan to a sub-range of row ids.
+    ///                         If not provided, the entire row range is considered.
+    /// @return A `Result` that is:
+    ///         - Successful with a reader(with global row id) if the index exists and loads
+    ///         correctly;
+    ///         - Successful with nullptr if no index was built for the given field with the given
+    ///         index type;
+    ///         - Error returns when loading fails (e.g., file corruption, I/O error,
+    ///           unsupported format).
+    virtual Result<std::shared_ptr<GlobalIndexReader>> CreateReader(
+        const std::string& field_name, const std::string& index_type,
+        const std::optional<RowRangeIndex>& row_range_index) const = 0;
+
     /// Creates several `GlobalIndexReader`s for a specific field (looked up by id),
     /// @param field_id         Field id of the indexed column.
     /// @param row_range_index  Optional row range that limits the scan to a sub-range of row ids.

src/paimon/core/global_index/global_index_scan_impl.cpp

@@ -134,27 +134,48 @@ Result<std::shared_ptr<GlobalIndexEvaluator>> GlobalIndexScanImpl::GetOrCreateIn
 Result<std::vector<std::shared_ptr<GlobalIndexReader>>> GlobalIndexScanImpl::CreateReaders(
     int32_t field_id, const std::optional<RowRangeIndex>& row_range_index) const {
     PAIMON_ASSIGN_OR_RAISE(DataField field, table_schema_->GetField(field_id));
-    return CreateReaders(field, row_range_index);
+    return CreateReaders(field, /*index_type=*/std::nullopt, row_range_index);
 }
 
 Result<std::vector<std::shared_ptr<GlobalIndexReader>>> GlobalIndexScanImpl::CreateReaders(
     const std::string& field_name, const std::optional<RowRangeIndex>& row_range_index) const {
     PAIMON_ASSIGN_OR_RAISE(DataField field, table_schema_->GetField(field_name));
-    return CreateReaders(field, row_range_index);
+    return CreateReaders(field, /*index_type=*/std::nullopt, row_range_index);
+}
+
+Result<std::shared_ptr<GlobalIndexReader>> GlobalIndexScanImpl::CreateReader(
+    const std::string& field_name, const std::string& index_type,
+    const std::optional<RowRangeIndex>& row_range_index) const {
+    PAIMON_ASSIGN_OR_RAISE(DataField field, table_schema_->GetField(field_name));
+    PAIMON_ASSIGN_OR_RAISE(
+        std::vector<std::shared_ptr<GlobalIndexReader>> readers,
+        CreateReaders(field, std::optional<std::string>(index_type), row_range_index));
+    if (readers.empty()) {
+        return std::shared_ptr<GlobalIndexReader>();
+    }
+    if (readers.size() != 1) {
+        return Status::Invalid(
+            fmt::format("invalid global index reader size, expected 1, actual {}", readers.size()));
+    }
+    return readers[0];
 }
 
 Result<std::vector<std::shared_ptr<GlobalIndexReader>>> GlobalIndexScanImpl::CreateReaders(
-    const DataField& field, const std::optional<RowRangeIndex>& row_range_index) const {
+    const DataField& field, const std::optional<std::string>& index_type,
+    const std::optional<RowRangeIndex>& row_range_index) const {
     auto field_iter = index_metas_.find(field.Id());
     if (field_iter == index_metas_.end()) {
         return std::vector<std::shared_ptr<GlobalIndexReader>>();
     }
     const auto& index_type_to_metas = field_iter->second;
     std::vector<std::shared_ptr<GlobalIndexReader>> readers;
-    readers.reserve(index_type_to_metas.size());
-    for (const auto& [index_type, range_to_metas] : index_type_to_metas) {
+    readers.reserve(index_type ? 1 : index_type_to_metas.size());
+    for (const auto& [current_index_type, range_to_metas] : index_type_to_metas) {
+        if (index_type && current_index_type != index_type.value()) {
+            continue;
+        }
         PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<GlobalIndexer> indexer,
-                               GlobalIndexerFactory::Get(index_type, options_.ToMap()));
+                               GlobalIndexerFactory::Get(current_index_type, options_.ToMap()));
         if (!indexer) {
             continue;
         }

src/paimon/core/global_index/global_index_scan_impl.h

@@ -18,6 +18,7 @@
 
 #include <map>
 #include <memory>
+#include <optional>
 #include <string>
 #include <vector>
 
@@ -47,6 +48,10 @@ class GlobalIndexScanImpl : public GlobalIndexScan {
         const std::string& field_name,
         const std::optional<RowRangeIndex>& row_range_index) const override;
 
+    Result<std::shared_ptr<GlobalIndexReader>> CreateReader(
+        const std::string& field_name, const std::string& index_type,
+        const std::optional<RowRangeIndex>& row_range_index) const override;
+
     Result<std::vector<std::shared_ptr<GlobalIndexReader>>> CreateReaders(
         int32_t field_id, const std::optional<RowRangeIndex>& row_range_index) const override;
 
@@ -65,7 +70,8 @@ class GlobalIndexScanImpl : public GlobalIndexScan {
     Result<std::shared_ptr<GlobalIndexEvaluator>> GetOrCreateIndexEvaluator();
 
     Result<std::vector<std::shared_ptr<GlobalIndexReader>>> CreateReaders(
-        const DataField& field, const std::optional<RowRangeIndex>& row_range_index) const;
+        const DataField& field, const std::optional<std::string>& index_type,
+        const std::optional<RowRangeIndex>& row_range_index) const;
 
     std::vector<GlobalIndexIOMeta> ToGlobalIndexIOMetas(
         const std::vector<std::shared_ptr<IndexFileMeta>>& metas) const;

test/inte/global_index_test.cpp

@@ -2661,6 +2661,33 @@ TEST_P(GlobalIndexTest, TestBTreeAndBitmapCoexist) {
     ASSERT_OK_AND_ASSIGN(auto index_readers, global_index_scan->CreateReaders("f0", std::nullopt));
     ASSERT_EQ(index_readers.size(), 2u);
 
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexReader> btree_reader,
+                         global_index_scan->CreateReader("f0", "btree", std::nullopt));
+    ASSERT_TRUE(btree_reader);
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexResult> btree_result,
+                         btree_reader->VisitEqual(Literal(FieldType::STRING, "Bob", 3)));
+    ASSERT_TRUE(btree_result);
+    ASSERT_EQ(btree_result->ToString(), "{1,2}");
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexResult> btree_less_than_result,
+                         btree_reader->VisitLessThan(Literal(FieldType::STRING, "Emily", 5)));
+    ASSERT_TRUE(btree_less_than_result);
+    ASSERT_EQ(btree_less_than_result->ToString(), "{0,1,2}");
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexReader> bitmap_reader,
+                         global_index_scan->CreateReader("f0", "bitmap", std::nullopt));
+    ASSERT_TRUE(bitmap_reader);
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexResult> bitmap_result,
+                         bitmap_reader->VisitEqual(Literal(FieldType::STRING, "Bob", 3)));
+    ASSERT_TRUE(bitmap_result);
+    ASSERT_EQ(bitmap_result->ToString(), "{1,2}");
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexResult> bitmap_less_than_result,
+                         bitmap_reader->VisitLessThan(Literal(FieldType::STRING, "Emily", 5)));
+    ASSERT_FALSE(bitmap_less_than_result);
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexReader> missing_reader,
+                         global_index_scan->CreateReader("f0", "lucene", std::nullopt));
+    ASSERT_FALSE(missing_reader);
+
     // Each reader individually should return the same result for Equal("Bob")
     for (const auto& index_reader : index_readers) {
         ASSERT_OK_AND_ASSIGN(auto result,