Merge branch 'main' into full-text-seach

Author: lxy-9602

build_and_package.sh

@@ -18,11 +18,13 @@ set -euo pipefail
 
 SOURCE_ROOT=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
 OUTPUT_DIR="$SOURCE_ROOT/output"
-BUILD_TYPE="release"
+BUILD_TYPE="Release"
+BUILD_DIR="$SOURCE_ROOT/build-release"
 BUILD_NAME="paimon-cpp"
 MAKE_CLEAN=false
 PACKAGE=false
-CMAKE_OPTIONS=""
+CMAKE_OPTIONS=()
+JOBS=""
 
 show_help() {
     cat << EOF
@@ -33,15 +35,16 @@ Options:
   -d, --debug       Build debug version
   -c, --clean       Clean build directory before building
   -p, --package     Package creation
+  -j, --jobs <num>  Number of parallel jobs for building (default: auto-detect)
   -h, --help        Show this help message
 
 CMake Options:
   Any unrecognized options will be passed directly to CMake.
   You can specify multiple CMake options.
 
 Examples:
-  $0 -r -p -DPAIMON_BUILD_SHARED=ON -DPAIMON_BUILD_STATIC=OFF
-  $0 --debug --clean --package
+  $0 -r -p -j 8 -DPAIMON_BUILD_SHARED=ON -DPAIMON_BUILD_STATIC=OFF
+  $0 --debug --clean --package --jobs 4
 
 EOF
 }
@@ -50,11 +53,14 @@ while [[ $# -gt 0 ]]; do
     case "$1" in
         -r|--release)
             BUILD_TYPE="Release"
+            BUILD_DIR="$SOURCE_ROOT/build-release"
+            BUILD_NAME="paimon-cpp"
             shift
             ;;
         -d|--debug)
             BUILD_TYPE="Debug"
-            BUILD_NAME=$BUILD_NAME"-debug"
+            BUILD_DIR="$SOURCE_ROOT/build-debug"
+            BUILD_NAME="paimon-cpp-debug"
             shift
             ;;
         -c|--clean)
