feat(mosaic): support Mosaic file format

.github/workflows/build_and_test.yaml

@@ -75,7 +75,6 @@ jobs:
           - name: asan-ubsan-x86_64
             build_args: --enable_asan --enable_ubsan
           - name: tsan-x86_64
-            skip_rust: true
             build_args: --enable_tsan
           - name: gcc-debug-aarch64
             runner: ubuntu-24.04-arm
@@ -96,7 +95,6 @@ jobs:
             build_args: --build_type Release
           - name: tsan-aarch64
             runner: ubuntu-24.04-arm
-            skip_rust: true
             build_args: --enable_tsan
     steps:
       - name: Checkout paimon-cpp
@@ -108,8 +106,7 @@ jobs:
         uses: ./.github/actions/setup-ccache
         with:
           cache-key-prefix: ccache-${{ matrix.name }}
-      - name: Install Rust toolchain (tantivy-fts)
-        if: ${{ !matrix.skip_rust }}
+      - name: Install Rust toolchain (Mosaic and tantivy-fts)
         shell: bash
         run: ci/scripts/setup_rust.sh
       - name: Install HTTP and TLS development dependencies

.github/workflows/gcc8_test.yaml

@@ -60,6 +60,9 @@ jobs:
           ls -la
       - name: Checkout paimon-cpp
         uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v6.0.1
+      - name: Install Rust toolchain (Mosaic)
+        shell: bash
+        run: ci/scripts/setup_rust.sh
       - name: Setup ccache
         uses: ./.github/actions/setup-ccache
         with:

.github/workflows/release_candidate.yaml

@@ -131,7 +131,7 @@ jobs:
           name: source-archive
           path: release/ci
 
-      - name: Install Rust toolchain (tantivy-fts)
+      - name: Install Rust toolchain (Mosaic and tantivy-fts)
         shell: bash
         run: ci/scripts/setup_rust.sh
 

CMakeLists.txt

@@ -60,6 +60,7 @@ option(PAIMON_USE_UBSAN "Use Undefined Behavior Sanitizer" OFF)
 option(PAIMON_USE_CXX11_ABI "Use C++11 ABI" ON)
 option(PAIMON_ENABLE_AVRO "Whether to enable avro file format" ON)
 option(PAIMON_ENABLE_ORC "Whether to enable orc file format" ON)
