Merge branch 'main' into feature/columnar-row-ref

Author: lxy-9602

CMakeLists.txt

@@ -290,6 +290,22 @@ if(PAIMON_ENABLE_LUMINA)
             DESTINATION ${CMAKE_INSTALL_LIBDIR})
 endif()
 
+if(PAIMON_ENABLE_LUCENE)
+    set(PAIMON_DICT_DEST "share/paimon/dict")
+
+    install(DIRECTORY ${JIEBA_DICT_DIR}/
+            DESTINATION ${PAIMON_DICT_DEST}
+            FILES_MATCHING
+            PATTERN "jieba.dict.utf8"
+            PATTERN "hmm_model.utf8"
+            PATTERN "idf.utf8"
+            PATTERN "stop_words.utf8"
+            PATTERN "user.dict.utf8"
+            PATTERN "pos_dict"
+            PATTERN ".git*" EXCLUDE
+            PATTERN "*.md" EXCLUDE)
+endif()
+
 install(DIRECTORY ${PROJECT_SOURCE_DIR}/include/
         DESTINATION "include"
         FILES_MATCHING
@@ -389,7 +405,6 @@ if(PAIMON_BUILD_TESTS)
         list(APPEND TEST_STATIC_LINK_LIBS paimon_lucene_index_shared)
         list(APPEND TEST_STATIC_LINK_LIBS "-Wl,--as-needed")
     endif()
-
 endif()
 
 include(CMakePackageConfigHelpers)

cmake_modules/ThirdpartyToolchain.cmake

@@ -322,8 +322,7 @@ macro(build_lucene)
     set(LUCENE_INCLUDE_DIR "${LUCENE_PREFIX}/include")
     # The include directory must exist before it is referenced by a target.
     file(MAKE_DIRECTORY "${LUCENE_INCLUDE_DIR}")
-    include_directories(SYSTEM ${LUCENE_INCLUDE_DIR} ${BOOST_INCLUDE_DIR}
-                        ${BOOST_EXTRA_INCLUDE_DIR})
+    include_directories(SYSTEM ${LUCENE_INCLUDE_DIR} ${BOOST_INCLUDE_DIR})
     add_library(lucene INTERFACE IMPORTED)
     target_include_directories(lucene SYSTEM INTERFACE "${LUCENE_INCLUDE_DIR}")
     target_compile_options(lucene INTERFACE -pthread)
@@ -343,6 +342,43 @@ macro(build_lucene)
     add_dependencies(lucene lucene_ep)
 endmacro()
 
+macro(build_jieba)
+    message(STATUS "Building jieba from source")
+    set(JIEBA_PREFIX "${CMAKE_CURRENT_BINARY_DIR}/jieba_ep-prefix")
+    set(JIEBA_INSTALL "${CMAKE_CURRENT_BINARY_DIR}/jieba_ep-install")
+    set(JIEBA_INCLUDE_DIR "${JIEBA_INSTALL}/include")
+    set(JIEBA_DICT_DIR "${JIEBA_INSTALL}/dict")
+    file(MAKE_DIRECTORY ${JIEBA_INCLUDE_DIR})
+    file(MAKE_DIRECTORY ${JIEBA_DICT_DIR})
+
+    set(JIEBA_CMAKE_ARGS
+        ${EP_COMMON_CMAKE_ARGS} "-DENABLE_TEST=OFF" "-DCPPJIEBA_TOP_LEVEL_PROJECT=OFF"
+        "-DCMAKE_INSTALL_PREFIX=${JIEBA_INSTALL}")
+
+    set(PATCH_FILE "${CMAKE_CURRENT_LIST_DIR}/jieba.diff")
+    externalproject_add(jieba_ep
+                        ${EP_COMMON_OPTIONS}
+                        GIT_REPOSITORY https://github.com/yanyiwu/cppjieba.git
+                        GIT_TAG ${PAIMON_JIEBA_BUILD_VERSION}
+                        GIT_SHALLOW FALSE
+                        GIT_PROGRESS TRUE
+                        GIT_SUBMODULES_RECURSE TRUE
+                        CMAKE_ARGS ${JIEBA_CMAKE_ARGS}
+                        LOG_PATCH ON
+                        PATCH_COMMAND ${CMAKE_COMMAND} -E chdir <SOURCE_DIR> bash -c
+                                      "[ -f .patched ] && echo '<SOURCE_DIR> patch already applied, ignore...' || patch -s -N -p1 -i '${PATCH_FILE}' && touch .patched"
+                        INSTALL_COMMAND bash -c
+                                        "cp -r ${JIEBA_PREFIX}/src/jieba_ep/include/* ${JIEBA_INSTALL}/include/ && cp -r ${JIEBA_PREFIX}/src/jieba_ep/dict/* ${JIEBA_INSTALL}/dict/ && cp -r ${JIEBA_PREFIX}/src/jieba_ep/deps/limonp/include/* ${JIEBA_INSTALL}/include/"
+    )
+
+    # The include directory must exist before it is referenced by a target.
+    include_directories(SYSTEM ${JIEBA_INCLUDE_DIR} ${JIEBA_DICT_DIR})
+    add_library(jieba INTERFACE IMPORTED)
+    target_include_directories(jieba SYSTEM
+                               INTERFACE "${JIEBA_INCLUDE_DIR} ${JIEBA_DICT_DIR}")
+    add_dependencies(jieba jieba_ep)
+endmacro()
+
 macro(build_rapidjson)
     message(STATUS "Building RapidJSON from source")
     set(RAPIDJSON_PREFIX "${CMAKE_CURRENT_BINARY_DIR}/rapidjson_ep-install")
@@ -1272,4 +1308,5 @@ endif()
 if(PAIMON_ENABLE_LUCENE)
     build_boost()
     build_lucene()
+    build_jieba()
 endif()

cmake_modules/jieba.diff

@@ -0,0 +1,16 @@
+diff --git a/include/cppjieba/KeywordExtractor.hpp b/include/cppjieba/KeywordExtractor.hpp
+index 24b2c40..c7c6a94 100644
+--- a/include/cppjieba/KeywordExtractor.hpp
++++ b/include/cppjieba/KeywordExtractor.hpp
+@@ -89,6 +89,11 @@ class KeywordExtractor {
+     std::partial_sort(keywords.begin(), keywords.begin() + topN, keywords.end(), Compare);
+     keywords.resize(topN);
+   }
++
++  const std::unordered_set<std::string>& GetStopWords() const {
++    return stopWords_;
++  }
++
+  private:
+   void LoadIdfDict(const std::string& idfPath) {
+     std::ifstream ifs(idfPath.c_str());

include/paimon/defs.h

@@ -287,6 +287,8 @@ struct PAIMON_EXPORT Options {
     /// "global-index.external-path" - Global index root directory, if not set, the global index
     /// files will be stored under the index directory.
     static const char GLOBAL_INDEX_EXTERNAL_PATH[];
+    /// "scan.tag-name" - Optional tag name used in case of "from-snapshot" scan mode.
+    static const char SCAN_TAG_NAME[];
 };
 
 static constexpr int64_t BATCH_WRITE_COMMIT_IDENTIFIER = std::numeric_limits<int64_t>::max();

include/paimon/read_context.h

@@ -53,8 +53,8 @@ class PAIMON_EXPORT ReadContext {
                 const std::shared_ptr<Executor>& executor,
                 const std::shared_ptr<FileSystem>& specific_file_system,
                 const std::map<std::string, std::string>& fs_scheme_to_identifier_map,
-                const std::map<std::string, std::string>& options, bool enable_prefetch_cache,
-                const CacheConfig& cache_config);
+                const std::map<std::string, std::string>& options,
+                PrefetchCacheMode prefetch_cache_mode, const CacheConfig& cache_config);
     ~ReadContext();
 
     const std::string& GetPath() const {
@@ -116,8 +116,8 @@ class PAIMON_EXPORT ReadContext {
         return specific_file_system_;
     }
 
-    bool EnablePrefetchCache() const {
-        return enable_prefetch_cache_;
+    PrefetchCacheMode GetPrefetchCacheMode() const {
+        return prefetch_cache_mode_;
     }
 
     const CacheConfig& GetCacheConfig() const {
@@ -142,7 +142,7 @@ class PAIMON_EXPORT ReadContext {
     std::shared_ptr<FileSystem> specific_file_system_;
     std::map<std::string, std::string> fs_scheme_to_identifier_map_;
     std::map<std::string, std::string> options_;
-    bool enable_prefetch_cache_;
+    PrefetchCacheMode prefetch_cache_mode_;
     CacheConfig cache_config_;
 };
 
@@ -226,13 +226,13 @@ class PAIMON_EXPORT ReadContextBuilder {
     /// @return Reference to this builder for method chaining.
     ReadContextBuilder& EnablePrefetch(bool enabled);
 
-    /// Enable or disable prefetch cache for read operations.
+    /// Set prefetch cache mode for read operations.
     ///
-    /// When enabled, a prefetch cache is used to prebuffer data ranges before they are needed,
+    /// A prefetch cache is used to prebuffer data ranges before they are needed,
     /// which can improve read performance by reducing redundant I/O operations.
-    /// @param enabled Whether to enable prefetch cache (default: true)
+    /// @param mode (default: PrefetchCacheMode::ALWAYS)
     /// @return Reference to this builder for method chaining.
-    ReadContextBuilder& EnablePrefetchCache(bool enabled);
+    ReadContextBuilder& SetPrefetchCacheMode(PrefetchCacheMode mode);
 
     /// Set the cache configuration for prefetch read operations.
     ///

include/paimon/utils/read_ahead_cache.h

@@ -34,6 +34,24 @@
 
 namespace paimon {
 
+/// PrefetchCacheMode
+/// Cache prefetch switch modes.
+/// Controls whether to enable cache prefetching under different circumstances, such as queries with
+/// predicates or bitmap indexes.
+///
+/// - ALWAYS: Enable cache in all scenarios.
+/// - EXCLUDE_PREDICATE: Disable cache when query has predicates.
+/// - EXCLUDE_BITMAP: Disable cache when using bitmap index.
+/// - EXCLUDE_BITMAP_OR_PREDICATE: Disable cache if query has predicates or bitmap index.
+/// - NEVER: Always disable cache.
+enum class PAIMON_EXPORT PrefetchCacheMode {
+    ALWAYS = 1,
+    EXCLUDE_PREDICATE = 2,
+    EXCLUDE_BITMAP = 3,
+    EXCLUDE_BITMAP_OR_PREDICATE = 4,
+    NEVER = 5
+};
+
 /// Configuration parameters for the read-ahead cache behavior.
 ///
 /// This struct controls various limits and prefetching strategies used by

src/paimon/CMakeLists.txt

@@ -266,6 +266,7 @@ set(PAIMON_CORE_SRCS
     core/table/source/table_read.cpp
     core/table/source/table_scan.cpp
     core/table/source/data_evolution_batch_scan.cpp
+    core/tag/tag.cpp
     core/utils/field_mapping.cpp
     core/utils/fields_comparator.cpp
     core/utils/file_store_path_factory.cpp
@@ -274,7 +275,8 @@ set(PAIMON_CORE_SRCS
     core/utils/partition_path_utils.cpp
     core/utils/primary_key_table_utils.cpp
     core/utils/snapshot_manager.cpp
-    core/utils/special_field_ids.cpp)
+    core/utils/special_field_ids.cpp
+    core/utils/tag_manager.cpp)
 
 add_paimon_lib(paimon
                SOURCES
@@ -565,6 +567,7 @@ if(PAIMON_BUILD_TESTS)
                     core/table/source/split_generator_test.cpp
                     core/table/source/startup_mode_test.cpp
                     core/table/source/table_scan_test.cpp
+                    core/tag/tag_test.cpp
                     core/utils/branch_manager_test.cpp
                     core/utils/field_mapping_test.cpp
                     core/utils/fields_comparator_test.cpp
@@ -574,6 +577,7 @@ if(PAIMON_BUILD_TESTS)
                     core/utils/offset_row_test.cpp
                     core/utils/partition_path_utils_test.cpp
                     core/utils/snapshot_manager_test.cpp
+                    core/utils/tag_manager_test.cpp
                     core/utils/primary_key_table_utils_test.cpp
                     core/utils/index_file_path_factories_test.cpp
                     STATIC_LINK_LIBS

src/paimon/common/defs.cpp

@@ -81,4 +81,5 @@ 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";
 const char Options::GLOBAL_INDEX_EXTERNAL_PATH[] = "global-index.external-path";
+const char Options::SCAN_TAG_NAME[] = "scan.tag-name";
 }  // namespace paimon

src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp

@@ -88,13 +88,14 @@ class ApplyBitmapIndexBatchReaderTest : public ::testing::Test,
             bool enable_prefetch = GetParam();
             if (enable_prefetch) {
                 MockFormatReaderBuilder reader_builder(data, target_type_, batch_size);
-                ASSERT_OK_AND_ASSIGN(file_batch_reader,
-                                     PrefetchFileBatchReaderImpl::Create(
-                                         /*data_file_path=*/"DUMMY", &reader_builder, fs_,
-                                         prefetch_batch_count, batch_size, prefetch_batch_count * 2,
-                                         /*enable_adaptive_prefetch_strategy=*/false, executor_,
-                                         /*initialize_read_ranges=*/true,
-                                         /*enable_prefetch_cache=*/true, CacheConfig(), pool_));
+                ASSERT_OK_AND_ASSIGN(
+                    file_batch_reader,
+                    PrefetchFileBatchReaderImpl::Create(
+                        /*data_file_path=*/"DUMMY", &reader_builder, fs_, prefetch_batch_count,
+                        batch_size, prefetch_batch_count * 2,
+                        /*enable_adaptive_prefetch_strategy=*/false, executor_,
+                        /*initialize_read_ranges=*/true,
+                        /*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS, CacheConfig(), pool_));
             } else {
                 file_batch_reader =
                     std::make_unique<MockFileBatchReader>(data, target_type_, batch_size);

src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp

@@ -45,7 +45,7 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> PrefetchFileBatchReaderImpl
     const std::shared_ptr<FileSystem>& fs, uint32_t prefetch_max_parallel_num, int32_t batch_size,
     uint32_t prefetch_batch_count, bool enable_adaptive_prefetch_strategy,
     const std::shared_ptr<Executor>& executor, bool initialize_read_ranges,
-    bool enable_prefetch_cache, const CacheConfig& cache_config,
+    PrefetchCacheMode prefetch_cache_mode, const CacheConfig& cache_config,
     const std::shared_ptr<MemoryPool>& pool) {
     if (prefetch_max_parallel_num == 0) {
         return Status::Invalid("prefetch max parallel num should be greater than 0.");
@@ -67,7 +67,7 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> PrefetchFileBatchReaderImpl
     }
 
     std::shared_ptr<ReadAheadCache> cache;
-    if (enable_prefetch_cache) {
+    if (prefetch_cache_mode != PrefetchCacheMode::NEVER) {
         PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input_stream, fs->Open(data_file_path));
         cache = std::make_shared<ReadAheadCache>(input_stream, cache_config, pool);
     }
@@ -102,9 +102,9 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> PrefetchFileBatchReaderImpl
     }
     uint32_t prefetch_queue_capacity = prefetch_batch_count / readers.size();
 
-    auto reader = std::unique_ptr<PrefetchFileBatchReaderImpl>(
-        new PrefetchFileBatchReaderImpl(readers, batch_size, prefetch_queue_capacity,
-                                        enable_adaptive_prefetch_strategy, executor, cache));
+    auto reader = std::unique_ptr<PrefetchFileBatchReaderImpl>(new PrefetchFileBatchReaderImpl(
+        readers, batch_size, prefetch_queue_capacity, enable_adaptive_prefetch_strategy, executor,
+        cache, prefetch_cache_mode));
     if (initialize_read_ranges) {
         // normally initialize read ranges should be false, as set read schema will refresh read
         // ranges, and set read schema will always be called before read.
@@ -116,11 +116,13 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> PrefetchFileBatchReaderImpl
 PrefetchFileBatchReaderImpl::PrefetchFileBatchReaderImpl(
     const std::vector<std::shared_ptr<PrefetchFileBatchReader>>& readers, int32_t batch_size,
     uint32_t prefetch_queue_capacity, bool enable_adaptive_prefetch_strategy,
-    const std::shared_ptr<Executor>& executor, const std::shared_ptr<ReadAheadCache>& cache)
+    const std::shared_ptr<Executor>& executor, const std::shared_ptr<ReadAheadCache>& cache,
+    PrefetchCacheMode cache_mode)
     : readers_(std::move(readers)),
       batch_size_(batch_size),
       executor_(executor),
       cache_(cache),
+      cache_mode_(cache_mode),
       prefetch_queue_capacity_(prefetch_queue_capacity),
       enable_adaptive_prefetch_strategy_(enable_adaptive_prefetch_strategy) {
     for (size_t i = 0; i < readers_.size(); i++) {
@@ -146,6 +148,7 @@ Status PrefetchFileBatchReaderImpl::SetReadSchema(
         PAIMON_RETURN_NOT_OK(reader->SetReadSchema(c_schema.get(), predicate, selection_bitmap));
     }
     selection_bitmap_ = selection_bitmap;
+    predicate_ = predicate;
     return RefreshReadRanges();
 }
 
@@ -275,10 +278,28 @@ Status PrefetchFileBatchReaderImpl::CleanUp() {
     return Status::OK();
 }
 
+bool PrefetchFileBatchReaderImpl::NeedInitCache() const {
+    switch (cache_mode_) {
+        case PrefetchCacheMode::NEVER:
+            return false;
+        case PrefetchCacheMode::EXCLUDE_PREDICATE:
+            return predicate_ == nullptr;
+        case PrefetchCacheMode::EXCLUDE_BITMAP:
+            return selection_bitmap_ == std::nullopt;
+        case PrefetchCacheMode::EXCLUDE_BITMAP_OR_PREDICATE:
+            return predicate_ == nullptr && selection_bitmap_ == std::nullopt;
+        case PrefetchCacheMode::ALWAYS:
+            return true;
+        default:
+            assert(false);
+            return true;
+    }
+}
+
 void PrefetchFileBatchReaderImpl::Workloop() {
     std::vector<std::future<void>> futures;
     futures.resize(readers_.size());
-    if (cache_) {
+    if (cache_ && NeedInitCache()) {
         auto read_ranges = readers_[0]->PreBufferRange();
         if (read_ranges.ok()) {
             std::vector<ByteRange> ranges;
@@ -288,11 +309,9 @@ void PrefetchFileBatchReaderImpl::Workloop() {
             auto s = cache_->Init(std::move(ranges));
             if (!s.ok()) {
                 SetReadStatus(s);
-                return;
             }
         } else {
             SetReadStatus(read_ranges.status());
-            return;
         }
     }