@@ -65,27 +71,40 @@ while [[ $# -gt 0 ]]; do
             PACKAGE=true
             shift
             ;;
+        -j|--jobs)
+            shift
+            if [[ $# -gt 0 && $1 =~ ^[0-9]+$ ]]; then
+                JOBS="$1"
+                shift
+            else
+                echo "Error: -j/--jobs requires a numeric argument" >&2
+                exit 1
+            fi
+            ;;
         -h|--help)
             show_help
             exit 0
             ;;
         *)
             # All remaining parameters are CMake options
-            CMAKE_OPTIONS="$CMAKE_OPTIONS $1"
+            CMAKE_OPTIONS+=("$1")
             shift
             ;;
     esac
 done
 
-CMAKE_OPTIONS=$(echo "$CMAKE_OPTIONS" | xargs)
-
 echo "========== Build Configuration =========="
 echo "Build Type: $BUILD_TYPE"
 echo "Package Name: $BUILD_NAME"
 echo "Clean Build: $MAKE_CLEAN"
 echo "Package: $PACKAGE"
-if [ -n "$CMAKE_OPTIONS" ]; then
-    echo "CMake Options: $CMAKE_OPTIONS"
+if [ -n "$JOBS" ]; then
+    echo "Parallel Jobs: $JOBS"
+else
+    echo "Parallel Jobs: auto-detect"
+fi
+if [ ${#CMAKE_OPTIONS[@]} -gt 0 ]; then
+    echo "CMake Options: ${CMAKE_OPTIONS[*]}"
 else
     echo "CMake Options: None"
 fi
@@ -95,33 +114,39 @@ echo "Step 1: Downloading dependencies..."
 "$SOURCE_ROOT"/third_party/download_dependencies.sh
 
 echo "Step 2: Building Paimon..."
-BUILD_DIR="$SOURCE_ROOT/build-$BUILD_TYPE"
 PACKAGE_DIR="$OUTPUT_DIR/$BUILD_NAME"
 
 if [ "$MAKE_CLEAN" = true ]; then
+    echo "Cleaning build directory: $BUILD_DIR"
     rm -rf "$BUILD_DIR"
 fi
 mkdir -p "$BUILD_DIR"
 cd "$BUILD_DIR"
+
 CMAKE_ARGS=(
+    -G "Ninja"
     -DCMAKE_BUILD_TYPE="$BUILD_TYPE"
     -DCMAKE_INSTALL_PREFIX="$PACKAGE_DIR"
 )
 
-if [ -n "$CMAKE_OPTIONS" ]; then
-    CMAKE_ARGS+=("$CMAKE_OPTIONS")
+if [ ${#CMAKE_OPTIONS[@]} -gt 0 ]; then
+    CMAKE_ARGS+=("${CMAKE_OPTIONS[@]}")
 fi
 
 cmake "${CMAKE_ARGS[@]}" ..
 
-JOBS=$(nproc 2>/dev/null || echo 4)
-make -j"$JOBS"
+# Set default JOBS if not specified
+if [ -z "$JOBS" ]; then
+    JOBS=$(nproc 2>/dev/null || echo 4)
+fi
+
+ninja -j"$JOBS"
 
 if [ "$PACKAGE" = true ]; then
     echo "Step 3: Packaging..."
     mkdir -p "$OUTPUT_DIR"
     cd "$BUILD_DIR"
-    make install
+    ninja install
     tar -czvf "$OUTPUT_DIR/$BUILD_NAME.tar.gz" -C "$OUTPUT_DIR" "$BUILD_NAME"
     echo "Package created: $OUTPUT_DIR/$BUILD_NAME.tar.gz"
 else

cmake_modules/orc.diff

@@ -30,6 +30,20 @@ index a914e5f26..efe1d4933 100644
    };
 
    // Specializations for char
+diff --git a/c++/include/orc/Reader.hh b/c++/include/orc/Reader.hh
+index b015b6491..585e50ec5 100644
+--- a/c++/include/orc/Reader.hh
++++ b/c++/include/orc/Reader.hh
+@@ -659,6 +659,9 @@ namespace orc {
+     virtual void preBuffer(const std::vector<uint32_t>& stripes,
+                            const std::list<uint64_t>& includeTypes) = 0;
+
++    virtual std::vector<std::pair<uint64_t, uint64_t>> preBufferRange(
++        const std::vector<uint32_t>& stripes, const std::list<uint64_t>& includeTypes) = 0;
++
+     /**
+      * Release cached entries whose right boundary is less than or equal to the given boundary.
+      * @param boundary the boundary value to release cache entries
 diff --git a/c++/src/ColumnReader.cc b/c++/src/ColumnReader.cc
 index af434c37c..08393259c 100644
 --- a/c++/src/ColumnReader.cc
@@ -328,6 +342,86 @@ index ed7fee737..a8ee8a67c 100644
      reserve(newSize);
      if (newSize > currentSize_) {
        memset(buf_ + currentSize_, 0, newSize - currentSize_);
+diff --git a/c++/src/Reader.cc b/c++/src/Reader.cc
+index c93c62f6c..2a821b622 100644
+--- a/c++/src/Reader.cc
++++ b/c++/src/Reader.cc
+@@ -1531,8 +1531,8 @@ namespace orc {
+     }
+   }
+
+-  void ReaderImpl::preBuffer(const std::vector<uint32_t>& stripes,
+-                             const std::list<uint64_t>& includeTypes) {
++  std::vector<std::pair<uint64_t, uint64_t>> ReaderImpl::preBufferRange(
++      const std::vector<uint32_t>& stripes, const std::list<uint64_t>& includeTypes) {
+     std::vector<uint32_t> newStripes;
+     for (auto stripe : stripes) {
+       if (stripe < static_cast<uint32_t>(footer_->stripes_size())) newStripes.push_back(stripe);
+@@ -1544,7 +1544,7 @@ namespace orc {
+     }
+
+     if (newStripes.empty() || newIncludeTypes.empty()) {
+-      return;
++      return {};
+     }
+
+     orc::RowReaderOptions rowReaderOptions;
+@@ -1553,7 +1553,7 @@ namespace orc {
+     std::vector<bool> selectedColumns;
+     columnSelector.updateSelected(selectedColumns, rowReaderOptions);
+
+-    std::vector<ReadRange> ranges;
++    std::vector<std::pair<uint64_t, uint64_t>> ranges;
+     ranges.reserve(newIncludeTypes.size());
+     for (auto stripe : newStripes) {
+       // get stripe information
+@@ -1598,17 +1598,23 @@ namespace orc {
+
+         offset += stream.length();
+       }
++    }
++    return ranges;
++  }
+
+-      {
+-        std::lock_guard<std::mutex> lock(contents_->readCacheMutex);
+-
+-        if (!contents_->readCache) {
+-          contents_->readCache = std::make_shared<ReadRangeCache>(
+-              getStream(), options_.getCacheOptions(), contents_->pool, contents_->readerMetrics);
+-        }
+-        contents_->readCache->cache(std::move(ranges));
+-      }
++  void ReaderImpl::preBuffer(const std::vector<uint32_t>& stripes,
++                             const std::list<uint64_t>& includeTypes) {
++    auto ranges = preBufferRange(stripes, includeTypes);
++    std::vector<ReadRange> read_ranges;
++    for (const auto& range : ranges) {
++      read_ranges.emplace_back(range.first, range.second);
++    }
++    std::lock_guard<std::mutex> lock(contents_->readCacheMutex);
++    if (!contents_->readCache) {
++      contents_->readCache = std::make_shared<ReadRangeCache>(
++          getStream(), options_.getCacheOptions(), contents_->pool, contents_->readerMetrics);
+     }
++    contents_->readCache->cache(std::move(read_ranges));
+   }
+
+   RowReader::~RowReader() {
+diff --git a/c++/src/Reader.hh b/c++/src/Reader.hh
+index 39ca73967..13da45a49 100644
+--- a/c++/src/Reader.hh
++++ b/c++/src/Reader.hh
+@@ -387,6 +387,9 @@ namespace orc {
+     std::map<uint32_t, BloomFilterIndex> getBloomFilters(
+         uint32_t stripeIndex, const std::set<uint32_t>& included) const override;
+
++    std::vector<std::pair<uint64_t, uint64_t>> preBufferRange(
++        const std::vector<uint32_t>& stripes, const std::list<uint64_t>& includeTypes) override;
++
+     void preBuffer(const std::vector<uint32_t>& stripes,
+                    const std::list<uint64_t>& includeTypes) override;
+     void releaseBuffer(uint64_t boundary) override;
 diff --git a/cmake_modules/ThirdpartyToolchain.cmake b/cmake_modules/ThirdpartyToolchain.cmake
 index 9b2c829c7..434841224 100644
 --- a/cmake_modules/ThirdpartyToolchain.cmake

include/paimon/catalog/catalog.h

@@ -46,9 +46,13 @@ class PAIMON_EXPORT Catalog {
     ///
     /// @param root_path Path to the root directory where the catalog is located.
     /// @param options Configuration options for catalog initialization.
+    /// @param file_system Specifies the file system for file operations.
+    ///                    If not set, use default file system (configured in
+    ///                    `Options::FILE_SYSTEM`)
     /// @return A result containing a unique pointer to a `Catalog` instance, or an error status.
     static Result<std::unique_ptr<Catalog>> Create(
-        const std::string& root_path, const std::map<std::string, std::string>& options);
+        const std::string& root_path, const std::map<std::string, std::string>& options,
+        const std::shared_ptr<FileSystem>& file_system = nullptr);
 
     virtual ~Catalog() = default;
 
@@ -124,6 +128,16 @@ class PAIMON_EXPORT Catalog {
     /// @return A string representing the expected location of the table.
     virtual std::string GetTableLocation(const Identifier& identifier) const = 0;
 
+    /// Returns the root path of the catalog.
+    ///
+    /// @return A string representing the root path of the catalog.
+    virtual std::string GetRootPath() const = 0;
+
+    /// Returns the file system used by the catalog.
+    ///
+    /// @return A shared pointer to the file system instance.
+    virtual std::shared_ptr<FileSystem> GetFileSystem() const = 0;
+
     /// Loads the latest schema of a specified table.
     ///
     /// @note System tables will not be supported.

include/paimon/commit_context.h

@@ -39,6 +39,7 @@ class PAIMON_EXPORT CommitContext {
                   bool ignore_empty_commit, bool use_rest_catalog_commit,
                   const std::shared_ptr<MemoryPool>& memory_pool,
                   const std::shared_ptr<Executor>& executor,
+                  const std::shared_ptr<FileSystem>& specific_file_system,
                   const std::map<std::string, std::string>& options);
     ~CommitContext();
 
@@ -66,6 +67,10 @@ class PAIMON_EXPORT CommitContext {
         return executor_;
     }
 
+    std::shared_ptr<FileSystem> GetSpecificFileSystem() const {
+        return specific_file_system_;
+    }
+
     const std::map<std::string, std::string>& GetOptions() const {
         return options_;
     }
@@ -77,6 +82,7 @@ class PAIMON_EXPORT CommitContext {
     bool use_rest_catalog_commit_;
     std::shared_ptr<MemoryPool> memory_pool_;
     std::shared_ptr<Executor> executor_;
+    std::shared_ptr<FileSystem> specific_file_system_;
     std::map<std::string, std::string> options_;
 };
 
@@ -129,6 +135,15 @@ class PAIMON_EXPORT CommitContextBuilder {
     /// @return Reference to this builder for method chaining.
     CommitContextBuilder& WithExecutor(const std::shared_ptr<Executor>& executor);
 
+    /// Sets a custom file system instance to be used for all file operations in this commit
+    /// context.
+    /// This bypasses the global file system registry and uses the provided implementation directly.
+    ///
+    /// @param file_system The file system to use.
+    /// @return Reference to this builder for method chaining.
+    /// @note If not set, use default file system (configured in `Options::FILE_SYSTEM`)
+    CommitContextBuilder& WithFileSystem(const std::shared_ptr<FileSystem>& file_system);
+
     /// Build and return a `CommitContext` instance with input validation.
     /// @return Result containing the constructed `CommitContext` or an error status.
     Result<std::unique_ptr<CommitContext>> Finish();

include/paimon/global_index/global_index_write_task.h

@@ -44,13 +44,16 @@ class PAIMON_EXPORT GlobalIndexWriteTask {
     /// @param options      Index-specific configuration (e.g., false positive rate for bloom
     /// filters).
     /// @param pool         Memory pool for temporary allocations during index construction.
-    //                      If `nullptr`, the system's default memory pool will be used.
+    ///                     If `nullptr`, the system's default memory pool will be used.
+    /// @param file_system  Specifies the file system for file operations.
+    ///                     If `nullptr`, use default file system.
     /// @return A `Result` containing a shared pointer to the `CommitMessage` with index metadata,
     ///         or an error if indexing fails (e.g., unsupported type, I/O error).
     static Result<std::shared_ptr<CommitMessage>> WriteIndex(
         const std::string& table_path, const std::string& field_name, const std::string& index_type,
         const std::shared_ptr<IndexedSplit>& indexed_split,
-        const std::map<std::string, std::string>& options, const std::shared_ptr<MemoryPool>& pool);
+        const std::map<std::string, std::string>& options, const std::shared_ptr<MemoryPool>& pool,
+        const std::shared_ptr<FileSystem>& file_system = nullptr);
 };
 
 }  // namespace paimon

include/paimon/migrate/file_meta_utils.h

@@ -56,12 +56,15 @@ class PAIMON_EXPORT FileMetaUtils {
     ///                         tables. Use empty map for non-partitioned tables.
     /// @param options Set a configuration options map to set some option entries which are not
     ///                defined in the table schema or whose values you want to overwrite.
+    /// @param file_system  Specifies the file system for file operations.
+    ///                     If `nullptr`, use default file system.
     /// @return Result containing a unique pointer to the generated `CommitMessage`,
     ///         or an error status if the migration cannot be performed.
     static Result<std::unique_ptr<CommitMessage>> GenerateCommitMessage(
         const std::vector<std::string>& src_data_files, const std::string& dst_table_path,
         const std::map<std::string, std::string>& partition_values,
-        const std::map<std::string, std::string>& options);
+        const std::map<std::string, std::string>& options,
+        const std::shared_ptr<FileSystem>& file_system = nullptr);
 };
 
 }  // namespace paimon