+option(PAIMON_ENABLE_MOSAIC "Whether to enable mosaic file format (Rust FFI)" OFF)
 option(PAIMON_ENABLE_JINDO "Whether to enable jindo file system" OFF)
 option(PAIMON_ENABLE_S3 "Whether to enable S3 file system" OFF)
 option(PAIMON_ENABLE_NETWORK_TESTS
@@ -86,6 +87,9 @@ endif()
 if(PAIMON_ENABLE_AVRO)
     add_definitions(-DPAIMON_ENABLE_AVRO)
 endif()
+if(PAIMON_ENABLE_MOSAIC)
+    add_definitions(-DPAIMON_ENABLE_MOSAIC)
+endif()
 if(PAIMON_ENABLE_JINDO)
     add_definitions(-DPAIMON_ENABLE_JINDO)
 endif()
@@ -388,6 +392,11 @@ if(PAIMON_BUILD_TESTS OR PAIMON_BUILD_BENCHMARKS)
     paimon_link_libraries_whole_archive(PAIMON_PARQUET_FILE_FORMAT_STATIC_LINK_LIBS
                                         paimon_parquet_file_format_static)
 
+    if(PAIMON_ENABLE_MOSAIC)
+        paimon_link_libraries_whole_archive(PAIMON_MOSAIC_FILE_FORMAT_STATIC_LINK_LIBS
+                                            paimon_mosaic_file_format_static)
+    endif()
+
     if(PAIMON_ENABLE_ORC)
         paimon_link_libraries_whole_archive(PAIMON_ORC_FILE_FORMAT_STATIC_LINK_LIBS
                                             paimon_orc_file_format_static)
@@ -439,6 +448,14 @@ if(PAIMON_BUILD_TESTS)
     paimon_link_libraries_whole_archive(PAIMON_PARQUET_FILE_FORMAT_STATIC_LINK_LIBS
                                         paimon_parquet_file_format_static)
 
+    if(PAIMON_ENABLE_MOSAIC)
+        paimon_link_libraries_no_as_needed(TEST_PLUGIN_LINK_LIBS
+                                           paimon_mosaic_file_format_shared)
+        list(APPEND TEST_STATIC_LINK_LIBS ${TEST_PLUGIN_LINK_LIBS})
+        paimon_link_libraries_whole_archive(PAIMON_MOSAIC_FILE_FORMAT_STATIC_LINK_LIBS
+                                            paimon_mosaic_file_format_static)
+    endif()
+
     if(PAIMON_ENABLE_ORC)
         paimon_link_libraries_no_as_needed(TEST_PLUGIN_LINK_LIBS
                                            paimon_orc_file_format_shared)
@@ -520,6 +537,9 @@ add_subdirectory(src/paimon/format/blob)
 add_subdirectory(src/paimon/format/orc)
 add_subdirectory(src/paimon/format/parquet)
 add_subdirectory(src/paimon/format/avro)
+if(PAIMON_ENABLE_MOSAIC)
+    add_subdirectory(src/paimon/format/mosaic)
+endif()
 if(PAIMON_ENABLE_LUMINA)
     add_subdirectory(src/paimon/global_index/lumina)
 endif()

ci/scripts/build_paimon.sh

@@ -147,6 +147,7 @@ CMAKE_ARGS=(
     "-G Ninja"
     "-DCMAKE_BUILD_TYPE=${build_type}"
     "-DPAIMON_BUILD_TESTS=ON"
+    "-DPAIMON_ENABLE_MOSAIC=ON"
     "-DPAIMON_ENABLE_JINDO=ON"
     "-DPAIMON_ENABLE_S3=ON"
     "-DPAIMON_ENABLE_LUMINA=${ENABLE_LUMINA}"

ci/scripts/setup_rust.sh

@@ -15,12 +15,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-# Install the Rust toolchain + cbindgen required to build the
-# tantivy-fts FFI crate (crates/tantivy_ffi) from CI.
+# Install the Rust toolchain used by the Mosaic and tantivy-fts FFI builds, plus cbindgen required
+# by tantivy-fts.
 #
 # The dev container (see .devcontainer/) already has these preinstalled;
-# this script is for the GitHub Actions runners. Called by
-# .github/workflows/build_and_test.yaml before ci/scripts/build_paimon.sh.
+# this script is for the GitHub Actions runners and is called before ci/scripts/build_paimon.sh.
 #
 # Idempotent: a second invocation is a no-op when the tools already exist.
 

cmake_modules/ThirdpartyToolchain.cmake

@@ -91,6 +91,18 @@ else()
     endif()
 endif()
 
+if(DEFINED ENV{PAIMON_MOSAIC_URL})
+    set(MOSAIC_SOURCE_URL "$ENV{PAIMON_MOSAIC_URL}")
+else()
+    if(EXISTS "${THIRDPARTY_DIR}/${PAIMON_MOSAIC_PKG_NAME}")
+        set_urls(MOSAIC_SOURCE_URL "${THIRDPARTY_DIR}/${PAIMON_MOSAIC_PKG_NAME}")
+    else()
+        set_urls(MOSAIC_SOURCE_URL
+                 "https://downloads.apache.org/paimon/paimon-mosaic-${PAIMON_MOSAIC_BUILD_VERSION}/${PAIMON_MOSAIC_PKG_NAME}"
+        )
+    endif()
+endif()
+
 if(DEFINED ENV{PAIMON_RAPIDJSON_URL})
     set(RAPIDJSON_SOURCE_URL "$ENV{PAIMON_RAPIDJSON_URL}")
 else()
@@ -1314,6 +1326,50 @@ macro(build_lumina)
     install(FILES "${LUMINA_DYNAMIC_LIB}" DESTINATION ${CMAKE_INSTALL_LIBDIR})
 endmacro()
 
+macro(build_mosaic)
+    message(STATUS "Building Apache Paimon Mosaic Rust FFI from source")
+    find_program(PAIMON_CARGO_EXECUTABLE cargo REQUIRED)
+
+    set(MOSAIC_PREFIX "${CMAKE_CURRENT_BINARY_DIR}/mosaic_ep-install")
+    set(MOSAIC_INCLUDE_DIR "${MOSAIC_PREFIX}/include")
+    set(MOSAIC_LIB_DIR "${MOSAIC_PREFIX}/${CMAKE_INSTALL_LIBDIR}")
+    set(MOSAIC_DYNAMIC_LIB
+        "${MOSAIC_LIB_DIR}/${CMAKE_SHARED_LIBRARY_PREFIX}paimon_mosaic_ffi${CMAKE_SHARED_LIBRARY_SUFFIX}"
+    )
+    set(MOSAIC_CARGO_TARGET_DIR "${CMAKE_CURRENT_BINARY_DIR}/mosaic_ep-cargo")
+    set(MOSAIC_CARGO_DYNAMIC_LIB
+        "${MOSAIC_CARGO_TARGET_DIR}/release/${CMAKE_SHARED_LIBRARY_PREFIX}paimon_mosaic_ffi${CMAKE_SHARED_LIBRARY_SUFFIX}"
+    )
+
+    file(MAKE_DIRECTORY "${MOSAIC_INCLUDE_DIR}")
+    file(MAKE_DIRECTORY "${MOSAIC_LIB_DIR}")
+
+    externalproject_add(mosaic_ep
+                        URL ${MOSAIC_SOURCE_URL}
+                        URL_HASH "SHA256=${PAIMON_MOSAIC_BUILD_SHA256_CHECKSUM}"
+                        ${THIRDPARTY_LOG_OPTIONS}
+                        CONFIGURE_COMMAND ""
+                        BUILD_COMMAND ${CMAKE_COMMAND} -E env
+                                      CARGO_TARGET_DIR=${MOSAIC_CARGO_TARGET_DIR}
+                                      ${PAIMON_CARGO_EXECUTABLE} build --release
+                                      --manifest-path <SOURCE_DIR>/ffi/Cargo.toml
+                        COMMAND ${CMAKE_COMMAND} -E copy_if_different
+                                ${MOSAIC_CARGO_DYNAMIC_LIB} ${MOSAIC_DYNAMIC_LIB}
+                        INSTALL_COMMAND ${CMAKE_COMMAND} -E copy_directory
+                                        <SOURCE_DIR>/include ${MOSAIC_INCLUDE_DIR}
+                        BUILD_BYPRODUCTS "${MOSAIC_DYNAMIC_LIB}")
+
+    add_library(paimon_mosaic_ffi SHARED IMPORTED GLOBAL)
+    set_target_properties(paimon_mosaic_ffi
+                          PROPERTIES IMPORTED_LOCATION "${MOSAIC_DYNAMIC_LIB}"
+                                     IMPORTED_NO_SONAME TRUE
+                                     INTERFACE_INCLUDE_DIRECTORIES
+                                     "${MOSAIC_INCLUDE_DIR}")
+    add_dependencies(paimon_mosaic_ffi mosaic_ep)
+
+    install(FILES "${MOSAIC_DYNAMIC_LIB}" DESTINATION ${CMAKE_INSTALL_LIBDIR})
+endmacro()
+
 macro(build_jindosdk_nextarch)
     message(STATUS "Building jindosdk-nextarch from local source")
 
@@ -1953,6 +2009,9 @@ paimon_warn_if_mixed_arrow_dependencies()
 resolve_dependency(TBB)
 resolve_dependency(glog)
 
+if(PAIMON_ENABLE_MOSAIC)
+    build_mosaic()
+endif()
 if(PAIMON_ENABLE_AVRO)
     resolve_dependency(Avro)
 endif()

include/paimon/defs.h

@@ -156,6 +156,10 @@ struct PAIMON_EXPORT Options {
     /// 9, but the read and write speed will significantly decrease. Default value is 1.
     static const char FILE_COMPRESSION_ZSTD_LEVEL[];
 
+    /// "file.block-size" - File block size of format, default value of orc stripe is 64 MB,
+    /// parquet row group is 128 MB, and Mosaic row group is 256 MB.
+    static const char FILE_BLOCK_SIZE[];
+
     /// "manifest.target-file-size" - Suggested file size of a manifest file.
     /// Default value is 8MB.
     static const char MANIFEST_TARGET_FILE_SIZE[];

src/paimon/common/data/binary_array.h

@@ -52,7 +52,7 @@ class MemorySegment;
 /// [size(int)] + [null bits(4-byte word boundaries)] + [values or offset&length] + [variable length
 /// part].
 
-class BinaryArray final : public BinarySection, public InternalArray {
+class PAIMON_EXPORT BinaryArray final : public BinarySection, public InternalArray {
  public:
     BinaryArray() = default;
 

src/paimon/common/data/binary_row.h

@@ -59,7 +59,7 @@ class MemoryPool;
 /// @note: Unlike the Java implementation where variable-length data may span multiple
 /// MemorySegments, in this C++ implementation both the fixed-length part and the
 /// variable-length part reside within a single MemorySegment.
-class BinaryRow final : public BinarySection, public InternalRow, public DataSetters {
+class PAIMON_EXPORT BinaryRow final : public BinarySection, public InternalRow, public DataSetters {
  public:
     BinaryRow() : BinaryRow(0) {}
     explicit BinaryRow(int32_t arity);

src/paimon/common/defs.cpp

@@ -46,6 +46,7 @@ const char Options::PAGE_SIZE[] = "page-size";
 const char Options::PARTITION_DEFAULT_NAME[] = "partition.default-name";
 const char Options::FILE_COMPRESSION[] = "file.compression";
 const char Options::FILE_COMPRESSION_ZSTD_LEVEL[] = "file.compression.zstd-level";
+const char Options::FILE_BLOCK_SIZE[] = "file.block-size";
 const char Options::MANIFEST_TARGET_FILE_SIZE[] = "manifest.target-file-size";
 const char Options::MANIFEST_FORMAT[] = "manifest.format";
 const char Options::MANIFEST_COMPRESSION[] = "manifest.compression";

src/paimon/common/reader/blob_fallback_batch_reader.cpp

@@ -31,6 +31,7 @@
 #include "paimon/common/metrics/metrics_impl.h"
 #include "paimon/common/reader/reader_utils.h"
 #include "paimon/common/table/special_fields.h"
+#include "paimon/common/utils/arrow/arrow_utils.h"
 #include "paimon/common/utils/arrow/mem_utils.h"
 #include "paimon/common/utils/arrow/status_utils.h"
 #include "paimon/common/utils/checked_cast.h"
@@ -238,11 +239,9 @@ Result<std::shared_ptr<arrow::Array>> BlobFallbackBatchReader::AssembleRowIdRun(
             break;
         }
     }
-    if (pieces.size() == 1 && pieces[0]->offset() == 0) {
+    if (pieces.size() == 1) {
         return pieces[0];
     }
-    // Concatenate flattens non-zero offsets left by Slice, so the exported batch honors the
-    // zero-offset BatchReader contract.
     PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> concat_array,
                                       arrow::Concatenate(pieces, arrow_pool_.get()));
     return concat_array;
@@ -303,11 +302,9 @@ Result<std::shared_ptr<arrow::Array>> BlobFallbackBatchReader::AssembleColumn(
         }
         run_start = run_end;
     }
-    if (pieces.size() == 1 && pieces[0]->offset() == 0) {
+    if (pieces.size() == 1) {
         return pieces[0];
     }
-    // Concatenate flattens non-zero offsets left by Slice, so the exported batch honors the
-    // zero-offset BatchReader contract.
     PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> concat_array,
                                       arrow::Concatenate(pieces, arrow_pool_.get()));
     return concat_array;
@@ -360,12 +357,14 @@ Result<BatchReader::ReadBatch> BlobFallbackBatchReader::NextBatch() {
                                AssembleColumn(field_idx, group_choice, group_chunks));
         columns.push_back(std::move(column));
     }
-    PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::StructArray> target_array,
+    PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::StructArray> array,
                                       arrow::StructArray::Make(columns, read_schema_->fields()));
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> normalized_array,
+                           ArrowUtils::NormalizeArrayOffsets(array, arrow_pool_.get()));
     std::unique_ptr<ArrowArray> c_array = std::make_unique<ArrowArray>();
     std::unique_ptr<ArrowSchema> c_schema = std::make_unique<ArrowSchema>();
     PAIMON_RETURN_NOT_OK_FROM_ARROW(
-        arrow::ExportArray(*target_array, c_array.get(), c_schema.get()));
+        arrow::ExportArray(*normalized_array, c_array.get(), c_schema.get()));
     return std::make_pair(std::move(c_array), std::move(c_schema));
 }
 

src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp

@@ -30,6 +30,8 @@
 #include "paimon/common/io/cache_input_stream.h"
 #include "paimon/common/metrics/metrics_impl.h"
 #include "paimon/common/reader/reader_utils.h"
+#include "paimon/common/utils/arrow/arrow_utils.h"
+#include "paimon/common/utils/arrow/mem_utils.h"
 #include "paimon/common/utils/arrow/status_utils.h"
 #include "paimon/common/utils/read_ahead_cache.h"
 #include "paimon/common/utils/scope_guard.h"
@@ -121,7 +123,7 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> PrefetchFileBatchReaderImpl
 
     auto reader = std::unique_ptr<PrefetchFileBatchReaderImpl>(
         new PrefetchFileBatchReaderImpl(readers, batch_size, prefetch_queue_capacity,
-                                        enable_adaptive_prefetch_strategy, executor, cache));
+                                        enable_adaptive_prefetch_strategy, executor, cache, pool));
     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.
@@ -133,11 +135,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,
+    const std::shared_ptr<MemoryPool>& pool)
     : readers_(std::move(readers)),
       batch_size_(batch_size),
       executor_(executor),
       cache_(cache),
+      arrow_pool_(GetArrowPool(pool)),
       prefetch_queue_capacity_(prefetch_queue_capacity),
       enable_adaptive_prefetch_strategy_(enable_adaptive_prefetch_strategy) {
     for (size_t i = 0; i < readers_.size(); i++) {
@@ -470,12 +474,15 @@ Status PrefetchFileBatchReaderImpl::HandleReadResult(
         } else if (slice_end < c_array->length) {
             // partially out of range, data before read_range.second has been effectively consumed
             readers_pos_[reader_idx]->store(read_range.second);
-            PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> src_array,
+            PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> array,
                                               arrow::ImportArray(c_array.get(), c_schema.get()));
-            auto array = src_array->Slice(0, slice_end);
+            std::shared_ptr<arrow::Array> sliced_array = array->Slice(/*offset=*/0, slice_end);
+            PAIMON_ASSIGN_OR_RAISE(
+                std::shared_ptr<arrow::Array> normalized_array,
+                ArrowUtils::NormalizeArrayOffsets(sliced_array, arrow_pool_.get()));
             PAIMON_RETURN_NOT_OK_FROM_ARROW(
-                arrow::ExportArray(*array, c_array.get(), c_schema.get()));
-            bitmap.RemoveRange(slice_end, src_array->length());
+                arrow::ExportArray(*normalized_array, c_array.get(), c_schema.get()));
+            bitmap.RemoveRange(slice_end, array->length());
             global_row_ids =
                 std::vector<uint64_t>(global_row_ids.begin(), global_row_ids.begin() + slice_end);
         } else {

src/paimon/common/reader/prefetch_file_batch_reader_impl.h

@@ -45,8 +45,13 @@
 
 struct ArrowSchema;
 
+namespace arrow {
+class MemoryPool;
+}  // namespace arrow
+
 namespace paimon {
 
+class MemoryPool;
 class ReaderBuilder;
 class FileSystem;
 class Executor;
@@ -113,7 +118,8 @@ class PrefetchFileBatchReaderImpl : public PrefetchFileBatchReader {
     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,
+        const std::shared_ptr<MemoryPool>& pool);
 
     Status CleanUp();
     void Workloop();
@@ -160,6 +166,7 @@ class PrefetchFileBatchReaderImpl : public PrefetchFileBatchReader {
     std::condition_variable cv_;
     std::shared_ptr<Executor> executor_;
     std::shared_ptr<ReadAheadCache> cache_;
+    std::unique_ptr<arrow::MemoryPool> arrow_pool_;
 
     mutable std::shared_mutex rw_mutex_;
     std::unique_ptr<std::thread> background_thread_;