From 202fb96d4de38d60599ba7be40130b1cc0dc7eac Mon Sep 17 00:00:00 2001 From: Zhang Jiawei <30893610+zjw1111@users.noreply.github.com> Date: Wed, 10 Jun 2026 10:23:55 +0800 Subject: [PATCH 01/11] fix(build): make package config relocatable (#306) --- CMakeLists.txt | 10 ++- PaimonConfig.cmake.in | 132 +-------------------------------- ci/scripts/build_paimon.sh | 14 ++-- cmake_modules/BuildUtils.cmake | 4 +- docs/source/build_system.rst | 32 ++++---- examples/CMakeLists.txt | 22 +++--- 6 files changed, 45 insertions(+), 169 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index 91a39823..7488e8aa 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -97,9 +97,9 @@ set(PAIMON_SOURCE_DIR ${PROJECT_SOURCE_DIR}) set(PAIMON_BINARY_DIR ${PROJECT_BINARY_DIR}) set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} "${CMAKE_CURRENT_SOURCE_DIR}/cmake_modules") set(BUILD_SUPPORT_DIR "${CMAKE_SOURCE_DIR}/build_support") -set(PAIMON_CMAKE_INSTALL_DIR "${CMAKE_INSTALL_LIBDIR}/cmake/${PROJECT_NAME}") include(GNUInstallDirs) +set(PAIMON_CMAKE_INSTALL_DIR "${CMAKE_INSTALL_LIBDIR}/cmake/Paimon") include(DefineOptions) if(PAIMON_OPTIONAL_INSTALL) @@ -422,11 +422,11 @@ write_basic_package_version_file( configure_package_config_file("${CMAKE_CURRENT_SOURCE_DIR}/PaimonConfig.cmake.in" "${CMAKE_CURRENT_BINARY_DIR}/PaimonConfig.cmake" - INSTALL_DESTINATION lib/cmake/Paimon) + INSTALL_DESTINATION ${PAIMON_CMAKE_INSTALL_DIR}) install(FILES "${CMAKE_CURRENT_BINARY_DIR}/PaimonConfig.cmake" "${CMAKE_CURRENT_BINARY_DIR}/PaimonConfigVersion.cmake" - DESTINATION lib/cmake/Paimon) + DESTINATION ${PAIMON_CMAKE_INSTALL_DIR}) config_summary_message() @@ -449,3 +449,7 @@ add_subdirectory(src/paimon/global_index/lucene) add_subdirectory(src/paimon/testing/mock) add_subdirectory(src/paimon/testing/utils) add_subdirectory(test/inte) + +install(EXPORT PaimonTargets + NAMESPACE Paimon:: + DESTINATION ${PAIMON_CMAKE_INSTALL_DIR}) diff --git a/PaimonConfig.cmake.in b/PaimonConfig.cmake.in index e56d9d55..559417ae 100644 --- a/PaimonConfig.cmake.in +++ b/PaimonConfig.cmake.in @@ -14,134 +14,8 @@ # See the License for the specific language governing permissions and # limitations under the License. -# Main library -add_library(paimon_shared SHARED IMPORTED) -set_target_properties(paimon_shared PROPERTIES - IMPORTED_LOCATION "@CMAKE_INSTALL_PREFIX@/@CMAKE_INSTALL_LIBDIR@/libpaimon.so" - INTERFACE_INCLUDE_DIRECTORIES "@CMAKE_INSTALL_PREFIX@/include" -) -add_library(paimon_static STATIC IMPORTED) -set_target_properties(paimon_static PROPERTIES - IMPORTED_LOCATION "@CMAKE_INSTALL_PREFIX@/@CMAKE_INSTALL_LIBDIR@/libpaimon.a" - INTERFACE_INCLUDE_DIRECTORIES "@CMAKE_INSTALL_PREFIX@/include" -) +@PACKAGE_INIT@ -# paimon_parquet_file_format -add_library(paimon_parquet_file_format_shared SHARED IMPORTED) -set_target_properties(paimon_parquet_file_format_shared PROPERTIES - IMPORTED_LOCATION "@CMAKE_INSTALL_PREFIX@/@CMAKE_INSTALL_LIBDIR@/libpaimon_parquet_file_format.so" - INTERFACE_INCLUDE_DIRECTORIES "@CMAKE_INSTALL_PREFIX@/include" -) -add_library(paimon_parquet_file_format_static STATIC IMPORTED) -set_target_properties(paimon_parquet_file_format_static PROPERTIES - IMPORTED_LOCATION "@CMAKE_INSTALL_PREFIX@/@CMAKE_INSTALL_LIBDIR@/libpaimon_parquet_file_format.a" - INTERFACE_INCLUDE_DIRECTORIES "@CMAKE_INSTALL_PREFIX@/include" -) +include("${CMAKE_CURRENT_LIST_DIR}/PaimonTargets.cmake") -# paimon_orc_file_format -add_library(paimon_orc_file_format_shared SHARED IMPORTED) -set_target_properties(paimon_orc_file_format_shared PROPERTIES - IMPORTED_LOCATION "@CMAKE_INSTALL_PREFIX@/@CMAKE_INSTALL_LIBDIR@/libpaimon_orc_file_format.so" - INTERFACE_INCLUDE_DIRECTORIES "@CMAKE_INSTALL_PREFIX@/include" -) -add_library(paimon_orc_file_format_static STATIC IMPORTED) -set_target_properties(paimon_orc_file_format_static PROPERTIES - IMPORTED_LOCATION "@CMAKE_INSTALL_PREFIX@/@CMAKE_INSTALL_LIBDIR@/libpaimon_orc_file_format.a" - INTERFACE_INCLUDE_DIRECTORIES "@CMAKE_INSTALL_PREFIX@/include" -) - -# paimon_local_file_system -add_library(paimon_local_file_system_shared SHARED IMPORTED) -set_target_properties(paimon_local_file_system_shared PROPERTIES - IMPORTED_LOCATION "@CMAKE_INSTALL_PREFIX@/@CMAKE_INSTALL_LIBDIR@/libpaimon_local_file_system.so" - INTERFACE_INCLUDE_DIRECTORIES "@CMAKE_INSTALL_PREFIX@/include" -) -add_library(paimon_local_file_system_static STATIC IMPORTED) -set_target_properties(paimon_local_file_system_static PROPERTIES - IMPORTED_LOCATION "@CMAKE_INSTALL_PREFIX@/@CMAKE_INSTALL_LIBDIR@/libpaimon_local_file_system.a" - INTERFACE_INCLUDE_DIRECTORIES "@CMAKE_INSTALL_PREFIX@/include" -) - -# paimon_avro_file_format -add_library(paimon_avro_file_format_shared SHARED IMPORTED) -set_target_properties(paimon_avro_file_format_shared PROPERTIES - IMPORTED_LOCATION "@CMAKE_INSTALL_PREFIX@/@CMAKE_INSTALL_LIBDIR@/libpaimon_avro_file_format.so" - INTERFACE_INCLUDE_DIRECTORIES "@CMAKE_INSTALL_PREFIX@/include" -) -add_library(paimon_avro_file_format_static STATIC IMPORTED) -set_target_properties(paimon_avro_file_format_static PROPERTIES - IMPORTED_LOCATION "@CMAKE_INSTALL_PREFIX@/@CMAKE_INSTALL_LIBDIR@/libpaimon_avro_file_format.a" - INTERFACE_INCLUDE_DIRECTORIES "@CMAKE_INSTALL_PREFIX@/include" -) - -# paimon_blob_file_format -add_library(paimon_blob_file_format_shared SHARED IMPORTED) -set_target_properties(paimon_blob_file_format_shared PROPERTIES - IMPORTED_LOCATION "@CMAKE_INSTALL_PREFIX@/@CMAKE_INSTALL_LIBDIR@/libpaimon_blob_file_format.so" - INTERFACE_INCLUDE_DIRECTORIES "@CMAKE_INSTALL_PREFIX@/include" -) -add_library(paimon_blob_file_format_static STATIC IMPORTED) -set_target_properties(paimon_blob_file_format_static PROPERTIES - IMPORTED_LOCATION "@CMAKE_INSTALL_PREFIX@/@CMAKE_INSTALL_LIBDIR@/libpaimon_blob_file_format.a" - INTERFACE_INCLUDE_DIRECTORIES "@CMAKE_INSTALL_PREFIX@/include" -) - -# paimon_file_index -add_library(paimon_file_index_shared SHARED IMPORTED) -set_target_properties(paimon_file_index_shared PROPERTIES - IMPORTED_LOCATION "@CMAKE_INSTALL_PREFIX@/@CMAKE_INSTALL_LIBDIR@/libpaimon_file_index.so" - INTERFACE_INCLUDE_DIRECTORIES "@CMAKE_INSTALL_PREFIX@/include" -) -add_library(paimon_file_index_static STATIC IMPORTED) -set_target_properties(paimon_file_index_static PROPERTIES - IMPORTED_LOCATION "@CMAKE_INSTALL_PREFIX@/@CMAKE_INSTALL_LIBDIR@/libpaimon_file_index.a" - INTERFACE_INCLUDE_DIRECTORIES "@CMAKE_INSTALL_PREFIX@/include" -) - -# paimon_global_index -add_library(paimon_global_index_shared SHARED IMPORTED) -set_target_properties(paimon_global_index_shared PROPERTIES - IMPORTED_LOCATION "@CMAKE_INSTALL_PREFIX@/@CMAKE_INSTALL_LIBDIR@/libpaimon_global_index.so" - INTERFACE_INCLUDE_DIRECTORIES "@CMAKE_INSTALL_PREFIX@/include" -) -add_library(paimon_global_index_static STATIC IMPORTED) -set_target_properties(paimon_global_index_static PROPERTIES - IMPORTED_LOCATION "@CMAKE_INSTALL_PREFIX@/@CMAKE_INSTALL_LIBDIR@/libpaimon_global_index.a" - INTERFACE_INCLUDE_DIRECTORIES "@CMAKE_INSTALL_PREFIX@/include" -) - -# paimon_jindo_file_system -add_library(paimon_jindo_file_system_shared SHARED IMPORTED) -set_target_properties(paimon_jindo_file_system_shared PROPERTIES - IMPORTED_LOCATION "@CMAKE_INSTALL_PREFIX@/@CMAKE_INSTALL_LIBDIR@/libpaimon_jindo_file_system.so" - INTERFACE_INCLUDE_DIRECTORIES "@CMAKE_INSTALL_PREFIX@/include" -) -add_library(paimon_jindo_file_system_static STATIC IMPORTED) -set_target_properties(paimon_jindo_file_system_static PROPERTIES - IMPORTED_LOCATION "@CMAKE_INSTALL_PREFIX@/@CMAKE_INSTALL_LIBDIR@/libpaimon_jindo_file_system.a" - INTERFACE_INCLUDE_DIRECTORIES "@CMAKE_INSTALL_PREFIX@/include" -) - -# paimon_lance_file_format -add_library(paimon_lance_file_format_shared SHARED IMPORTED) -set_target_properties(paimon_lance_file_format_shared PROPERTIES - IMPORTED_LOCATION "@CMAKE_INSTALL_PREFIX@/@CMAKE_INSTALL_LIBDIR@/libpaimon_lance_file_format.so" - INTERFACE_INCLUDE_DIRECTORIES "@CMAKE_INSTALL_PREFIX@/include" -) -add_library(paimon_lance_file_format_static STATIC IMPORTED) -set_target_properties(paimon_lance_file_format_static PROPERTIES - IMPORTED_LOCATION "@CMAKE_INSTALL_PREFIX@/@CMAKE_INSTALL_LIBDIR@/libpaimon_lance_file_format.a" - INTERFACE_INCLUDE_DIRECTORIES "@CMAKE_INSTALL_PREFIX@/include" -) - -# paimon_lumina_index -add_library(paimon_lumina_index_shared SHARED IMPORTED) -set_target_properties(paimon_lumina_index_shared PROPERTIES - IMPORTED_LOCATION "@CMAKE_INSTALL_PREFIX@/@CMAKE_INSTALL_LIBDIR@/libpaimon_lumina_index.so" - INTERFACE_INCLUDE_DIRECTORIES "@CMAKE_INSTALL_PREFIX@/include" -) -add_library(paimon_lumina_index_static STATIC IMPORTED) -set_target_properties(paimon_lumina_index_static PROPERTIES - IMPORTED_LOCATION "@CMAKE_INSTALL_PREFIX@/@CMAKE_INSTALL_LIBDIR@/libpaimon_lumina_index.a" - INTERFACE_INCLUDE_DIRECTORIES "@CMAKE_INSTALL_PREFIX@/include" -) +check_required_components(Paimon) diff --git a/ci/scripts/build_paimon.sh b/ci/scripts/build_paimon.sh index 235434a7..16e6fe04 100755 --- a/ci/scripts/build_paimon.sh +++ b/ci/scripts/build_paimon.sh @@ -21,7 +21,7 @@ source_dir=${1} enable_sanitizer=${2:-false} check_clang_tidy=${3:-false} build_type=${4:-Debug} -build_dir=${1}/build +build_dir="${source_dir}/build" # Display ccache status if available if command -v ccache &> /dev/null; then @@ -32,8 +32,8 @@ else echo "=== ccache not found, compiling without cache acceleration ===" fi -mkdir ${build_dir} -pushd ${build_dir} +mkdir -p "${build_dir}" +pushd "${build_dir}" ENABLE_LUMINA="ON" if [[ "${CC:-}" == *"gcc-8"* ]] || [[ "${CXX:-}" == *"g++-8"* ]]; then @@ -56,9 +56,9 @@ if [[ "${enable_sanitizer}" == "true" ]]; then ) fi -cmake "${CMAKE_ARGS[@]}" ${source_dir} -cmake --build . -- -j$(nproc) -ctest --output-on-failure -j $(nproc) +cmake "${CMAKE_ARGS[@]}" "${source_dir}" +cmake --build . -- -j "$(nproc)" +ctest --output-on-failure -j "$(nproc)" if [[ "${check_clang_tidy}" == "true" ]]; then cmake --build . --target check-clang-tidy @@ -72,4 +72,4 @@ fi popd -rm -rf ${build_dir} +rm -rf "${build_dir}" diff --git a/cmake_modules/BuildUtils.cmake b/cmake_modules/BuildUtils.cmake index fa83b076..21ddc16e 100644 --- a/cmake_modules/BuildUtils.cmake +++ b/cmake_modules/BuildUtils.cmake @@ -150,7 +150,7 @@ function(add_paimon_lib LIB_NAME) -Wl,--gc-sections) install(TARGETS ${LIB_NAME}_shared ${INSTALL_IS_OPTIONAL} - EXPORT ${LIB_NAME}_targets + EXPORT PaimonTargets RUNTIME DESTINATION ${RUNTIME_INSTALL_DIR} LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR} ARCHIVE DESTINATION ${CMAKE_INSTALL_LIBDIR} @@ -197,7 +197,7 @@ function(add_paimon_lib LIB_NAME) PUBLIC "$") install(TARGETS ${LIB_NAME}_static ${INSTALL_IS_OPTIONAL} - EXPORT ${LIB_NAME}_targets + EXPORT PaimonTargets RUNTIME DESTINATION ${RUNTIME_INSTALL_DIR} LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR} ARCHIVE DESTINATION ${CMAKE_INSTALL_LIBDIR} diff --git a/docs/source/build_system.rst b/docs/source/build_system.rst index 6a61b68a..8bcf7147 100644 --- a/docs/source/build_system.rst +++ b/docs/source/build_system.rst @@ -52,10 +52,10 @@ for data format and file system. find_package(Paimon REQUIRED) add_executable(my_example my_example.cc) - target_link_libraries(my_example PRIVATE arrow_shared - paimon_shared - paimon_parquet_file_format_shared - paimon_local_file_system_shared) + target_link_libraries(my_example PRIVATE Arrow::arrow_shared + Paimon::paimon_shared + Paimon::paimon_parquet_file_format_shared + Paimon::paimon_local_file_system_shared) Available variables and targets ------------------------------- @@ -64,10 +64,10 @@ The directive ``find_package(Paimon REQUIRED)`` instructs CMake to locate a Paimon C++ installation on your system. If successful, it sets ``Paimon_FOUND`` to true if the Paimon C++ libraries were found. -It also defines the following linkable targets (plain strings, not variables): +It also defines the following linkable imported targets: -* ``paimon_shared`` links to the Paimon shared libraries -* ``paimon_static`` links to the Paimon static libraries +* ``Paimon::paimon_shared`` links to the Paimon shared libraries +* ``Paimon::paimon_static`` links to the Paimon static libraries In most cases, it is recommended to use the Paimon shared libraries. @@ -78,21 +78,21 @@ Paimon provides a set of built-in optional plugins that you can link to as neede - File format plugins: - - ``paimon_parquet_file_format_shared`` / ``paimon_parquet_file_format_static`` - - ``paimon_orc_file_format_shared`` / ``paimon_orc_file_format_static`` - - ``paimon_avro_file_format_shared`` / ``paimon_avro_file_format_static`` - - ``paimon_blob_file_format_shared`` / ``paimon_blob_file_format_static`` + - ``Paimon::paimon_parquet_file_format_shared`` / ``Paimon::paimon_parquet_file_format_static`` + - ``Paimon::paimon_orc_file_format_shared`` / ``Paimon::paimon_orc_file_format_static`` + - ``Paimon::paimon_avro_file_format_shared`` / ``Paimon::paimon_avro_file_format_static`` + - ``Paimon::paimon_blob_file_format_shared`` / ``Paimon::paimon_blob_file_format_static`` - File system plugins: - - ``paimon_local_file_system_shared`` / ``paimon_local_file_system_static`` - - ``paimon_jindo_file_system_shared`` / ``paimon_jindo_file_system_static`` + - ``Paimon::paimon_local_file_system_shared`` / ``Paimon::paimon_local_file_system_static`` + - ``Paimon::paimon_jindo_file_system_shared`` / ``Paimon::paimon_jindo_file_system_static`` - Index plugins: - - ``paimon_file_index_shared`` / ``paimon_file_index_static`` - - ``paimon_lumina_index_shared`` / ``paimon_lumina_index_static`` - - ``paimon_lucene_index_shared`` / ``paimon_lucene_index_static`` + - ``Paimon::paimon_file_index_shared`` / ``Paimon::paimon_file_index_static`` + - ``Paimon::paimon_lumina_index_shared`` / ``Paimon::paimon_lumina_index_static`` + - ``Paimon::paimon_lucene_index_shared`` / ``Paimon::paimon_lucene_index_static`` .. note:: diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index 082cf040..4fe7a8be 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -26,20 +26,18 @@ find_package(Paimon CONFIG REQUIRED) add_executable(read_write_demo read_write_demo.cpp) add_executable(clean_demo clean_demo.cpp) -include_directories(${PAIMON_INCLUDE_DIRS}) - set(CMAKE_EXE_LINKER_FLAGS "-Wl,--no-as-needed ${CMAKE_EXE_LINKER_FLAGS}") target_link_libraries(read_write_demo - PRIVATE arrow_shared - paimon_shared - paimon_parquet_file_format_shared - paimon_orc_file_format_shared - paimon_local_file_system_shared) + PRIVATE Arrow::arrow_shared + Paimon::paimon_shared + Paimon::paimon_parquet_file_format_shared + Paimon::paimon_orc_file_format_shared + Paimon::paimon_local_file_system_shared) target_link_libraries(clean_demo - PRIVATE arrow_shared - paimon_shared - paimon_parquet_file_format_shared - paimon_orc_file_format_shared - paimon_local_file_system_shared) + PRIVATE Arrow::arrow_shared + Paimon::paimon_shared + Paimon::paimon_parquet_file_format_shared + Paimon::paimon_orc_file_format_shared + Paimon::paimon_local_file_system_shared) From f98b491fc2a3ac3913e62df577d08975f5b21cbc Mon Sep 17 00:00:00 2001 From: Yonghao Fang Date: Wed, 3 Jun 2026 09:18:58 +0800 Subject: [PATCH 02/11] fix: fix thread-safe problem for file store path factory (#323) --- .../core/utils/file_store_path_factory.cpp | 28 +++++- .../core/utils/file_store_path_factory.h | 5 +- .../utils/file_store_path_factory_test.cpp | 89 +++++++++++++++++++ 3 files changed, 117 insertions(+), 5 deletions(-) diff --git a/src/paimon/core/utils/file_store_path_factory.cpp b/src/paimon/core/utils/file_store_path_factory.cpp index 97a3bead..d17b223f 100644 --- a/src/paimon/core/utils/file_store_path_factory.cpp +++ b/src/paimon/core/utils/file_store_path_factory.cpp @@ -221,24 +221,44 @@ Result FileStorePathFactory::GetPartitionString(const BinaryRow& pa if (partition.GetSizeInBytes() == 0) { return Status::Invalid("invalid binary row partition"); } - auto iter = row_to_str_cache_.find(partition); - if (PAIMON_LIKELY(iter != row_to_str_cache_.end())) { - return iter->second; + { + std::shared_lock read_lock(cache_mutex_); + auto iter = row_to_str_cache_.find(partition); + if (PAIMON_LIKELY(iter != row_to_str_cache_.end())) { + return iter->second; + } } + std::vector> part_values; PAIMON_ASSIGN_OR_RAISE(part_values, partition_computer_->GeneratePartitionVector(partition)); PAIMON_ASSIGN_OR_RAISE(std::string part_str, PartitionPathUtils::GeneratePartitionPath(part_values)); + + std::unique_lock write_lock(cache_mutex_); + auto iter = row_to_str_cache_.find(partition); + if (PAIMON_LIKELY(iter != row_to_str_cache_.end())) { + return iter->second; + } return row_to_str_cache_.insert({partition, part_str}).first->second; } Result FileStorePathFactory::ToBinaryRow( const std::map& partition) const { + { + std::shared_lock read_lock(cache_mutex_); + auto iter = map_to_row_cache_.find(partition); + if (PAIMON_LIKELY(iter != map_to_row_cache_.end())) { + return iter->second; + } + } + + PAIMON_ASSIGN_OR_RAISE(BinaryRow row, partition_computer_->ToBinaryRow(partition)); + + std::unique_lock write_lock(cache_mutex_); auto iter = map_to_row_cache_.find(partition); if (PAIMON_LIKELY(iter != map_to_row_cache_.end())) { return iter->second; } - PAIMON_ASSIGN_OR_RAISE(BinaryRow row, partition_computer_->ToBinaryRow(partition)); return map_to_row_cache_.insert({partition, row}).first->second; } diff --git a/src/paimon/core/utils/file_store_path_factory.h b/src/paimon/core/utils/file_store_path_factory.h index 8e07bcec..1890f2bc 100644 --- a/src/paimon/core/utils/file_store_path_factory.h +++ b/src/paimon/core/utils/file_store_path_factory.h @@ -22,6 +22,8 @@ #include #include #include +#include +#include #include #include #include @@ -99,7 +101,7 @@ class FileStorePathFactory : public std::enable_shared_from_this GetPartitionString(const BinaryRow& partition) const; std::string NewManifestFile() const { return PathUtil::JoinPath( @@ -169,6 +171,7 @@ class FileStorePathFactory : public std::enable_shared_from_this, BinaryRow> map_to_row_cache_; mutable std::unordered_map row_to_str_cache_; + mutable std::shared_mutex cache_mutex_; }; } // namespace paimon diff --git a/src/paimon/core/utils/file_store_path_factory_test.cpp b/src/paimon/core/utils/file_store_path_factory_test.cpp index f76f8368..41a9635e 100644 --- a/src/paimon/core/utils/file_store_path_factory_test.cpp +++ b/src/paimon/core/utils/file_store_path_factory_test.cpp @@ -18,7 +18,10 @@ #include "paimon/core/utils/file_store_path_factory.h" +#include +#include #include +#include #include #include "arrow/type.h" @@ -360,6 +363,92 @@ TEST_F(FileStorePathFactoryTest, TestToBinaryRowAndToPartitionString) { } } +TEST_F(FileStorePathFactoryTest, TestConcurrentToBinaryRowAndGetPartitionString) { + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + + arrow::FieldVector fields = { + arrow::field("f0", arrow::boolean()), arrow::field("f1", arrow::int8()), + arrow::field("f2", arrow::int8()), arrow::field("f3", arrow::int16()), + arrow::field("f4", arrow::int16()), arrow::field("f5", arrow::int32()), + arrow::field("hr", arrow::int32()), arrow::field("f7", arrow::int64()), + arrow::field("dt", arrow::utf8()), arrow::field("f9", arrow::float32()), + arrow::field("f10", arrow::float64()), arrow::field("f11", arrow::utf8()), + arrow::field("f12", arrow::binary()), arrow::field("non-partition-field", arrow::int32())}; + auto schema = arrow::schema(fields); + + ASSERT_OK_AND_ASSIGN( + std::shared_ptr path_factory, + FileStorePathFactory::Create(dir->Str(), schema, {"dt", "hr"}, "default", + /*identifier=*/"mock_format", /*data_file_prefix=*/"data-", + /*legacy_partition_name_enabled=*/true, + /*external_paths=*/std::vector(), + /*global_index_external_path=*/std::nullopt, + /*index_file_in_data_file_dir=*/false, mem_pool_)); + + const std::vector, std::string>> test_cases = { + {{{"dt", "20211224"}, {"hr", "23"}}, "dt=20211224/hr=23/"}, + {{{"dt", "default"}, {"hr", "17"}}, "dt=default/hr=17/"}, + {{{"dt", "20240812"}, {"hr", "5"}}, "dt=20240812/hr=5/"}, + {{{"dt", " a "}, {"hr", "22"}}, "dt= a /hr=22/"}, + }; + + std::atomic has_error = false; + std::mutex error_mutex; + std::string first_error; + + constexpr int32_t kThreadNum = 8; + constexpr int32_t kIterationsPerThread = 500; + std::vector threads; + threads.reserve(kThreadNum); + + for (int32_t thread_id = 0; thread_id < kThreadNum; ++thread_id) { + threads.emplace_back([&, thread_id]() { + for (int32_t i = 0; i < kIterationsPerThread; ++i) { + const auto& [partition_map, expected] = + test_cases[(thread_id + i) % test_cases.size()]; + + auto row_result = path_factory->ToBinaryRow(partition_map); + if (!row_result.ok()) { + std::lock_guard lock(error_mutex); + has_error.store(true, std::memory_order_relaxed); + if (first_error.empty()) { + first_error = "ToBinaryRow failed: " + row_result.status().ToString(); + } + return; + } + + auto partition_str_result = path_factory->GetPartitionString(row_result.value()); + if (!partition_str_result.ok()) { + std::lock_guard lock(error_mutex); + has_error.store(true, std::memory_order_relaxed); + if (first_error.empty()) { + first_error = "GetPartitionString failed: " + + partition_str_result.status().ToString(); + } + return; + } + + if (partition_str_result.value() != expected) { + std::lock_guard lock(error_mutex); + has_error.store(true, std::memory_order_relaxed); + if (first_error.empty()) { + first_error = "Unexpected partition string, expected=" + expected + + ", actual=" + partition_str_result.value(); + } + return; + } + } + }); + } + + for (auto& thread : threads) { + thread.join(); + } + + ASSERT_FALSE(has_error.load(std::memory_order_relaxed)) << first_error; +} + TEST_F(FileStorePathFactoryTest, TestCreateIndexFileFactory) { { // test without external path & index_file_in_data_file_dir = false From 77842642a6e94646abfdcdbe5a41ad3a5d15870a Mon Sep 17 00:00:00 2001 From: lxy <38709059+lxy-9602@users.noreply.github.com> Date: Fri, 29 May 2026 10:36:31 +0800 Subject: [PATCH 03/11] feat(lumina): support null values in vector column during index building (#310) --- .../lumina/lumina_global_index.cpp | 71 +++++-- .../global_index/lumina/lumina_global_index.h | 2 + .../lumina/lumina_global_index_test.cpp | 199 +++++++++++++++++- test/inte/global_index_test.cpp | 41 +++- 4 files changed, 285 insertions(+), 28 deletions(-) diff --git a/src/paimon/global_index/lumina/lumina_global_index.cpp b/src/paimon/global_index/lumina/lumina_global_index.cpp index 5976acd0..2d13aa12 100644 --- a/src/paimon/global_index/lumina/lumina_global_index.cpp +++ b/src/paimon/global_index/lumina/lumina_global_index.cpp @@ -179,8 +179,12 @@ Result> LuminaGlobalIndex::CreateReader( class LuminaDataset : public ::lumina::api::Dataset { public: LuminaDataset(int64_t element_count, uint32_t dimension, - const std::vector>& array_vec) - : element_count_(element_count), dimension_(dimension), array_vec_(array_vec) {} + const std::vector>& array_vec, + const std::vector& start_ids) + : element_count_(element_count), + dimension_(dimension), + array_vec_(array_vec), + start_ids_(start_ids) {} uint32_t Dim() const noexcept override { return dimension_; @@ -202,8 +206,8 @@ class LuminaDataset : public ::lumina::api::Dataset { vector_buffer.resize(value_array_length); memcpy(vector_buffer.data(), value_ptr, sizeof(float) * value_array_length); id_buffer.resize(element_count); - std::iota(id_buffer.begin(), id_buffer.end(), id_); - id_ += element_count; + std::iota(id_buffer.begin(), id_buffer.end(), + static_cast<::lumina::core::vector_id_t>(start_ids_[cursor_])); // release the array when copy to vector_buffer value_array.reset(); @@ -215,8 +219,8 @@ class LuminaDataset : public ::lumina::api::Dataset { int64_t element_count_; uint32_t dimension_; std::vector> array_vec_; + std::vector start_ids_; size_t cursor_ = 0; - ::lumina::core::vector_id_t id_ = 0; }; LuminaIndexWriter::LuminaIndexWriter(const std::string& field_name, @@ -256,35 +260,62 @@ Status LuminaIndexWriter::AddBatch(::ArrowArray* arrow_array, auto list_field_array = std::dynamic_pointer_cast(field_array); CHECK_NOT_NULL(list_field_array, "invalid input array in LuminaIndexWriter, field array must be list array"); - auto value_array = std::dynamic_pointer_cast(list_field_array->values()); - CHECK_NOT_NULL( - value_array, - "invalid input array in LuminaIndexWriter, field value array must be float array"); - if (value_array->null_count() != 0) { - return Status::Invalid("field value array in LuminaIndexWriter is invalid, must not null"); - } - if (value_array->length() != field_length * dimension_) { - return Status::Invalid(fmt::format( - "invalid input array in LuminaIndexWriter, length of field array [{}] multiplied " - "dimension [{}] must match length of field value array [{}]", - field_length, dimension_, value_array->length())); + + // Split into contiguous non-null segments, skipping null rows in the list field. + int64_t segment_start = -1; + for (int64_t i = 0; i <= field_length; i++) { + bool is_null = (i < field_length) && list_field_array->IsNull(i); + bool is_end = (i == field_length); + + if (!is_null && !is_end && segment_start == -1) { + segment_start = i; + } + + if ((is_null || is_end) && segment_start != -1) { + int64_t segment_len = i - segment_start; + // Use value_offset to precisely locate the float range for this segment + auto value_start_offset = list_field_array->value_offset(segment_start); + auto value_end_offset = list_field_array->value_offset(segment_start + segment_len); + int64_t value_length = value_end_offset - value_start_offset; + auto sliced_values = std::dynamic_pointer_cast( + list_field_array->values()->Slice(value_start_offset, value_length)); + CHECK_NOT_NULL(sliced_values, + "invalid sliced value array in LuminaIndexWriter, must be float array"); + if (sliced_values->null_count() != 0) { + return Status::Invalid( + "field value array in LuminaIndexWriter is invalid, must not null"); + } + if (sliced_values->length() != segment_len * static_cast(dimension_)) { + return Status::Invalid(fmt::format( + "invalid input array in LuminaIndexWriter, length of field array [{}] " + "multiplied dimension [{}] must match length of field value array [{}]", + segment_len, dimension_, sliced_values->length())); + } + array_vec_.push_back(std::move(sliced_values)); + array_start_ids_.push_back(count_ + segment_start); + indexed_count_ += segment_len; + segment_start = -1; + } } + count_ += array->length(); - array_vec_.push_back(std::move(value_array)); return Status::OK(); } Result> LuminaIndexWriter::Finish() { + if (indexed_count_ == 0) { + return std::vector(); + } ::lumina::core::MemoryResourceConfig memory_resource(pool_.get()); PAIMON_ASSIGN_OR_RAISE_FROM_LUMINA( ::lumina::api::LuminaBuilder builder, ::lumina::api::LuminaBuilder::Create(builder_options_, memory_resource)); // pretrain - LuminaDataset dataset1(count_, dimension_, array_vec_); + LuminaDataset dataset1(indexed_count_, dimension_, array_vec_, array_start_ids_); PAIMON_RETURN_NOT_OK_FROM_LUMINA(builder.PretrainFrom(dataset1)); // insert data - LuminaDataset dataset2(count_, dimension_, array_vec_); + LuminaDataset dataset2(indexed_count_, dimension_, array_vec_, array_start_ids_); std::vector>().swap(array_vec_); PAIMON_RETURN_NOT_OK_FROM_LUMINA(builder.InsertFrom(dataset2)); diff --git a/src/paimon/global_index/lumina/lumina_global_index.h b/src/paimon/global_index/lumina/lumina_global_index.h index 281f5185..572361a0 100644 --- a/src/paimon/global_index/lumina/lumina_global_index.h +++ b/src/paimon/global_index/lumina/lumina_global_index.h @@ -93,6 +93,7 @@ class LuminaIndexWriter : public GlobalIndexWriter { private: int64_t count_ = 0; + int64_t indexed_count_ = 0; std::shared_ptr pool_; std::string field_name_; std::shared_ptr arrow_type_; @@ -102,6 +103,7 @@ class LuminaIndexWriter : public GlobalIndexWriter { ::lumina::api::IOOptions io_options_; std::map lumina_options_; std::vector> array_vec_; + std::vector array_start_ids_; }; class LuminaIndexReader : public GlobalIndexReader { diff --git a/src/paimon/global_index/lumina/lumina_global_index_test.cpp b/src/paimon/global_index/lumina/lumina_global_index_test.cpp index d0bda5fc..0f314776 100644 --- a/src/paimon/global_index/lumina/lumina_global_index_test.cpp +++ b/src/paimon/global_index/lumina/lumina_global_index_test.cpp @@ -160,7 +160,7 @@ class LuminaGlobalIndexTest : public ::testing::Test { return struct_array; } - private: + protected: std::shared_ptr pool_ = GetDefaultPool(); std::shared_ptr fs_ = std::make_shared(); std::map options_ = {{"lumina.index.dimension", "4"}, @@ -472,4 +472,201 @@ TEST_F(LuminaGlobalIndexTest, TestHighCardinalityAndMultiThreadSearch) { } } +TEST_F(LuminaGlobalIndexTest, TestWriteWithNullRows) { + auto test_root_dir = paimon::test::UniqueTestDirectory::Create(); + ASSERT_TRUE(test_root_dir); + std::string test_root = test_root_dir->Str(); + + // Array with null at row 1 (middle): rows 0,2,3 are valid, row 1 is null + // This should split into two segments: [0,0] and [2,3] + std::shared_ptr array_with_null = + arrow::ipc::internal::json::ArrayFromJSON(data_type_, + R"([ + [[0.0, 0.0, 0.0, 0.0]], + [null], + [[1.0, 0.0, 1.0, 0.0]], + [[1.0, 1.0, 1.0, 1.0]] + ])") + .ValueOrDie(); + + ASSERT_OK_AND_ASSIGN( + auto meta, WriteGlobalIndex(test_root, data_type_, options_, array_with_null, Range(0, 3))); + ASSERT_OK_AND_ASSIGN(auto reader, + CreateGlobalIndexReader(test_root, data_type_, options_, meta)); + { + // Search should return ids 0, 2, 3 (skipping null row 1) + ASSERT_OK_AND_ASSIGN( + auto scored_result, + reader->VisitVectorSearch(std::make_shared( + /*field_name=*/"f0", /*limit=*/4, query_, /*filter=*/nullptr, + /*predicate=*/nullptr, /*distance_type=*/std::nullopt, /*options=*/options_))); + // Only 3 vectors indexed (row 1 is null), so limit=4 returns 3 + CheckResult(scored_result, {3l, 2l, 0l}, {0.01f, 2.21f, 4.21f}); + } +} + +TEST_F(LuminaGlobalIndexTest, TestWriteWithMultipleNullSegments) { + auto test_root_dir = paimon::test::UniqueTestDirectory::Create(); + ASSERT_TRUE(test_root_dir); + std::string test_root = test_root_dir->Str(); + + // Nulls at rows 0, 2, 5: valid rows are 1, 3, 4 + // Splits into segments: [1,1], [3,4] + std::shared_ptr array_with_nulls = + arrow::ipc::internal::json::ArrayFromJSON(data_type_, + R"([ + [null], + [[0.0, 1.0, 0.0, 1.0]], + [null], + [[1.0, 0.0, 1.0, 0.0]], + [[1.0, 1.0, 1.0, 1.0]], + [null] + ])") + .ValueOrDie(); + + ASSERT_OK_AND_ASSIGN(auto meta, WriteGlobalIndex(test_root, data_type_, options_, + array_with_nulls, Range(0, 5))); + ASSERT_OK_AND_ASSIGN(auto reader, + CreateGlobalIndexReader(test_root, data_type_, options_, meta)); + { + ASSERT_OK_AND_ASSIGN( + auto scored_result, + reader->VisitVectorSearch(std::make_shared( + /*field_name=*/"f0", /*limit=*/4, query_, /*filter=*/nullptr, + /*predicate=*/nullptr, /*distance_type=*/std::nullopt, /*options=*/options_))); + // Only 3 vectors indexed at ids 1, 3, 4 + CheckResult(scored_result, {4l, 1l, 3l}, {0.01f, 2.01f, 2.21f}); + } +} + +TEST_F(LuminaGlobalIndexTest, TestWriteWithAllNullRows) { + auto test_root_dir = paimon::test::UniqueTestDirectory::Create(); + ASSERT_TRUE(test_root_dir); + std::string test_root = test_root_dir->Str(); + + // All rows are null — no vectors to index + std::shared_ptr all_null_array = + arrow::ipc::internal::json::ArrayFromJSON(data_type_, + R"([ + [null], + [null], + [null] + ])") + .ValueOrDie(); + + auto global_index = std::make_shared(options_); + auto path_factory = std::make_shared(test_root); + auto file_writer = std::make_shared(fs_, path_factory); + + ASSERT_OK_AND_ASSIGN( + std::shared_ptr global_writer, + global_index->CreateWriter("f0", CreateArrowSchema(data_type_).get(), file_writer, pool_)); + + ArrowArray c_array; + ASSERT_TRUE(arrow::ExportArray(*all_null_array, &c_array).ok()); + std::vector row_ids = {0, 1, 2}; + ASSERT_OK(global_writer->AddBatch(&c_array, std::move(row_ids))); + // Finish with zero indexed vectors — returns empty metas + ASSERT_OK_AND_ASSIGN(auto result_metas, global_writer->Finish()); + ASSERT_TRUE(result_metas.empty()); +} + +TEST_F(LuminaGlobalIndexTest, TestWriteWithNullAndFilter) { + auto test_root_dir = paimon::test::UniqueTestDirectory::Create(); + ASSERT_TRUE(test_root_dir); + std::string test_root = test_root_dir->Str(); + + // Null at row 2: valid rows are 0, 1, 3 + std::shared_ptr array_with_null = + arrow::ipc::internal::json::ArrayFromJSON(data_type_, + R"([ + [[0.0, 0.0, 0.0, 0.0]], + [[0.0, 1.0, 0.0, 1.0]], + [null], + [[1.0, 1.0, 1.0, 1.0]] + ])") + .ValueOrDie(); + + ASSERT_OK_AND_ASSIGN( + auto meta, WriteGlobalIndex(test_root, data_type_, options_, array_with_null, Range(0, 3))); + ASSERT_OK_AND_ASSIGN(auto reader, + CreateGlobalIndexReader(test_root, data_type_, options_, meta)); + { + // Filter: only allow ids < 3 (filters out id=3), so only ids 0, 1 remain + auto filter = [](int64_t id) -> bool { return id < 3; }; + ASSERT_OK_AND_ASSIGN( + auto scored_result, + reader->VisitVectorSearch(std::make_shared( + /*field_name=*/"f0", /*limit=*/4, query_, filter, + /*predicate=*/nullptr, /*distance_type=*/std::nullopt, /*options=*/options_))); + CheckResult(scored_result, {1l, 0l}, {2.01f, 4.21f}); + } +} + +TEST_F(LuminaGlobalIndexTest, TestWriteWithNullAcrossMultipleBatches) { + auto test_root_dir = paimon::test::UniqueTestDirectory::Create(); + ASSERT_TRUE(test_root_dir); + std::string test_root = test_root_dir->Str(); + + // Batch 1: rows 0-2, null at row 1 → indexed ids: {0, 2} + std::shared_ptr batch1 = arrow::ipc::internal::json::ArrayFromJSON(data_type_, + R"([ + [[0.0, 0.0, 0.0, 0.0]], + [null], + [[1.0, 0.0, 1.0, 0.0]] + ])") + .ValueOrDie(); + + // Batch 2: rows 3-5, null at row 3 → indexed ids: {4, 5} + std::shared_ptr batch2 = arrow::ipc::internal::json::ArrayFromJSON(data_type_, + R"([ + [null], + [[1.0, 1.0, 1.0, 1.0]], + [[0.0, 1.0, 0.0, 1.0]] + ])") + .ValueOrDie(); + + auto global_index = std::make_shared(options_); + auto path_factory = std::make_shared(test_root); + auto file_writer = std::make_shared(fs_, path_factory); + + ASSERT_OK_AND_ASSIGN( + std::shared_ptr global_writer, + global_index->CreateWriter("f0", CreateArrowSchema(data_type_).get(), file_writer, pool_)); + + // AddBatch 1: row_ids {0, 1, 2} + { + ArrowArray c_array; + ASSERT_TRUE(arrow::ExportArray(*batch1, &c_array).ok()); + std::vector row_ids = {0, 1, 2}; + ASSERT_OK(global_writer->AddBatch(&c_array, std::move(row_ids))); + } + // AddBatch 2: row_ids {3, 4, 5} + { + ArrowArray c_array; + ASSERT_TRUE(arrow::ExportArray(*batch2, &c_array).ok()); + std::vector row_ids = {3, 4, 5}; + ASSERT_OK(global_writer->AddBatch(&c_array, std::move(row_ids))); + } + + ASSERT_OK_AND_ASSIGN(auto result_metas, global_writer->Finish()); + ASSERT_EQ(result_metas.size(), 1); + + ASSERT_OK_AND_ASSIGN(auto reader, + CreateGlobalIndexReader(test_root, data_type_, options_, result_metas[0])); + { + // Search all: should return ids {0, 2, 4, 5}, never {1, 3} + ASSERT_OK_AND_ASSIGN( + auto scored_result, + reader->VisitVectorSearch(std::make_shared( + /*field_name=*/"f0", /*limit=*/10, query_, /*filter=*/nullptr, + /*predicate=*/nullptr, /*distance_type=*/std::nullopt, /*options=*/options_))); + // id 0: [0,0,0,0] → L2 dist to [1,1,1,1.1] = 4.21 + // id 2: [1,0,1,0] → L2 dist = 2.21 + // id 4: [1,1,1,1] → L2 dist = 0.01 + // id 5: [0,1,0,1] → L2 dist = 2.01 + CheckResult(scored_result, {4l, 5l, 2l, 0l}, {0.01f, 2.01f, 2.21f, 4.21f}); + } +} + } // namespace paimon::lumina::test diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp index dce39d83..edcc38c4 100644 --- a/test/inte/global_index_test.cpp +++ b/test/inte/global_index_test.cpp @@ -991,8 +991,10 @@ TEST_P(GlobalIndexTest, TestWriteCommitScanReadIndexWithScore) { ["Bob", [0.0, 1.0, 0.0, 1.0], 10, 12.1], ["Emily", [1.0, 0.0, 1.0, 0.0], 10, 13.1], ["Tony", [1.0, 1.0, 1.0, 1.0], 10, 14.1], +["NullGuy1", null, 10, 20.0], ["Lucy", [10.0, 10.0, 10.0, 10.0], 20, 15.1], ["Bob", [10.0, 11.0, 10.0, 11.0], 20, 16.1], +["NullGuy2", null, 20, 21.0], ["Tony", [11.0, 10.0, 11.0, 10.0], 20, 17.1], ["Alice", [11.0, 11.0, 11.0, 11.0], 20, 18.1], ["Paul", [10.0, 10.0, 10.0, 10.0], 20, 19.1] @@ -1004,7 +1006,7 @@ TEST_P(GlobalIndexTest, TestWriteCommitScanReadIndexWithScore) { // write and commit lumina index ASSERT_OK(WriteIndex(table_path, /*partition_filters=*/{}, "f1", "lumina", - /*options=*/lumina_options, Range(0, 8))); + /*options=*/lumina_options, Range(0, 10))); auto scan_and_check_result = [&](const std::vector& read_row_ranges, const std::shared_ptr& expected_array, @@ -1022,8 +1024,8 @@ TEST_P(GlobalIndexTest, TestWriteCommitScanReadIndexWithScore) { result_fields.insert(result_fields.begin(), SpecialFields::ValueKind().ArrowField()); result_fields.push_back(SpecialFields::IndexScore().ArrowField()); std::map id_to_score = {{0, 4.21f}, {1, 2.01f}, {2, 2.21f}, - {3, 0.01f}, {4, 322.21f}, {5, 360.01f}, - {6, 360.21f}, {7, 398.01}, {8, 322.21f}}; + {3, 0.01f}, {5, 322.21f}, {6, 360.01f}, + {8, 360.21f}, {9, 398.01f}, {10, 322.21f}}; { auto expected_array = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(result_fields), R"([ @@ -1038,7 +1040,8 @@ TEST_P(GlobalIndexTest, TestWriteCommitScanReadIndexWithScore) { [0, "Paul", [10.0, 10.0, 10.0, 10.0], 20, 19.1, 322.21] ])") .ValueOrDie(); - scan_and_check_result({Range(0, 8)}, expected_array, id_to_score); + scan_and_check_result({Range(0, 3), Range(5, 6), Range(8, 10)}, expected_array, + id_to_score); } { auto expected_array = @@ -1049,7 +1052,7 @@ TEST_P(GlobalIndexTest, TestWriteCommitScanReadIndexWithScore) { [0, "Paul", [10.0, 10.0, 10.0, 10.0], 20, 19.1, 322.21] ])") .ValueOrDie(); - scan_and_check_result({Range(2, 3), Range(7, 8)}, expected_array, id_to_score); + scan_and_check_result({Range(2, 3), Range(9, 10)}, expected_array, id_to_score); } { auto expected_array = @@ -1057,7 +1060,7 @@ TEST_P(GlobalIndexTest, TestWriteCommitScanReadIndexWithScore) { [0, "Bob", [10.0, 11.0, 10.0, 11.0], 20, 16.1, 360.01] ])") .ValueOrDie(); - scan_and_check_result({Range(5, 5)}, expected_array, id_to_score); + scan_and_check_result({Range(6, 6)}, expected_array, id_to_score); } { auto expected_array = @@ -1068,7 +1071,31 @@ TEST_P(GlobalIndexTest, TestWriteCommitScanReadIndexWithScore) { [0, "Paul", [10.0, 10.0, 10.0, 10.0], 20, 19.1, null] ])") .ValueOrDie(); - scan_and_check_result({Range(2, 3), Range(7, 8)}, expected_array, /*id_to_score=*/{}); + scan_and_check_result({Range(2, 3), Range(9, 10)}, expected_array, /*id_to_score=*/{}); + } + { + // Verify null rows (id 4 and 7) are never recalled by vector search + ASSERT_OK_AND_ASSIGN( + std::shared_ptr global_index_scan, + GlobalIndexScan::Create(table_path, /*snapshot_id=*/std::nullopt, + /*partitions=*/std::nullopt, lumina_options, fs_, + /*executor=*/nullptr, pool_)); + ASSERT_OK_AND_ASSIGN(auto lumina_readers, + global_index_scan->CreateReaders("f1", std::nullopt)); + ASSERT_EQ(lumina_readers.size(), 1u); + std::vector query = {1.0f, 1.0f, 1.0f, 1.1f}; + auto vector_search = std::make_shared( + "f1", /*limit=*/20, query, /*filter=*/nullptr, + /*predicate=*/nullptr, /*distance_type=*/std::nullopt, /*options=*/lumina_options); + ASSERT_OK_AND_ASSIGN(auto scored_result, + lumina_readers[0]->VisitVectorSearch(vector_search)); + auto typed_result = std::dynamic_pointer_cast(scored_result); + ASSERT_TRUE(typed_result); + // Should recall 9 vectors (11 total rows - 2 null rows) + ASSERT_EQ(typed_result->bitmap_.Cardinality(), 9u); + // Null row ids 4 and 7 must not be in the result + ASSERT_FALSE(typed_result->bitmap_.Contains(4)); + ASSERT_FALSE(typed_result->bitmap_.Contains(7)); } } #endif From 0f306d1ca996d03f24b2e7d5c5314cd0d057d6b7 Mon Sep 17 00:00:00 2001 From: lszskye <57179283+lszskye@users.noreply.github.com> Date: Tue, 30 Jun 2026 02:22:09 -0700 Subject: [PATCH 04/11] feat: add CreateReader api with field_name & index_type (#391) --- .../paimon/global_index/global_index_scan.h | 16 +++++++++ .../global_index/global_index_scan_impl.cpp | 33 +++++++++++++++---- .../global_index/global_index_scan_impl.h | 8 ++++- test/inte/global_index_test.cpp | 27 +++++++++++++++ 4 files changed, 77 insertions(+), 7 deletions(-) diff --git a/include/paimon/global_index/global_index_scan.h b/include/paimon/global_index/global_index_scan.h index d925e2f0..d8ba799d 100644 --- a/include/paimon/global_index/global_index_scan.h +++ b/include/paimon/global_index/global_index_scan.h @@ -100,6 +100,22 @@ class PAIMON_EXPORT GlobalIndexScan { const std::string& field_name, const std::optional& 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> CreateReader( + const std::string& field_name, const std::string& index_type, + const std::optional& 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. diff --git a/src/paimon/core/global_index/global_index_scan_impl.cpp b/src/paimon/core/global_index/global_index_scan_impl.cpp index 40820a2c..83f12395 100644 --- a/src/paimon/core/global_index/global_index_scan_impl.cpp +++ b/src/paimon/core/global_index/global_index_scan_impl.cpp @@ -131,27 +131,48 @@ Result> GlobalIndexScanImpl::GetOrCreateIn Result>> GlobalIndexScanImpl::CreateReaders( int32_t field_id, const std::optional& 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>> GlobalIndexScanImpl::CreateReaders( const std::string& field_name, const std::optional& 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> GlobalIndexScanImpl::CreateReader( + const std::string& field_name, const std::string& index_type, + const std::optional& row_range_index) const { + PAIMON_ASSIGN_OR_RAISE(DataField field, table_schema_->GetField(field_name)); + PAIMON_ASSIGN_OR_RAISE( + std::vector> readers, + CreateReaders(field, std::optional(index_type), row_range_index)); + if (readers.empty()) { + return std::shared_ptr(); + } + if (readers.size() != 1) { + return Status::Invalid( + fmt::format("invalid global index reader size, expected 1, actual {}", readers.size())); + } + return readers[0]; } Result>> GlobalIndexScanImpl::CreateReaders( - const DataField& field, const std::optional& row_range_index) const { + const DataField& field, const std::optional& index_type, + const std::optional& row_range_index) const { auto field_iter = index_metas_.find(field.Id()); if (field_iter == index_metas_.end()) { return std::vector>(); } const auto& index_type_to_metas = field_iter->second; std::vector> 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 indexer, - GlobalIndexerFactory::Get(index_type, options_.ToMap())); + GlobalIndexerFactory::Get(current_index_type, options_.ToMap())); if (!indexer) { continue; } diff --git a/src/paimon/core/global_index/global_index_scan_impl.h b/src/paimon/core/global_index/global_index_scan_impl.h index d5f10d06..a7921d77 100644 --- a/src/paimon/core/global_index/global_index_scan_impl.h +++ b/src/paimon/core/global_index/global_index_scan_impl.h @@ -21,6 +21,7 @@ #include #include +#include #include #include @@ -50,6 +51,10 @@ class GlobalIndexScanImpl : public GlobalIndexScan { const std::string& field_name, const std::optional& row_range_index) const override; + Result> CreateReader( + const std::string& field_name, const std::string& index_type, + const std::optional& row_range_index) const override; + Result>> CreateReaders( int32_t field_id, const std::optional& row_range_index) const override; @@ -68,7 +73,8 @@ class GlobalIndexScanImpl : public GlobalIndexScan { Result> GetOrCreateIndexEvaluator(); Result>> CreateReaders( - const DataField& field, const std::optional& row_range_index) const; + const DataField& field, const std::optional& index_type, + const std::optional& row_range_index) const; std::vector ToGlobalIndexIOMetas( const std::vector>& metas) const; diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp index edcc38c4..f87d4331 100644 --- a/test/inte/global_index_test.cpp +++ b/test/inte/global_index_test.cpp @@ -2660,6 +2660,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 btree_reader, + global_index_scan->CreateReader("f0", "btree", std::nullopt)); + ASSERT_TRUE(btree_reader); + ASSERT_OK_AND_ASSIGN(std::shared_ptr 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 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 bitmap_reader, + global_index_scan->CreateReader("f0", "bitmap", std::nullopt)); + ASSERT_TRUE(bitmap_reader); + ASSERT_OK_AND_ASSIGN(std::shared_ptr 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 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 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, From 5507c474ae83df0ead452327b0ee80317639a060 Mon Sep 17 00:00:00 2001 From: lxy <38709059+lxy-9602@users.noreply.github.com> Date: Thu, 2 Jul 2026 20:17:38 +0800 Subject: [PATCH 05/11] test: add tests for executor (#396) --- .../common/executor/default_executor_test.cpp | 57 +++++++++++++++++++ 1 file changed, 57 insertions(+) diff --git a/src/paimon/common/executor/default_executor_test.cpp b/src/paimon/common/executor/default_executor_test.cpp index 26d84f85..2745a69b 100644 --- a/src/paimon/common/executor/default_executor_test.cpp +++ b/src/paimon/common/executor/default_executor_test.cpp @@ -124,4 +124,61 @@ TEST(DefaultExecutorTest, TestAddTaskAfterShutdownNowIgnored) { ASSERT_EQ(executed_count.load(), 0); } +TEST(DefaultExecutorTest, TestAddTaskFromMultipleThreads) { + auto executor = CreateDefaultExecutor(/*thread_count=*/4); + + constexpr int32_t kSubmitterCount = 8; + constexpr int32_t kTaskCountPerSubmitter = 64; + constexpr int32_t kTotalTaskCount = kSubmitterCount * kTaskCountPerSubmitter; + + std::vector> executed_slots(kTotalTaskCount); + for (auto& executed_slot : executed_slots) { + executed_slot.store(0); + } + std::vector> task_promises(kTotalTaskCount); + std::vector> task_futures; + task_futures.reserve(kTotalTaskCount); + for (auto& task_promise : task_promises) { + task_futures.push_back(task_promise.get_future()); + } + std::atomic ready_submitter_count = 0; + std::atomic executed_count = 0; + std::promise start_signal; + std::shared_future start_future = start_signal.get_future().share(); + std::vector submitters; + submitters.reserve(kSubmitterCount); + + for (int32_t submitter_index = 0; submitter_index < kSubmitterCount; ++submitter_index) { + submitters.emplace_back([&, submitter_index]() { + ++ready_submitter_count; + start_future.wait(); + for (int32_t task_index = 0; task_index < kTaskCountPerSubmitter; ++task_index) { + const int32_t slot_index = submitter_index * kTaskCountPerSubmitter + task_index; + executor->Add([&, slot_index]() { + ++executed_slots[slot_index]; + ++executed_count; + task_promises[slot_index].set_value(); + }); + } + }); + } + + for (int32_t retry = 0; retry < 100 && ready_submitter_count.load() < kSubmitterCount; + ++retry) { + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + } + const int32_t ready_submitter_count_before_start = ready_submitter_count.load(); + start_signal.set_value(); + for (auto& submitter : submitters) { + submitter.join(); + } + ASSERT_EQ(kSubmitterCount, ready_submitter_count_before_start); + Wait(task_futures); + + ASSERT_EQ(kTotalTaskCount, executed_count.load()); + for (const auto& executed_slot : executed_slots) { + ASSERT_EQ(1, executed_slot.load()); + } +} + } // namespace paimon::test From 7b1c72775ed2af103a5ec69a1692e37538cd5b17 Mon Sep 17 00:00:00 2001 From: lszskye <57179283+lszskye@users.noreply.github.com> Date: Thu, 2 Jul 2026 18:57:45 -0700 Subject: [PATCH 06/11] fix: fix IsThreadSafe() in UnionGlobalIndexReader (#395) --- .../union_global_index_reader.cpp | 11 +++++++- .../global_index/union_global_index_reader.h | 4 +-- .../union_global_index_reader_test.cpp | 27 +++++++++++++++---- 3 files changed, 33 insertions(+), 9 deletions(-) diff --git a/src/paimon/common/global_index/union_global_index_reader.cpp b/src/paimon/common/global_index/union_global_index_reader.cpp index 7f4e5cc0..a4de3b21 100644 --- a/src/paimon/common/global_index/union_global_index_reader.cpp +++ b/src/paimon/common/global_index/union_global_index_reader.cpp @@ -185,6 +185,15 @@ Result> UnionGlobalIndexReader::Union(ReaderA return merged_result; } +bool UnionGlobalIndexReader::IsThreadSafe() const { + for (const auto& reader : readers_) { + if (!reader->IsThreadSafe()) { + return false; + } + } + return true; +} + template std::vector UnionGlobalIndexReader::ExecuteAllReaders( const std::function&)>& action) { @@ -202,7 +211,7 @@ std::vector UnionGlobalIndexReader::ExecuteAllReaders( futures.reserve(readers_.size()); for (const auto& reader : readers_) { futures.push_back( - Via(executor_.get(), [&action, &reader]() -> R { return action(reader); })); + Via(executor_.get(), [&action, reader]() -> R { return action(reader); })); } return CollectAll(futures); } diff --git a/src/paimon/common/global_index/union_global_index_reader.h b/src/paimon/common/global_index/union_global_index_reader.h index 526324ed..efd42df5 100644 --- a/src/paimon/common/global_index/union_global_index_reader.h +++ b/src/paimon/common/global_index/union_global_index_reader.h @@ -61,9 +61,7 @@ class UnionGlobalIndexReader : public GlobalIndexReader { Result> VisitFullTextSearch( const std::shared_ptr& full_text_search) override; - bool IsThreadSafe() const override { - return false; - } + bool IsThreadSafe() const override; std::string GetIndexType() const override { return "union"; diff --git a/src/paimon/common/global_index/union_global_index_reader_test.cpp b/src/paimon/common/global_index/union_global_index_reader_test.cpp index 78efd345..8d55db42 100644 --- a/src/paimon/common/global_index/union_global_index_reader_test.cpp +++ b/src/paimon/common/global_index/union_global_index_reader_test.cpp @@ -64,6 +64,10 @@ class FakeReader : public GlobalIndexReader { has_scored_result_ = true; } + void SetThreadSafe(bool thread_safe) { + thread_safe_ = thread_safe; + } + /// Counts how many times any Visit* method was invoked. Useful to assert all readers /// are exercised by UnionGlobalIndexReader. int InvocationCount() const { @@ -150,7 +154,7 @@ class FakeReader : public GlobalIndexReader { } bool IsThreadSafe() const override { - return true; + return thread_safe_; } std::string GetIndexType() const override { @@ -179,6 +183,7 @@ class FakeReader : public GlobalIndexReader { std::vector scored_row_ids_; std::vector scored_scores_; bool has_scored_result_ = false; + bool thread_safe_ = true; std::atomic invocation_count_{0}; }; @@ -518,12 +523,24 @@ TEST_F(UnionGlobalIndexReaderTest, TestVisitVectorSearchErrorPropagation) { ASSERT_NOK_WITH_MSG(union_reader.VisitVectorSearch(nullptr), "vector search failure"); } -TEST_F(UnionGlobalIndexReaderTest, TestIsThreadSafeAlwaysFalse) { - auto reader = std::make_shared(); - std::vector> readers = {reader}; +TEST_F(UnionGlobalIndexReaderTest, TestIsThreadSafeReturnsTrueWhenAllReadersAreSafe) { + auto reader1 = std::make_shared(); + auto reader2 = std::make_shared(); + + std::vector> readers = {reader1, reader2}; + UnionGlobalIndexReader union_reader(std::move(readers), nullptr); + + ASSERT_TRUE(union_reader.IsThreadSafe()); +} + +TEST_F(UnionGlobalIndexReaderTest, TestIsThreadSafeReturnsFalseWhenAnyReaderIsNotSafe) { + auto reader1 = std::make_shared(); + auto reader2 = std::make_shared(); + reader2->SetThreadSafe(false); + + std::vector> readers = {reader1, reader2}; UnionGlobalIndexReader union_reader(std::move(readers), nullptr); - // UnionGlobalIndexReader is not thread-safe regardless of inner readers ASSERT_FALSE(union_reader.IsThreadSafe()); } From 14d5512f3db282bd2d7cb05aec3f0693e743999c Mon Sep 17 00:00:00 2001 From: lxy <38709059+lxy-9602@users.noreply.github.com> Date: Fri, 3 Jul 2026 08:50:50 +0800 Subject: [PATCH 07/11] fix: release global index writer before reader to avoid mem issue when write index failed (#398) --- .../global_index/global_index_write_task.cpp | 14 +++++--- test/inte/global_index_test.cpp | 35 +++++++++++++++++++ 2 files changed, 45 insertions(+), 4 deletions(-) diff --git a/src/paimon/core/global_index/global_index_write_task.cpp b/src/paimon/core/global_index/global_index_write_task.cpp index 1d2932f2..a42f39a1 100644 --- a/src/paimon/core/global_index/global_index_write_task.cpp +++ b/src/paimon/core/global_index/global_index_write_task.cpp @@ -23,6 +23,7 @@ #include "paimon/common/table/special_fields.h" #include "paimon/common/types/data_field.h" #include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/scope_guard.h" #include "paimon/core/core_options.h" #include "paimon/core/global_index/global_index_file_manager.h" #include "paimon/core/io/data_increment.h" @@ -207,16 +208,21 @@ Result> GlobalIndexWriteTask::WriteIndex( std::shared_ptr index_file_manager, CreateGlobalIndexFileManager(table_path, table_schema, core_options, pool)); + // create batch reader + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr batch_reader, + CreateBatchReader(table_path, field_name, indexed_split, core_options, pool)); + // create global index writer PAIMON_ASSIGN_OR_RAISE(DataField field, table_schema->GetField(field_name)); PAIMON_ASSIGN_OR_RAISE( std::shared_ptr global_index_writer, CreateGlobalIndexWriter(index_type, field, index_file_manager, core_options, pool)); - // create batch reader - PAIMON_ASSIGN_OR_RAISE( - std::unique_ptr batch_reader, - CreateBatchReader(table_path, field_name, indexed_split, core_options, pool)); + ScopeGuard guard([&]() { + global_index_writer.reset(); + batch_reader.reset(); + }); // read from data split and write to index writer PAIMON_ASSIGN_OR_RAISE( diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp index f87d4331..6528c243 100644 --- a/test/inte/global_index_test.cpp +++ b/test/inte/global_index_test.cpp @@ -294,6 +294,41 @@ TEST_P(GlobalIndexTest, TestWriteLuminaIndex) { ASSERT_TRUE(expected_commit_message->TEST_Equal(*index_commit_msg_impl)); } +TEST_P(GlobalIndexTest, TestWriteLuminaIndexWithMismatchedDimension) { + arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()), + arrow::field("f1", arrow::list(arrow::float32()))}; + auto schema = arrow::schema(fields); + std::map lumina_options = {{"lumina.index.dimension", "3"}, + {"lumina.index.type", "bruteforce"}, + {"lumina.distance.metric", "l2"}, + {"lumina.encoding.type", "rawf32"}, + {"lumina.search.parallel_number", "10"}}; + + std::map options = { + {Options::MANIFEST_FORMAT, "orc"}, {Options::FILE_FORMAT, file_format_}, + {Options::FILE_SYSTEM, "local"}, {Options::ROW_TRACKING_ENABLED, "true"}, + {Options::DATA_EVOLUTION_ENABLED, "true"}, {Options::READ_BATCH_SIZE, "1"}}; + + CreateTable(/*partition_keys=*/{}, schema, options); + std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar"); + + std::vector write_cols = schema->field_names(); + auto src_array = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([ + ["a", [0.0, 0.0, 0.0]], + ["b", [0.0, 0.0, 0.0, 0.0]] + ])") + .ValueOrDie(); + + ASSERT_OK_AND_ASSIGN(auto commit_msgs, WriteArray(table_path, write_cols, src_array)); + ASSERT_OK(Commit(table_path, commit_msgs)); + + ASSERT_NOK_WITH_MSG( + WriteIndex(table_path, /*partition_filters=*/{}, "f1", "lumina", + /*options=*/lumina_options, Range(0, 1)), + "invalid input array in LuminaIndexWriter, length of field array [1] multiplied " + "dimension [3] must match length of field value array [4]"); +} + TEST_P(GlobalIndexTest, TestWriteIndex) { CreateTable(); std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar"); From 6a87938edea28155e7cb99fa37263c14e2b04ff4 Mon Sep 17 00:00:00 2001 From: lxy <38709059+lxy-9602@users.noreply.github.com> Date: Wed, 22 Jul 2026 22:32:18 +0800 Subject: [PATCH 08/11] fix: complete undefined shift fixes from #308 Migrate the two remaining hunks from alibaba/paimon-cpp@620adbe3590d4fea2c9835eebc5bb66fe41eb633; the other hunks are already present in Apache. --- src/paimon/common/data/binary_array_writer.cpp | 3 ++- src/paimon/common/data/binary_row_writer.h | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/src/paimon/common/data/binary_array_writer.cpp b/src/paimon/common/data/binary_array_writer.cpp index 7845311c..2a0961f4 100644 --- a/src/paimon/common/data/binary_array_writer.cpp +++ b/src/paimon/common/data/binary_array_writer.cpp @@ -98,7 +98,8 @@ int32_t BinaryArrayWriter::GetFieldOffset(int32_t pos) const { } void BinaryArrayWriter::SetOffsetAndSize(int32_t pos, int32_t offset, int64_t size) { - const int64_t offset_and_size = (static_cast(offset) << 32) | size; + const auto offset_and_size = + static_cast((static_cast(offset) << 32) | static_cast(size)); segment_.PutValue(GetElementOffset(pos, 8), offset_and_size); } diff --git a/src/paimon/common/data/binary_row_writer.h b/src/paimon/common/data/binary_row_writer.h index 06a58165..954ba1d6 100644 --- a/src/paimon/common/data/binary_row_writer.h +++ b/src/paimon/common/data/binary_row_writer.h @@ -103,7 +103,8 @@ class BinaryRowWriter : public AbstractBinaryWriter { } void SetOffsetAndSize(int32_t pos, int32_t offset, int64_t size) override { - const int64_t offset_and_size = (static_cast(offset) << 32) | size; + const auto offset_and_size = static_cast((static_cast(offset) << 32) | + static_cast(size)); segment_.PutValue(GetFieldOffset(pos), offset_and_size); } From a6196af122317e8426583e94a100e8c06b771c72 Mon Sep 17 00:00:00 2001 From: "jinli.zjw" Date: Wed, 22 Jul 2026 22:36:11 +0800 Subject: [PATCH 09/11] chore: remove remaining Lance references --- CMakeLists.txt | 27 +------- src/paimon/common/utils/file_type_test.cpp | 2 - .../core/operation/abstract_split_read.cpp | 4 -- .../operation/data_evolution_split_read.cpp | 2 +- .../operation/orphan_files_cleaner_impl.cpp | 2 +- .../operation/orphan_files_cleaner_test.cpp | 2 - .../core/operation/raw_file_split_read.cpp | 2 +- test/inte/append_compaction_inte_test.cpp | 3 - test/inte/blob_table_inte_test.cpp | 27 +++----- test/inte/data_evolution_table_test.cpp | 42 ++++++------- test/inte/global_index_test.cpp | 21 +++---- test/inte/pk_compaction_inte_test.cpp | 5 +- test/inte/scan_and_read_inte_test.cpp | 61 ------------------- test/inte/write_and_read_inte_test.cpp | 5 +- test/inte/write_inte_test.cpp | 21 +------ 15 files changed, 43 insertions(+), 183 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index 7488e8aa..8a435547 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -67,18 +67,12 @@ endif() if(PAIMON_USE_CXX11_ABI) add_definitions(-D_GLIBCXX_USE_CXX11_ABI=1) else() - # lance and lumina are provided in so file, cannot be recompiled with C++11 ABI off. - if(PAIMON_ENABLE_LANCE) - message(FATAL_ERROR "Lance cannot be enabled with C++11 ABI off") - endif() + # Lumina is provided as a shared library and cannot be recompiled with C++11 ABI off. if(PAIMON_ENABLE_LUMINA) message(FATAL_ERROR "Lumina cannot be enabled with C++11 ABI off") endif() add_definitions(-D_GLIBCXX_USE_CXX11_ABI=0) endif() -if(PAIMON_ENABLE_LANCE) - add_definitions(-DPAIMON_ENABLE_LANCE) -endif() if(PAIMON_ENABLE_LUMINA) add_definitions(-DPAIMON_ENABLE_LUMINA) endif() @@ -289,13 +283,6 @@ set(PAIMON_SHARED_PRIVATE_LINK_LIBS ${PAIMON_STATIC_LINK_LIBS}) add_subdirectory(third_party/roaring_bitmap EXCLUDE_FROM_ALL) add_subdirectory(third_party/xxhash EXCLUDE_FROM_ALL) -if(PAIMON_ENABLE_LANCE) - add_subdirectory(third_party/lance EXCLUDE_FROM_ALL) - link_directories(third_party/lance/lance_lib/target/release) - install(FILES ${PROJECT_SOURCE_DIR}/third_party/lance/lance_lib/target/release/liblance_lib_rc.so - DESTINATION ${CMAKE_INSTALL_LIBDIR}) -endif() - list(APPEND PAIMON_LINK_LIBS ${CMAKE_DL_LIBS}) list(APPEND PAIMON_SHARED_INSTALL_INTERFACE_LIBS ${CMAKE_DL_LIBS}) @@ -327,10 +314,6 @@ include_directories(${CMAKE_CURRENT_SOURCE_DIR}/include) include_directories("${CMAKE_SOURCE_DIR}/third_party/roaring_bitmap") include_directories("${CMAKE_SOURCE_DIR}/third_party/xxhash") -if(PAIMON_ENABLE_LANCE) - include_directories("${CMAKE_SOURCE_DIR}/third_party/lance") -endif() - include_directories(SYSTEM ${ARROW_INCLUDE_DIR}) include_directories(SYSTEM ${TBB_INCLUDE_DIR}) @@ -380,11 +363,6 @@ if(PAIMON_BUILD_TESTS) paimon_blob_file_format_shared "-Wl,--as-needed") - if(PAIMON_ENABLE_LANCE) - list(APPEND TEST_STATIC_LINK_LIBS "-Wl,--no-as-needed") - list(APPEND TEST_STATIC_LINK_LIBS paimon_lance_file_format_shared) - list(APPEND TEST_STATIC_LINK_LIBS "-Wl,--as-needed") - endif() if(PAIMON_ENABLE_ORC) list(APPEND TEST_STATIC_LINK_LIBS "-Wl,--no-as-needed") list(APPEND TEST_STATIC_LINK_LIBS paimon_orc_file_format_shared) @@ -439,9 +417,6 @@ 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_LANCE) - add_subdirectory(src/paimon/format/lance) -endif() if(PAIMON_ENABLE_LUMINA) add_subdirectory(src/paimon/global_index/lumina) endif() diff --git a/src/paimon/common/utils/file_type_test.cpp b/src/paimon/common/utils/file_type_test.cpp index a358b7a1..b9698130 100644 --- a/src/paimon/common/utils/file_type_test.cpp +++ b/src/paimon/common/utils/file_type_test.cpp @@ -80,8 +80,6 @@ TEST(FileTypeTest, TestDefaultData) { FileType::kData); ASSERT_EQ(FileTypeUtils::Classify("dfs://cluster/db/p=1/bucket-0/data-a1b2c3d4-0.blob"), FileType::kData); - ASSERT_EQ(FileTypeUtils::Classify("dfs://cluster/db/p=1/bucket-0/data-a1b2c3d4-0.vector.lance"), - FileType::kData); ASSERT_EQ(FileTypeUtils::Classify("dfs://cluster/db/random/unknown.bin"), FileType::kData); } diff --git a/src/paimon/core/operation/abstract_split_read.cpp b/src/paimon/core/operation/abstract_split_read.cpp index fb57c0ea..30b2215a 100644 --- a/src/paimon/core/operation/abstract_split_read.cpp +++ b/src/paimon/core/operation/abstract_split_read.cpp @@ -148,10 +148,6 @@ Result> AbstractSplitRead::CreateFileBatchReade const std::shared_ptr& file_meta, const std::string& data_file_path, const ReaderBuilder* reader_builder) const { PAIMON_ASSIGN_OR_RAISE(std::string file_format_identifier, file_meta->FileFormat()); - if (file_format_identifier == "lance") { - // lance do not support stream build with input stream - return reader_builder->Build(data_file_path); - } if (context_->EnablePrefetch() && file_format_identifier != "blob" && file_format_identifier != "avro") { PAIMON_ASSIGN_OR_RAISE( diff --git a/src/paimon/core/operation/data_evolution_split_read.cpp b/src/paimon/core/operation/data_evolution_split_read.cpp index 5e017683..81c19c65 100644 --- a/src/paimon/core/operation/data_evolution_split_read.cpp +++ b/src/paimon/core/operation/data_evolution_split_read.cpp @@ -199,7 +199,7 @@ Result> DataEvolutionSplitRead::ApplyIndexAndDv std::unique_ptr reader; if (!file_reader->SupportPreciseBitmapSelection() && selection_row_ids) { - // several format(e.g. lance, blob) will return accurate batch result, where + // Some formats (for example blob) return an accurate batch result, where // ApplyBitmapIndexBatchReader is not necessary reader = std::make_unique( std::move(file_reader), std::move(selection_row_ids).value()); diff --git a/src/paimon/core/operation/orphan_files_cleaner_impl.cpp b/src/paimon/core/operation/orphan_files_cleaner_impl.cpp index 5790d308..e582ae45 100644 --- a/src/paimon/core/operation/orphan_files_cleaner_impl.cpp +++ b/src/paimon/core/operation/orphan_files_cleaner_impl.cpp @@ -78,7 +78,7 @@ bool OrphanFilesCleanerImpl::SupportToClean(const std::string& file_name) { return true; } } - static std::vector supported_formats = {".orc", ".parquet", ".avro", ".lance"}; + static std::vector supported_formats = {".orc", ".parquet", ".avro"}; for (const auto& format : supported_formats) { if (StringUtils::StartsWith(file_name, "data-") && StringUtils::EndsWith(file_name, format)) { diff --git a/src/paimon/core/operation/orphan_files_cleaner_test.cpp b/src/paimon/core/operation/orphan_files_cleaner_test.cpp index 4fb94f8b..37731049 100644 --- a/src/paimon/core/operation/orphan_files_cleaner_test.cpp +++ b/src/paimon/core/operation/orphan_files_cleaner_test.cpp @@ -41,8 +41,6 @@ TEST(OrphanFilesCleanerTest, TestSupportToClean) { OrphanFilesCleanerImpl::SupportToClean("data-2d5ea1ea-77c1-47ff-bb87-19a509962a37-0.avro")); ASSERT_TRUE(OrphanFilesCleanerImpl::SupportToClean( "data-2d5ea1ea-77c1-47ff-bb87-19a509962a37-0.parquet")); - ASSERT_TRUE(OrphanFilesCleanerImpl::SupportToClean( - "data-2d5ea1ea-77c1-47ff-bb87-19a509962a37-0.lance")); ASSERT_TRUE( OrphanFilesCleanerImpl::SupportToClean("manifest-3ea5ee21-d399-4f1c-a749-2fc63dbf0852-0")); ASSERT_TRUE(OrphanFilesCleanerImpl::SupportToClean( diff --git a/src/paimon/core/operation/raw_file_split_read.cpp b/src/paimon/core/operation/raw_file_split_read.cpp index c3512541..c06fb055 100644 --- a/src/paimon/core/operation/raw_file_split_read.cpp +++ b/src/paimon/core/operation/raw_file_split_read.cpp @@ -182,7 +182,7 @@ Result> RawFileSplitRead::ApplyIndexAndDvReader std::unique_ptr reader; if (!file_reader->SupportPreciseBitmapSelection() && actual_selection) { - // several format(e.g. lance, blob) will return accurate batch result, where + // Some formats (for example blob) return an accurate batch result, where // ApplyBitmapIndexBatchReader is not necessary reader = std::make_unique(std::move(file_reader), std::move(actual_selection).value()); diff --git a/test/inte/append_compaction_inte_test.cpp b/test/inte/append_compaction_inte_test.cpp index eda4d6cd..88d81151 100644 --- a/test/inte/append_compaction_inte_test.cpp +++ b/test/inte/append_compaction_inte_test.cpp @@ -149,9 +149,6 @@ std::vector GetTestValuesForAppendCompactionInteTest() { #ifdef PAIMON_ENABLE_ORC values.emplace_back("orc"); #endif -#ifdef PAIMON_ENABLE_LANCE - values.emplace_back("lance"); -#endif #ifdef PAIMON_ENABLE_AVRO values.emplace_back("avro"); #endif diff --git a/test/inte/blob_table_inte_test.cpp b/test/inte/blob_table_inte_test.cpp index eb3ba04b..ca6df4f1 100644 --- a/test/inte/blob_table_inte_test.cpp +++ b/test/inte/blob_table_inte_test.cpp @@ -250,9 +250,6 @@ std::vector GetTestValuesForBlobTableInteTest() { #ifdef PAIMON_ENABLE_ORC values.emplace_back("orc"); #endif -#ifdef PAIMON_ENABLE_LANCE - values.emplace_back("lance"); -#endif #ifdef PAIMON_ENABLE_AVRO values.emplace_back("avro"); #endif @@ -434,7 +431,7 @@ TEST_P(BlobTableInteTest, TestBasic) { .ValueOrDie()); ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array)); - if (GetParam() != "lance") { + { // read with row tracking auto expected_row_tracking_array = std::dynamic_pointer_cast( arrow::ipc::internal::json::ArrayFromJSON( @@ -539,7 +536,7 @@ TEST_P(BlobTableInteTest, TestMultipleAppends) { .ValueOrDie()); ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array)); - if (GetParam() != "lance") { + { // read with row tracking auto expected_row_tracking_array = std::dynamic_pointer_cast( arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({ @@ -659,7 +656,7 @@ TEST_P(BlobTableInteTest, TestMultipleAppendsDifferentFirstRowIds) { .ValueOrDie()); ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array)); - if (GetParam() != "lance") { + { // read with row tracking auto expected_row_tracking_array = std::dynamic_pointer_cast( arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({ @@ -814,7 +811,7 @@ TEST_P(BlobTableInteTest, TestExternalPath) { .ValueOrDie()); ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array)); - if (GetParam() != "lance") { + { // read with row tracking auto expected_row_tracking_array = std::dynamic_pointer_cast( arrow::ipc::internal::json::ArrayFromJSON( @@ -833,9 +830,6 @@ TEST_P(BlobTableInteTest, TestExternalPath) { TEST_P(BlobTableInteTest, TestPartitionWithPredicate) { auto file_format = GetParam(); - if (file_format == "lance") { - return; - } std::vector partition_keys = {"f0"}; std::map options = { {Options::MANIFEST_FORMAT, "orc"}, {Options::FILE_FORMAT, GetParam()}, @@ -953,8 +947,8 @@ TEST_P(BlobTableInteTest, TestPartitionWithPredicate) { } TEST_P(BlobTableInteTest, TestPredicate) { - if (GetParam() == "lance" || GetParam() == "avro") { - // lance and avro do not have stats + if (GetParam() == "avro") { + // Avro does not have stats. return; } CreateTable(); @@ -1023,9 +1017,6 @@ TEST_P(BlobTableInteTest, TestPredicate) { } TEST_P(BlobTableInteTest, TestIOException) { - if (GetParam() == "lance") { - return; - } std::string table_path; // write and commit with I/O exception bool write_run_complete = false; @@ -1098,7 +1089,7 @@ TEST_P(BlobTableInteTest, TestIOException) { TEST_P(BlobTableInteTest, TestReadTableWithDenseStats) { auto file_format = GetParam(); - if (file_format == "lance" || file_format == "avro") { + if (file_format == "avro") { return; } std::string table_path = @@ -1162,7 +1153,7 @@ TEST_P(BlobTableInteTest, TestReadTableWithDenseStats) { TEST_P(BlobTableInteTest, TestDataEvolutionAndAlterTable) { auto file_format = GetParam(); - if (file_format == "lance" || file_format == "avro") { + if (file_format == "avro") { return; } std::string table_path = paimon::test::GetDataDir() + file_format + @@ -1505,7 +1496,7 @@ TEST_P(BlobTableInteTest, TestAppendWriteWithNullBlob) { TEST_P(BlobTableInteTest, TestReadTableWithMultiBlobFields) { auto file_format = GetParam(); - if (file_format == "lance" || file_format == "avro") { + if (file_format == "avro") { return; } std::string table_path = paimon::test::GetDataDir() + file_format + diff --git a/test/inte/data_evolution_table_test.cpp b/test/inte/data_evolution_table_test.cpp index efcd7b09..4c040fcf 100644 --- a/test/inte/data_evolution_table_test.cpp +++ b/test/inte/data_evolution_table_test.cpp @@ -288,7 +288,7 @@ TEST_P(DataEvolutionTableTest, TestBasic) { .ValueOrDie()); ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array)); - if (GetParam() != "lance") { + { // read with row tracking auto expected_row_tracking_array = std::dynamic_pointer_cast( arrow::ipc::internal::json::ArrayFromJSON( @@ -417,7 +417,7 @@ TEST_P(DataEvolutionTableTest, TestMultipleAppends) { /*predicate=*/nullptr, /*row_ranges=*/row_ranges)); } - if (GetParam() != "lance") { + { // read with row tracking auto expected_row_tracking_array = std::dynamic_pointer_cast( arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({ @@ -492,7 +492,7 @@ TEST_P(DataEvolutionTableTest, TestOnlySomeColumns) { .ValueOrDie()); ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array)); - if (GetParam() != "lance") { + { // read with row tracking auto expected_row_tracking_array = std::dynamic_pointer_cast( arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({ @@ -561,7 +561,7 @@ TEST_P(DataEvolutionTableTest, TestNullValues) { .ValueOrDie()); ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array)); - if (GetParam() != "lance") { + { // read with row tracking auto expected_row_tracking_array = std::dynamic_pointer_cast( arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({ @@ -644,7 +644,7 @@ TEST_P(DataEvolutionTableTest, TestMultipleAppendsDifferentFirstRowIds) { .ValueOrDie()); ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array)); - if (GetParam() != "lance") { + { // read with row tracking auto expected_row_tracking_array = std::dynamic_pointer_cast( arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({ @@ -735,7 +735,7 @@ TEST_P(DataEvolutionTableTest, TestOnlyRowTrackingEnabled) { ASSERT_OK_AND_ASSIGN(auto commit_msgs, WriteArray(table_path, write_cols0, src_array0)); ASSERT_OK(Commit(table_path, commit_msgs)); - if (GetParam() != "lance") { + { // read with row tracking auto expected_row_tracking_array = std::dynamic_pointer_cast( arrow::ipc::internal::json::ArrayFromJSON( @@ -803,7 +803,7 @@ TEST_P(DataEvolutionTableTest, TestExternalPath) { .ValueOrDie()); ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array)); - if (GetParam() != "lance") { + { // read with row tracking auto expected_row_tracking_array = std::dynamic_pointer_cast( arrow::ipc::internal::json::ArrayFromJSON( @@ -881,7 +881,7 @@ TEST_P(DataEvolutionTableTest, TestWithPartitionSimple) { .ValueOrDie()); ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array)); - if (GetParam() != "lance") { + { // test only read partition fields auto expected_array_only_partition = std::dynamic_pointer_cast( arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[1]}), R"([ @@ -992,7 +992,7 @@ TEST_P(DataEvolutionTableTest, TestWithPartitionWithoutPartitionFieldsInFile) { /*row_ranges=*/row_ranges)); } - if (GetParam() != "lance") { + { // read with row tracking auto expected_row_tracking_array = std::dynamic_pointer_cast( arrow::ipc::internal::json::ArrayFromJSON( @@ -1012,7 +1012,7 @@ TEST_P(DataEvolutionTableTest, TestWithPartitionWithoutPartitionFieldsInFile) { TEST_P(DataEvolutionTableTest, TestPartitionWithPredicate) { auto file_format = GetParam(); - if (file_format == "lance" || file_format == "avro") { + if (file_format == "avro") { return; } std::vector partition_keys = {"f1"}; @@ -1185,7 +1185,7 @@ TEST_P(DataEvolutionTableTest, TestPartitionWithPredicate) { TEST_P(DataEvolutionTableTest, TestAlterTable) { auto file_format = GetParam(); - if (file_format == "lance" || file_format == "avro") { + if (file_format == "avro") { return; } std::string table_path = paimon::test::GetDataDir() + file_format + @@ -1282,7 +1282,7 @@ TEST_P(DataEvolutionTableTest, TestAlterTable) { TEST_P(DataEvolutionTableTest, TestReadCompactFiles) { auto file_format = GetParam(); - if (file_format == "lance" || file_format == "avro") { + if (file_format == "avro") { return; } std::string table_path = @@ -1312,7 +1312,7 @@ TEST_P(DataEvolutionTableTest, TestReadCompactFiles) { TEST_P(DataEvolutionTableTest, TestReadTableWithDenseStats) { auto file_format = GetParam(); - if (file_format == "lance" || file_format == "avro") { + if (file_format == "avro") { return; } std::string table_path = paimon::test::GetDataDir() + file_format + @@ -1393,7 +1393,7 @@ TEST_P(DataEvolutionTableTest, TestReadTableWithDenseStats) { TEST_P(DataEvolutionTableTest, TestScanAndReadWithIndex) { auto file_format = GetParam(); - if (file_format == "lance" || file_format == "avro") { + if (file_format == "avro") { return; } // only f2 has index @@ -1533,8 +1533,8 @@ TEST_P(DataEvolutionTableTest, TestScanAndReadWithIndex) { } TEST_P(DataEvolutionTableTest, TestPredicate) { - if (GetParam() == "lance" || GetParam() == "avro") { - // lance and avro do not have stats + if (GetParam() == "avro") { + // Avro does not have stats. return; } CreateTable(); @@ -1602,9 +1602,6 @@ TEST_P(DataEvolutionTableTest, TestPredicate) { } TEST_P(DataEvolutionTableTest, TestIOException) { - if (GetParam() == "lance") { - return; - } std::string table_path; // write and commit with I/O exception bool write_run_complete = false; @@ -1839,8 +1836,8 @@ TEST_P(DataEvolutionTableTest, TestWithRowIds) { /*predicate=*/nullptr, /*row_ranges=*/row_ranges)); } - if (GetParam() == "lance" || GetParam() == "avro") { - // as lance and avro do not support stats + if (GetParam() == "avro") { + // Avro does not support stats. return; } { @@ -1905,9 +1902,6 @@ std::vector GetTestValuesForDataEvolutionTableTest() { #ifdef PAIMON_ENABLE_ORC values.emplace_back("orc"); #endif -#ifdef PAIMON_ENABLE_LANCE - values.emplace_back("lance"); -#endif #ifdef PAIMON_ENABLE_AVRO values.emplace_back("avro"); #endif diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp index 6528c243..3091dc48 100644 --- a/test/inte/global_index_test.cpp +++ b/test/inte/global_index_test.cpp @@ -463,7 +463,7 @@ TEST_P(GlobalIndexTest, TestWriteIndexWithPartition) { #endif TEST_P(GlobalIndexTest, TestScanIndex) { - if (file_format_ == "lance" || file_format_ == "avro") { + if (file_format_ == "avro") { return; } @@ -632,7 +632,7 @@ TEST_P(GlobalIndexTest, TestScanIndex) { } TEST_P(GlobalIndexTest, TestScanIndexWithSpecificSnapshot) { - if (file_format_ == "lance" || file_format_ == "avro") { + if (file_format_ == "avro") { return; } @@ -681,7 +681,7 @@ TEST_P(GlobalIndexTest, TestScanIndexWithSpecificSnapshot) { } TEST_P(GlobalIndexTest, TestScanIndexWithSpecificSnapshotWithNoIndex) { - if (file_format_ == "lance" || file_format_ == "avro") { + if (file_format_ == "avro") { return; } @@ -706,7 +706,7 @@ TEST_P(GlobalIndexTest, TestScanIndexWithSpecificSnapshotWithNoIndex) { } TEST_P(GlobalIndexTest, TestScanIndexWithRange) { - if (file_format_ == "lance" || file_format_ == "avro") { + if (file_format_ == "avro") { return; } @@ -742,7 +742,7 @@ TEST_P(GlobalIndexTest, TestScanIndexWithRange) { } TEST_P(GlobalIndexTest, TestScanIndexWithPartition) { - if (file_format_ == "lance" || file_format_ == "avro") { + if (file_format_ == "avro") { return; } @@ -797,7 +797,7 @@ TEST_P(GlobalIndexTest, TestScanIndexWithPartition) { } TEST_P(GlobalIndexTest, TestScanUnregisteredIndex) { - if (file_format_ == "lance" || file_format_ == "avro") { + if (file_format_ == "avro") { return; } auto factory_creator = FactoryCreator::GetInstance(); @@ -1734,9 +1734,6 @@ TEST_P(GlobalIndexTest, TestDataEvolutionBatchScanWithExternalPath) { } TEST_P(GlobalIndexTest, TestIOException) { - if (file_format_ == "lance") { - return; - } arrow::FieldVector fields = { arrow::field("f0", arrow::utf8()), arrow::field("f1", arrow::list(arrow::float32())), arrow::field("f2", arrow::int32()), arrow::field("f3", arrow::float64())}; @@ -2811,7 +2808,7 @@ TEST_P(GlobalIndexTest, TestBTreeAndBitmapCoexist) { } TEST_P(GlobalIndexTest, TestBTreeScanWithPartitionWithMultiMeta) { - if (file_format_ == "lance" || file_format_ == "avro") { + if (file_format_ == "avro") { return; } std::string table_path = @@ -3113,10 +3110,6 @@ std::vector GetTestValuesForGlobalIndexTest() { values.emplace_back("orc", false); values.emplace_back("orc", true); #endif -#ifdef PAIMON_ENABLE_LANCE - values.emplace_back("lance", false); - values.emplace_back("lance", true); -#endif #ifdef PAIMON_ENABLE_AVRO values.emplace_back("avro", false); values.emplace_back("avro", true); diff --git a/test/inte/pk_compaction_inte_test.cpp b/test/inte/pk_compaction_inte_test.cpp index 6e2bb870..b9962209 100644 --- a/test/inte/pk_compaction_inte_test.cpp +++ b/test/inte/pk_compaction_inte_test.cpp @@ -2907,7 +2907,7 @@ TEST_F(PkCompactionInteTest, RemoteLookupFileWithSchemaEvolution) { // 6. ScanAndVerify after full compact TEST_P(PkCompactionInteTest, TestLookupCompatibility) { auto file_format = GetParam(); - if (file_format == "lance" || file_format == "avro") { + if (file_format == "avro") { return; } // Step 1: Copy pk_compact_lookup table to temp dir. @@ -3084,9 +3084,6 @@ std::vector GetTestValuesForCompactionInteTest() { #ifdef PAIMON_ENABLE_ORC values.emplace_back("orc"); #endif -#ifdef PAIMON_ENABLE_LANCE - values.emplace_back("lance"); -#endif #ifdef PAIMON_ENABLE_AVRO values.emplace_back("avro"); #endif diff --git a/test/inte/scan_and_read_inte_test.cpp b/test/inte/scan_and_read_inte_test.cpp index 572d402e..ad4c66fe 100644 --- a/test/inte/scan_and_read_inte_test.cpp +++ b/test/inte/scan_and_read_inte_test.cpp @@ -2182,67 +2182,6 @@ TEST_P(ScanAndReadInteTest, TestScanWithPredicateAndReadWithUnorderedFieldForPar ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); } -#ifdef PAIMON_ENABLE_LANCE -TEST_F(ScanAndReadInteTest, TestScanWithPredicateAndReadWithUnorderedFieldForLance) { - auto test_dir = UniqueTestDirectory::Create("local"); - arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()), - arrow::field("f1", arrow::int32()), - arrow::field("f2", arrow::float64())}; - auto schema = arrow::schema(fields); - std::map options = {{Options::MANIFEST_FORMAT, "orc"}, - {Options::FILE_FORMAT, "lance"}, - {Options::TARGET_FILE_SIZE, "1024"}, - {Options::BUCKET, "-1"}, - {Options::FILE_SYSTEM, "local"}}; - ASSERT_OK_AND_ASSIGN(auto helper, - TestHelper::Create(test_dir->Str(), schema, /*partition_keys=*/{}, - /*primary_keys=*/{}, options, - /*is_streaming_mode=*/false)); - std::string table_path = test_dir->Str() + "/foo.db/bar"; - int64_t commit_identifier = 0; - std::string data = R"([ - ["banana", 2, 3.5], - ["dog", 1, 2000.5], - ["lucy", 14, 10000.5], - ["mouse", 100, 10.5] - ])"; - ASSERT_OK_AND_ASSIGN(std::unique_ptr batch, - TestHelper::MakeRecordBatch(arrow::struct_(fields), data, - /*partition_map=*/{}, /*bucket=*/0, {})); - ASSERT_OK_AND_ASSIGN(auto commit_msgs, - helper->WriteAndCommit(std::move(batch), commit_identifier++, - /*expected_commit_messages=*/std::nullopt)); - ScanContextBuilder scan_context_builder(table_path); - // predicate does not take effective as lance file does not have stats - auto predicate = PredicateBuilder::GreaterThan( - /*field_index=*/2, /*field_name=*/"f2", FieldType::DOUBLE, Literal(50000.2)); - scan_context_builder.SetPredicate(predicate); - ASSERT_OK_AND_ASSIGN(auto scan_context, scan_context_builder.Finish()); - ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); - ASSERT_OK_AND_ASSIGN(auto result_plan, table_scan->CreatePlan()); - ASSERT_EQ(result_plan->SnapshotId().value(), 1); - - ReadContextBuilder read_context_builder(table_path); - read_context_builder.SetReadSchema({"f2", "f0"}); - ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish()); - ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context))); - ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(result_plan->Splits())); - ASSERT_OK_AND_ASSIGN(auto read_result, ReadResultCollector::CollectResult(batch_reader.get())); - - // check result - auto expected = std::make_shared( - arrow::ipc::internal::json::ArrayFromJSON( - arrow::struct_({arrow::field("_VALUE_KIND", arrow::int8()), fields[2], fields[0]}), - R"([[0, 3.5, "banana"], - [0, 2000.5, "dog"], - [0, 10000.5, "lucy"], - [0, 10.5, "mouse"]])") - .ValueOrDie()); - ASSERT_TRUE(expected); - ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString(); -} -#endif - TEST_P(ScanAndReadInteTest, TestAppendTableWithMultipleFileFormat) { auto [file_format, enable_prefetch] = GetParam(); if (file_format != "parquet") { diff --git a/test/inte/write_and_read_inte_test.cpp b/test/inte/write_and_read_inte_test.cpp index 603e32d8..3f9fc624 100644 --- a/test/inte/write_and_read_inte_test.cpp +++ b/test/inte/write_and_read_inte_test.cpp @@ -181,7 +181,7 @@ TEST_P(WriteAndReadInteTest, TestPKSimple) { } TEST_P(WriteAndReadInteTest, TestNestedType) { - // map use list(struct(key, value)) as lance does not support map + // Represent a map as list(struct(key, value)) for cross-format compatibility. arrow::FieldVector fields = { arrow::field("f1", arrow::list(arrow::struct_({arrow::field("key", arrow::int8()), arrow::field("value", arrow::int16())}))), @@ -858,9 +858,6 @@ std::vector> GetTestValuesForWriteAndReadInt values.emplace_back("orc", "local"); // values.emplace_back("parquet", "jindo"); #endif -#ifdef PAIMON_ENABLE_LANCE - values.emplace_back("lance", "local"); -#endif #ifdef PAIMON_ENABLE_AVRO values.emplace_back("avro", "local"); #endif diff --git a/test/inte/write_inte_test.cpp b/test/inte/write_inte_test.cpp index 236c6370..62c61b7b 100644 --- a/test/inte/write_inte_test.cpp +++ b/test/inte/write_inte_test.cpp @@ -196,10 +196,10 @@ class WriteInteTest : public testing::Test, public ::testing::WithParamInterface std::shared_ptr ReconstructDataFileMeta( const std::shared_ptr& file_meta) const { - if (GetParam() != "lance" && GetParam() != "avro") { + if (GetParam() != "avro") { return file_meta; } - // for lance and avro format, all stats is null + // For the avro format, all stats are null. auto new_meta = std::make_shared( file_meta->file_name, file_meta->file_size, file_meta->row_count, file_meta->min_key, file_meta->max_key, file_meta->key_stats, file_meta->value_stats, @@ -338,9 +338,6 @@ std::vector GetTestValuesForWriteInteTest() { #ifdef PAIMON_ENABLE_ORC values.emplace_back("orc"); #endif -#ifdef PAIMON_ENABLE_LANCE - values.emplace_back("lance"); -#endif #ifdef PAIMON_ENABLE_AVRO values.emplace_back("avro"); #endif @@ -760,10 +757,6 @@ TEST_P(WriteInteTest, TestAppendTableStreamWriteWithPartitionAndMultiBuckets) { } TEST_P(WriteInteTest, TestAppendTableWriteWithComplexType) { - if (GetParam() == "lance") { - // lance do not support map - return; - } auto dir = UniqueTestDirectory::Create(); arrow::FieldVector fields = { arrow::field("f1", arrow::map(arrow::int8(), arrow::int16())), @@ -1585,10 +1578,6 @@ TEST_P(WriteInteTest, TestPkTableWriteWithNoPartitionKey) { } TEST_P(WriteInteTest, TestPkTableWriteWithComplexType) { - if (GetParam() == "lance") { - // lance do not support map - return; - } auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); arrow::FieldVector fields = { @@ -2432,7 +2421,7 @@ TEST_P(WriteInteTest, TestWriteAndCommitIOException) { TEST_P(WriteInteTest, TestWriteWithFieldId) { auto file_format = GetParam(); - if (file_format == "lance" || file_format == "avro") { + if (file_format == "avro") { return; } // prepare write schema and write data @@ -3303,10 +3292,6 @@ TEST_P(WriteInteTest, TestWriteMemoryUse) { } TEST_P(WriteInteTest, TestAppendTableWithAllNull) { - if (GetParam() == "lance") { - // lance do not support map - return; - } auto dir = UniqueTestDirectory::Create(); arrow::FieldVector fields = { arrow::field("f0", arrow::boolean()), From c3b893237577cd9f156156d47272875827a6d91c Mon Sep 17 00:00:00 2001 From: lszskye <57179283+lszskye@users.noreply.github.com> Date: Thu, 4 Jun 2026 14:12:51 +0800 Subject: [PATCH 10/11] feat: remove support for FLOAT/DOUBLE type partition field (#333) --- src/paimon/core/io/field_mapping_reader.cpp | 10 --- .../core/io/field_mapping_reader_test.cpp | 71 +++++++------------ src/paimon/core/schema/schema_validation.cpp | 16 +++-- src/paimon/core/schema/schema_validation.h | 4 +- .../core/schema/schema_validation_test.cpp | 57 +++++++++------ .../core/stats/simple_stats_collector.cpp | 34 --------- .../stats/simple_stats_collector_test.cpp | 39 +++++----- src/paimon/core/utils/field_mapping_test.cpp | 51 ++++++------- 8 files changed, 116 insertions(+), 166 deletions(-) diff --git a/src/paimon/core/io/field_mapping_reader.cpp b/src/paimon/core/io/field_mapping_reader.cpp index afd5a5e1..148b62db 100644 --- a/src/paimon/core/io/field_mapping_reader.cpp +++ b/src/paimon/core/io/field_mapping_reader.cpp @@ -220,16 +220,6 @@ Result> FieldMappingReader::GenerateSinglePartitio scalar = std::make_shared(value); break; } - case arrow::Type::type::FLOAT: { - float value = partition_.GetFloat(partition_info_.value().idx_in_partition[idx]); - scalar = std::make_shared(value); - break; - } - case arrow::Type::type::DOUBLE: { - double value = partition_.GetDouble(partition_info_.value().idx_in_partition[idx]); - scalar = std::make_shared(value); - break; - } case arrow::Type::type::STRING: { BinaryString value = partition_.GetString(partition_info_.value().idx_in_partition[idx]); diff --git a/src/paimon/core/io/field_mapping_reader_test.cpp b/src/paimon/core/io/field_mapping_reader_test.cpp index 52775fc3..4b64a0b0 100644 --- a/src/paimon/core/io/field_mapping_reader_test.cpp +++ b/src/paimon/core/io/field_mapping_reader_test.cpp @@ -209,72 +209,53 @@ class FieldMappingReaderTest : public ::testing::Test { TEST_F(FieldMappingReaderTest, TestGenerateSinglePartitionArray) { PartitionInfo partition_info; - // read schema: p9-p0 - // partition key: p0-p9 - partition_info.partition_read_schema = {DataField(9, arrow::field("p9", arrow::date32())), - DataField(8, arrow::field("p8", arrow::binary())), - DataField(7, arrow::field("p7", arrow::utf8())), - DataField(6, arrow::field("p6", arrow::float64())), - DataField(5, arrow::field("p5", arrow::float32())), + // read schema: p7-p0 + // partition key: p0-p7 + partition_info.partition_read_schema = {DataField(7, arrow::field("p7", arrow::date32())), + DataField(6, arrow::field("p6", arrow::binary())), + DataField(5, arrow::field("p5", arrow::utf8())), DataField(4, arrow::field("p4", arrow::int64())), DataField(3, arrow::field("p3", arrow::int32())), DataField(2, arrow::field("p2", arrow::int16())), DataField(1, arrow::field("p1", arrow::int8())), DataField(0, arrow::field("p0", arrow::boolean()))}; - partition_info.idx_in_target_read_schema = {0, 1, 2, 3, 4, 5, 6, 7, 8, 9}; - partition_info.idx_in_partition = {9, 8, 7, 6, 5, 4, 3, 2, 1, 0}; + partition_info.idx_in_target_read_schema = {0, 1, 2, 3, 4, 5, 6, 7}; + partition_info.idx_in_partition = {7, 6, 5, 4, 3, 2, 1, 0}; NonPartitionInfo non_part_info; auto field_mapping = std::make_unique(partition_info, non_part_info, /*non_exist_field_info=*/std::nullopt); auto partition = BinaryRowGenerator::GenerateRow( {false, static_cast(1), static_cast(2), static_cast(3), - static_cast(4), static_cast(5.1), 6.21, std::string("7"), - std::make_shared("8", pool_.get()), 100}, + static_cast(4), std::string("5"), std::make_shared("6", pool_.get()), 100}, pool_.get()); auto mapping_reader = std::make_unique( - /*field_count=*/10, /*reader=*/nullptr, partition, std::move(field_mapping), pool_); + /*field_count=*/8, /*reader=*/nullptr, partition, std::move(field_mapping), pool_); - { - ASSERT_OK_AND_ASSIGN(auto p9_array, mapping_reader->GenerateSinglePartitionArray( - /*idx in read schema*/ 0, /*batch_size=*/2)); - ASSERT_EQ(p9_array->length(), 2); - ASSERT_EQ(arrow::internal::checked_cast(p9_array.get())->Value(0), - 100); - } - { - ASSERT_OK_AND_ASSIGN(auto p8_array, mapping_reader->GenerateSinglePartitionArray( - /*idx in read schema*/ 1, /*batch_size=*/2)); - ASSERT_EQ(p8_array->length(), 2); - ASSERT_EQ(arrow::internal::checked_cast(p8_array.get())->Value(0), - "8"); - } { ASSERT_OK_AND_ASSIGN(auto p7_array, mapping_reader->GenerateSinglePartitionArray( - /*idx in read schema*/ 2, /*batch_size=*/1)); - ASSERT_EQ(p7_array->length(), 1); - ASSERT_EQ(arrow::internal::checked_cast(p7_array.get())->Value(0), - "7"); + /*idx=*/0, /*batch_size=*/2)); + ASSERT_EQ(p7_array->length(), 2); + ASSERT_EQ(arrow::internal::checked_cast(p7_array.get())->Value(0), + 100); } { ASSERT_OK_AND_ASSIGN(auto p6_array, mapping_reader->GenerateSinglePartitionArray( - /*idx in read schema*/ 3, /*batch_size=*/1)); - ASSERT_EQ( - arrow::internal::checked_cast*>(p6_array.get()) - ->Value(0), - static_cast(6.21)); + /*idx=*/1, /*batch_size=*/2)); + ASSERT_EQ(p6_array->length(), 2); + ASSERT_EQ(arrow::internal::checked_cast(p6_array.get())->Value(0), + "6"); } { ASSERT_OK_AND_ASSIGN(auto p5_array, mapping_reader->GenerateSinglePartitionArray( - /*idx in read schema*/ 4, /*batch_size=*/1)); - ASSERT_EQ( - arrow::internal::checked_cast*>(p5_array.get()) - ->Value(0), - static_cast(5.1)); + /*idx=*/2, /*batch_size=*/1)); + ASSERT_EQ(p5_array->length(), 1); + ASSERT_EQ(arrow::internal::checked_cast(p5_array.get())->Value(0), + "5"); } { ASSERT_OK_AND_ASSIGN(auto p4_array, mapping_reader->GenerateSinglePartitionArray( - /*idx in read schema*/ 5, /*batch_size=*/1)); + /*idx=*/3, /*batch_size=*/1)); ASSERT_EQ( arrow::internal::checked_cast*>(p4_array.get()) ->Value(0), @@ -282,7 +263,7 @@ TEST_F(FieldMappingReaderTest, TestGenerateSinglePartitionArray) { } { ASSERT_OK_AND_ASSIGN(auto p3_array, mapping_reader->GenerateSinglePartitionArray( - /*idx in read schema*/ 6, /*batch_size=*/1)); + /*idx=*/4, /*batch_size=*/1)); ASSERT_EQ( arrow::internal::checked_cast*>(p3_array.get()) ->Value(0), @@ -290,7 +271,7 @@ TEST_F(FieldMappingReaderTest, TestGenerateSinglePartitionArray) { } { ASSERT_OK_AND_ASSIGN(auto p2_array, mapping_reader->GenerateSinglePartitionArray( - /*idx in read schema*/ 7, /*batch_size=*/1)); + /*idx=*/5, /*batch_size=*/1)); ASSERT_EQ( arrow::internal::checked_cast*>(p2_array.get()) ->Value(0), @@ -298,7 +279,7 @@ TEST_F(FieldMappingReaderTest, TestGenerateSinglePartitionArray) { } { ASSERT_OK_AND_ASSIGN(auto p1_array, mapping_reader->GenerateSinglePartitionArray( - /*idx in read schema*/ 8, /*batch_size=*/1)); + /*idx=*/6, /*batch_size=*/1)); ASSERT_EQ( arrow::internal::checked_cast*>(p1_array.get()) ->Value(0), @@ -306,7 +287,7 @@ TEST_F(FieldMappingReaderTest, TestGenerateSinglePartitionArray) { } { ASSERT_OK_AND_ASSIGN(auto p0_array, mapping_reader->GenerateSinglePartitionArray( - /*idx in read schema*/ 9, /*batch_size=*/1)); + /*idx=*/7, /*batch_size=*/1)); ASSERT_EQ(arrow::internal::checked_cast(p0_array.get())->Value(0), false); } diff --git a/src/paimon/core/schema/schema_validation.cpp b/src/paimon/core/schema/schema_validation.cpp index 28949a4b..17037ed9 100644 --- a/src/paimon/core/schema/schema_validation.cpp +++ b/src/paimon/core/schema/schema_validation.cpp @@ -73,9 +73,9 @@ Status SchemaValidation::ValidateTableSchema(const TableSchema& schema) { ValidateOnlyContainPrimitiveType(schema.Fields(), schema.PrimaryKeys(), "primary key")); PAIMON_RETURN_NOT_OK( ValidateOnlyContainPrimitiveType(schema.Fields(), schema.PartitionKeys(), "partition")); - // TODO(lisizhuo.lsz): C++ Paimon do not support timestamp & decimal type in partition keys for - // now. - PAIMON_RETURN_NOT_OK(ValidateNotContainComplexType(schema.Fields(), schema.PartitionKeys())); + // TODO(lisizhuo.lsz): C++ Paimon do not support timestamp & decimal & float & double type in + // partition keys for now. + PAIMON_RETURN_NOT_OK(ValidateNotContainSpecificType(schema.Fields(), schema.PartitionKeys())); PAIMON_ASSIGN_OR_RAISE(CoreOptions options, CoreOptions::FromMap(schema.Options())); PAIMON_RETURN_NOT_OK(ValidateBucket(schema, options)); @@ -163,7 +163,7 @@ Status SchemaValidation::ValidateOnlyContainPrimitiveType( return Status::OK(); } -Status SchemaValidation::ValidateNotContainComplexType( +Status SchemaValidation::ValidateNotContainSpecificType( const std::vector& fields, const std::vector& field_names) { if (field_names.empty()) { return Status::OK(); @@ -178,8 +178,12 @@ Status SchemaValidation::ValidateNotContainComplexType( auto field = it->second; if (IsComplexType(field)) { return Status::Invalid( - fmt::format("The field {} in partition field {} is unsupported", - field->ToString(), it->first)); + fmt::format("partition field {} cannot be TIMESTAMP/DECIMAL/BLOB", field_name)); + } + if (field->type()->id() == arrow::Type::FLOAT || + field->type()->id() == arrow::Type::DOUBLE) { + return Status::Invalid( + fmt::format("partition field {} cannot be FLOAT/DOUBLE", field_name)); } } else { assert(false); diff --git a/src/paimon/core/schema/schema_validation.h b/src/paimon/core/schema/schema_validation.h index af73ee6b..778f8a35 100644 --- a/src/paimon/core/schema/schema_validation.h +++ b/src/paimon/core/schema/schema_validation.h @@ -54,8 +54,8 @@ class SchemaValidation { static Status ValidateOnlyContainPrimitiveType(const std::vector& fields, const std::vector& field_names, const std::string& error_message_intro); - static Status ValidateNotContainComplexType(const std::vector& fields, - const std::vector& field_names); + static Status ValidateNotContainSpecificType(const std::vector& fields, + const std::vector& field_names); static Status ValidateBucket(const TableSchema& schema, const CoreOptions& options); static Status ValidateDefaultValues(const TableSchema& schema) { return Status::NotImplemented("validate default values not implemented"); diff --git a/src/paimon/core/schema/schema_validation_test.cpp b/src/paimon/core/schema/schema_validation_test.cpp index 0a08efe0..9dc7472a 100644 --- a/src/paimon/core/schema/schema_validation_test.cpp +++ b/src/paimon/core/schema/schema_validation_test.cpp @@ -393,19 +393,34 @@ TEST(SchemaValidationTest, NonPrimitivePartitionKeyStruct) { "field f1 is unsupported"); } -TEST(SchemaValidationTest, TestComplexPartitionKey) { - auto f0 = arrow::field("f0", arrow::utf8()); - auto f1 = arrow::field("f1", arrow::decimal128(5, 2)); - auto f2 = arrow::field("f2", arrow::float64()); - arrow::FieldVector fields = {f0, f1, f2}; - auto schema = arrow::schema(fields); - std::vector primary_keys = {"f0", "f1"}; - std::vector partition_keys = {"f1"}; - ASSERT_OK_AND_ASSIGN( - std::shared_ptr table_schema, - TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, {})); - ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema), - "partition field f1 is unsupported"); +TEST(SchemaValidationTest, TestSpecificPartitionKey) { + { + auto f0 = arrow::field("f0", arrow::utf8()); + auto f1 = arrow::field("f1", arrow::decimal128(5, 2)); + auto f2 = arrow::field("f2", arrow::float64()); + arrow::FieldVector fields = {f0, f1, f2}; + auto schema = arrow::schema(fields); + std::vector primary_keys = {"f0", "f1"}; + std::vector partition_keys = {"f1"}; + ASSERT_OK_AND_ASSIGN( + std::shared_ptr table_schema, + TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, {})); + ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema), + "partition field f1 cannot be TIMESTAMP/DECIMAL/BLOB"); + } + { + auto f0 = arrow::field("f0", arrow::utf8()); + auto f1 = arrow::field("f1", arrow::float64()); + arrow::FieldVector fields = {f0, f1}; + auto schema = arrow::schema(fields); + std::vector primary_keys = {"f0", "f1"}; + std::vector partition_keys = {"f1"}; + ASSERT_OK_AND_ASSIGN( + std::shared_ptr table_schema, + TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, {})); + ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema), + "partition field f1 cannot be FLOAT/DOUBLE"); + } } TEST(SchemaValidationTest, TestComplexPartitionKeyWithBlob) { @@ -419,7 +434,7 @@ TEST(SchemaValidationTest, TestComplexPartitionKeyWithBlob) { std::shared_ptr table_schema, TableSchema::Create(/*schema_id=*/0, schema, partition_keys, /*primary_keys=*/{}, {})); ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema), - "partition field f1 is unsupported"); + "partition field f1 cannot be TIMESTAMP/DECIMAL/BLOB"); } TEST(SchemaValidationTest, TestDateTypePartitionKey) { @@ -511,8 +526,8 @@ TEST(SchemaValidationTest, ValidateBucket) { "The number of buckets needs to be greater than 0."); } { - std::vector primary_keys = {"f0", "f1"}; - std::vector partition_keys = {"f2"}; + std::vector primary_keys = {"f0", "f2"}; + std::vector partition_keys = {"f1"}; std::map options = {{Options::BUCKET, "2"}, {Options::BUCKET_KEY, "f0"}}; ASSERT_OK_AND_ASSIGN( @@ -524,7 +539,7 @@ TEST(SchemaValidationTest, ValidateBucket) { } { std::vector primary_keys = {}; - std::vector partition_keys = {"f2"}; + std::vector partition_keys = {"f1"}; std::map options = {{Options::BUCKET, "2"}}; ASSERT_OK_AND_ASSIGN( std::shared_ptr table_schema, @@ -533,7 +548,7 @@ TEST(SchemaValidationTest, ValidateBucket) { "You should define a 'bucket-key' for bucketed append mode"); } { - std::vector partition_keys = {"f2"}; + std::vector partition_keys = {"f1"}; std::map options = {{"full-compaction.delta-commits", "2"}}; ASSERT_OK_AND_ASSIGN(std::shared_ptr table_schema, TableSchema::Create(/*schema_id=*/0, schema, partition_keys, @@ -633,11 +648,11 @@ TEST(SchemaValidationTest, ValidateSequenceField) { std::map options = {{Options::BUCKET, "-1"}, {Options::SEQUENCE_FIELD, "f0,f1,f2"}}; ASSERT_OK_AND_ASSIGN(std::shared_ptr table_schema, - TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{"f2"}, - /*primary_keys=*/{"f0", "f1"}, options)); + TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{"f1"}, + /*primary_keys=*/{"f0", "f2"}, options)); ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema), "You cannot use sequence.field in cross partition update case (Primary " - "key constraint 'f0, f1' not including all partition fields 'f2')."); + "key constraint 'f0, f2' not including all partition fields 'f1')."); } } diff --git a/src/paimon/core/stats/simple_stats_collector.cpp b/src/paimon/core/stats/simple_stats_collector.cpp index 7cd04404..c0f2c8f4 100644 --- a/src/paimon/core/stats/simple_stats_collector.cpp +++ b/src/paimon/core/stats/simple_stats_collector.cpp @@ -133,40 +133,6 @@ Status SimpleStatsCollector::Collect(const BinaryRow& row) { } break; } - case arrow::Type::FLOAT: { - if (column_stats_[i] == nullptr) { - column_stats_[i] = ColumnStats::CreateFloatColumnStats( - std::nullopt, std::nullopt, std::nullopt); - } - auto typed_stats = dynamic_cast(column_stats_[i].get()); - if (typed_stats == nullptr) { - assert(false); - return Status::Invalid("cast typed stats failed"); - } - if (!row.IsNullAt(i)) { - typed_stats->Collect(row.GetFloat(i)); - } else { - typed_stats->Collect(std::nullopt); - } - break; - } - case arrow::Type::DOUBLE: { - if (column_stats_[i] == nullptr) { - column_stats_[i] = ColumnStats::CreateDoubleColumnStats( - std::nullopt, std::nullopt, std::nullopt); - } - auto typed_stats = dynamic_cast(column_stats_[i].get()); - if (typed_stats == nullptr) { - assert(false); - return Status::Invalid("cast typed stats failed"); - } - if (!row.IsNullAt(i)) { - typed_stats->Collect(row.GetDouble(i)); - } else { - typed_stats->Collect(std::nullopt); - } - break; - } case arrow::Type::STRING: case arrow::Type::BINARY: { if (column_stats_[i] == nullptr) { diff --git a/src/paimon/core/stats/simple_stats_collector_test.cpp b/src/paimon/core/stats/simple_stats_collector_test.cpp index bac30b4e..d3a860ad 100644 --- a/src/paimon/core/stats/simple_stats_collector_test.cpp +++ b/src/paimon/core/stats/simple_stats_collector_test.cpp @@ -41,9 +41,8 @@ TEST(SimpleStatsCollectorTest, TestSimple) { arrow::FieldVector fields = { arrow::field("f0", arrow::boolean()), arrow::field("f1", arrow::int8()), arrow::field("f2", arrow::int16()), arrow::field("f3", arrow::int32()), - arrow::field("f4", arrow::int64()), arrow::field("f5", arrow::float32()), - arrow::field("f6", arrow::float64()), arrow::field("f7", arrow::utf8()), - arrow::field("f8", arrow::date32()), + arrow::field("f4", arrow::int64()), arrow::field("f5", arrow::utf8()), + arrow::field("f6", arrow::date32()), }; auto schema = arrow::schema(fields); @@ -51,25 +50,21 @@ TEST(SimpleStatsCollectorTest, TestSimple) { auto pool = GetDefaultPool(); ASSERT_OK(collector.Collect(BinaryRowGenerator::GenerateRow( {true, static_cast(1), static_cast(1), static_cast(1), - static_cast(1), static_cast(3.0), static_cast(3.0), - std::string("abc"), 2025}, + static_cast(1), std::string("abc"), 2025}, pool.get()))); ASSERT_OK(collector.Collect(BinaryRowGenerator::GenerateRow( {false, static_cast(2), static_cast(2), static_cast(2), - static_cast(2), static_cast(6.0), static_cast(6.0), - std::string("bcd"), 2026}, + static_cast(2), std::string("bcd"), 2026}, pool.get()))); ASSERT_OK_AND_ASSIGN(auto col_stats, collector.GetResult()); ASSERT_OK_AND_ASSIGN(SimpleStats stats, SimpleStatsConverter::ToBinary(col_stats, pool.get())); auto expected_stats = BinaryRowGenerator::GenerateStats( {false, static_cast(1), static_cast(1), static_cast(1), - static_cast(1), static_cast(3.0), static_cast(3.0), - std::string("abc"), 2025}, + static_cast(1), std::string("abc"), 2025}, {true, static_cast(2), static_cast(2), static_cast(2), - static_cast(2), static_cast(6.0), static_cast(6.0), - std::string("bcd"), 2026}, - std::vector({0, 0, 0, 0, 0, 0, 0, 0, 0}), GetDefaultPool().get()); + static_cast(2), std::string("bcd"), 2026}, + std::vector({0, 0, 0, 0, 0, 0, 0}), GetDefaultPool().get()); ASSERT_EQ(stats, expected_stats); } @@ -78,29 +73,27 @@ TEST(SimpleStatsCollectorTest, TestNull) { arrow::FieldVector fields = { arrow::field("f0", arrow::boolean()), arrow::field("f1", arrow::int8()), arrow::field("f2", arrow::int16()), arrow::field("f3", arrow::int32()), - arrow::field("f4", arrow::int64()), arrow::field("f5", arrow::float32()), - arrow::field("f6", arrow::float64()), arrow::field("f7", arrow::utf8()), - arrow::field("f8", arrow::date32()), arrow::field("key", arrow::int32()), + arrow::field("f4", arrow::int64()), arrow::field("f5", arrow::utf8()), + arrow::field("f6", arrow::date32()), arrow::field("key", arrow::int32()), }; auto schema = arrow::schema(fields); SimpleStatsCollector collector(schema); auto pool = GetDefaultPool(); - ASSERT_OK(collector.Collect( - BinaryRowGenerator::GenerateRow({NullType(), NullType(), NullType(), NullType(), NullType(), - NullType(), NullType(), NullType(), NullType(), 100}, - pool.get()))); + ASSERT_OK(collector.Collect(BinaryRowGenerator::GenerateRow( + {NullType(), NullType(), NullType(), NullType(), NullType(), NullType(), NullType(), 100}, + pool.get()))); ASSERT_OK_AND_ASSIGN(auto col_stats, collector.GetResult()); - ASSERT_EQ(10, col_stats.size()); + ASSERT_EQ(8, col_stats.size()); ASSERT_OK_AND_ASSIGN(SimpleStats stats, SimpleStatsConverter::ToBinary(col_stats, pool.get())); ASSERT_EQ(stats.MinValues(), stats.MaxValues()); - for (size_t i = 0; i < 9; ++i) { + for (size_t i = 0; i < 7; ++i) { ASSERT_TRUE(stats.MinValues().IsNullAt(i)); } - ASSERT_EQ(stats.MinValues().GetInt(9), 100); + ASSERT_EQ(stats.MinValues().GetInt(7), 100); ASSERT_OK_AND_ASSIGN(std::vector expected, stats.NullCounts().ToLongArray()); - ASSERT_EQ(expected, std::vector({1l, 1l, 1l, 1l, 1l, 1l, 1l, 1l, 1l, 0l})); + ASSERT_EQ(expected, std::vector({1l, 1l, 1l, 1l, 1l, 1l, 1l, 0l})); } TEST(SimpleStatsCollectorTest, TestInvalidPartition) { diff --git a/src/paimon/core/utils/field_mapping_test.cpp b/src/paimon/core/utils/field_mapping_test.cpp index ed1b0560..153932df 100644 --- a/src/paimon/core/utils/field_mapping_test.cpp +++ b/src/paimon/core/utils/field_mapping_test.cpp @@ -154,15 +154,19 @@ TEST_F(FieldMappingTest, TestPartitionKeysEqualSchema) { FieldType::INT, Literal(20)); ASSERT_OK_AND_ASSIGN(auto predicate, PredicateBuilder::And({equal, not_equal})); - std::vector partition_keys = {"f0", "f1", "f2", "f3"}; + std::vector fields = {DataField(0, arrow::field("f0", arrow::utf8())), + DataField(1, arrow::field("f1", arrow::int32())), + DataField(2, arrow::field("f2", arrow::int32()))}; + std::shared_ptr schema = DataField::ConvertDataFieldsToArrowSchema(fields); + std::vector partition_keys = {"f0", "f1", "f2"}; ASSERT_OK_AND_ASSIGN(auto mapping_builder, - FieldMappingBuilder::Create(schema_, partition_keys, predicate)); - ASSERT_OK_AND_ASSIGN(auto mapping, mapping_builder->CreateFieldMapping(schema_)); + FieldMappingBuilder::Create(schema, partition_keys, predicate)); + ASSERT_OK_AND_ASSIGN(auto mapping, mapping_builder->CreateFieldMapping(schema)); PartitionInfo expected_part_info; - expected_part_info.partition_read_schema = fields_; - expected_part_info.idx_in_target_read_schema = {0, 1, 2, 3}; - expected_part_info.idx_in_partition = {0, 1, 2, 3}; + expected_part_info.partition_read_schema = fields; + expected_part_info.idx_in_target_read_schema = {0, 1, 2}; + expected_part_info.idx_in_partition = {0, 1, 2}; expected_part_info.partition_filter = std::dynamic_pointer_cast(predicate); CheckPartitionInfo(mapping->partition_info.value(), expected_part_info); @@ -356,7 +360,7 @@ TEST_F(FieldMappingTest, TestSchemaEvolution) { // add field / delete field / rename / casting // without predicate std::vector data_fields = {DataField(0, arrow::field("key0", arrow::int32())), - DataField(1, arrow::field("key1", arrow::float64())), + DataField(1, arrow::field("key1", arrow::int64())), DataField(2, arrow::field("a", arrow::int32())), DataField(3, arrow::field("b", arrow::int32())), DataField(4, arrow::field("c", arrow::int32())), @@ -365,7 +369,7 @@ TEST_F(FieldMappingTest, TestSchemaEvolution) { DataField::ConvertDataFieldsToArrowSchema(data_fields); std::vector read_fields = {DataField(0, arrow::field("key0", arrow::int32())), - DataField(1, arrow::field("key1", arrow::float64())), + DataField(1, arrow::field("key1", arrow::int64())), DataField(3, arrow::field("c", arrow::int64())), DataField(5, arrow::field("a", arrow::float32())), DataField(7, arrow::field("d", arrow::int32())), @@ -380,9 +384,8 @@ TEST_F(FieldMappingTest, TestSchemaEvolution) { ASSERT_OK_AND_ASSIGN(auto mapping, mapping_builder->CreateFieldMapping(data_fields)); PartitionInfo expected_part_info; - expected_part_info.partition_read_schema = { - DataField(0, arrow::field("key0", arrow::int32())), - DataField(1, arrow::field("key1", arrow::float64()))}; + expected_part_info.partition_read_schema = {DataField(0, arrow::field("key0", arrow::int32())), + DataField(1, arrow::field("key1", arrow::int64()))}; expected_part_info.idx_in_target_read_schema = {0, 1}; expected_part_info.idx_in_partition = {0, 1}; expected_part_info.partition_filter = nullptr; @@ -420,7 +423,7 @@ TEST_F(FieldMappingTest, TestSchemaEvolutionWithPredicate) { // field_7 is added to the middle; // field_8 is added to the last field. std::vector data_fields = {DataField(0, arrow::field("key0", arrow::int32())), - DataField(1, arrow::field("key1", arrow::float64())), + DataField(1, arrow::field("key1", arrow::int64())), DataField(2, arrow::field("a", arrow::int32())), DataField(3, arrow::field("b", arrow::decimal128(5, 2))), DataField(4, arrow::field("c", arrow::int32())), @@ -428,7 +431,7 @@ TEST_F(FieldMappingTest, TestSchemaEvolutionWithPredicate) { DataField(6, arrow::field("k", arrow::int32()))}; std::vector table_fields = {DataField(0, arrow::field("key0", arrow::int32())), - DataField(1, arrow::field("key1", arrow::float64())), + DataField(1, arrow::field("key1", arrow::int64())), DataField(6, arrow::field("k", arrow::int32())), DataField(3, arrow::field("c", arrow::decimal128(6, 3))), DataField(7, arrow::field("d", arrow::int32())), @@ -441,7 +444,7 @@ TEST_F(FieldMappingTest, TestSchemaEvolutionWithPredicate) { auto greater_or_equal = PredicateBuilder::GreaterOrEqual( /*field_index=*/0, /*field_name=*/"key0", FieldType::INT, Literal(4)); auto equal = PredicateBuilder::Equal(/*field_index=*/1, /*field_name=*/"key1", - FieldType::DOUBLE, Literal(3.0)); + FieldType::BIGINT, Literal(3l)); auto less_or_equal = PredicateBuilder::LessOrEqual(/*field_index=*/2, /*field_name=*/"k", FieldType::INT, Literal(10)); // greater_than will not be pushed down, as with casting, only integer predicates can be pushed @@ -466,9 +469,8 @@ TEST_F(FieldMappingTest, TestSchemaEvolutionWithPredicate) { ASSERT_OK_AND_ASSIGN(auto mapping, mapping_builder->CreateFieldMapping(data_fields)); PartitionInfo expected_part_info; - expected_part_info.partition_read_schema = { - DataField(0, arrow::field("key0", arrow::int32())), - DataField(1, arrow::field("key1", arrow::float64()))}; + expected_part_info.partition_read_schema = {DataField(0, arrow::field("key0", arrow::int32())), + DataField(1, arrow::field("key1", arrow::int64()))}; expected_part_info.idx_in_target_read_schema = {0, 1}; expected_part_info.idx_in_partition = {0, 1}; expected_part_info.partition_filter = @@ -514,7 +516,7 @@ TEST_F(FieldMappingTest, TestSchemaEvolutionWithPredicate2) { // field_7 is added to the middle; // field_8 is added to the last field. std::vector data_fields = {DataField(0, arrow::field("key0", arrow::int32())), - DataField(1, arrow::field("key1", arrow::float64())), + DataField(1, arrow::field("key1", arrow::int64())), DataField(2, arrow::field("a", arrow::int32())), DataField(3, arrow::field("b", arrow::decimal128(5, 2))), DataField(4, arrow::field("c", arrow::int32())), @@ -522,7 +524,7 @@ TEST_F(FieldMappingTest, TestSchemaEvolutionWithPredicate2) { DataField(6, arrow::field("k", arrow::int32()))}; std::vector table_fields = {DataField(0, arrow::field("key0", arrow::int32())), - DataField(1, arrow::field("key1", arrow::float64())), + DataField(1, arrow::field("key1", arrow::int64())), DataField(6, arrow::field("k", arrow::int32())), DataField(3, arrow::field("c", arrow::decimal128(6, 3))), DataField(7, arrow::field("d", arrow::int32())), @@ -531,7 +533,7 @@ TEST_F(FieldMappingTest, TestSchemaEvolutionWithPredicate2) { // the field order of read schema and table schema is inconsistent std::vector read_fields = {DataField(7, arrow::field("d", arrow::int32())), - DataField(1, arrow::field("key1", arrow::float64())), + DataField(1, arrow::field("key1", arrow::int64())), DataField(5, arrow::field("a", arrow::int64())), DataField(8, arrow::field("e", arrow::int32())), DataField(6, arrow::field("k", arrow::int32())), @@ -543,7 +545,7 @@ TEST_F(FieldMappingTest, TestSchemaEvolutionWithPredicate2) { auto greater_or_equal = PredicateBuilder::GreaterOrEqual( /*field_index=*/6, /*field_name=*/"key0", FieldType::INT, Literal(4)); auto equal = PredicateBuilder::Equal(/*field_index=*/1, /*field_name=*/"key1", - FieldType::DOUBLE, Literal(3.0)); + FieldType::BIGINT, Literal(3l)); auto less_or_equal = PredicateBuilder::LessOrEqual(/*field_index=*/4, /*field_name=*/"k", FieldType::INT, Literal(10)); // greater_than will not be pushed down, as with casting, only integer predicates can be pushed @@ -568,15 +570,14 @@ TEST_F(FieldMappingTest, TestSchemaEvolutionWithPredicate2) { ASSERT_OK_AND_ASSIGN(auto mapping, mapping_builder->CreateFieldMapping(data_fields)); PartitionInfo expected_part_info; - expected_part_info.partition_read_schema = { - DataField(0, arrow::field("key0", arrow::int32())), - DataField(1, arrow::field("key1", arrow::float64()))}; + expected_part_info.partition_read_schema = {DataField(0, arrow::field("key0", arrow::int32())), + DataField(1, arrow::field("key1", arrow::int64()))}; expected_part_info.idx_in_target_read_schema = {6, 1}; expected_part_info.idx_in_partition = {0, 1}; auto greater_or_equal_new = PredicateBuilder::GreaterOrEqual( /*field_index=*/0, /*field_name=*/"key0", FieldType::INT, Literal(4)); auto equal_new = PredicateBuilder::Equal(/*field_index=*/1, /*field_name=*/"key1", - FieldType::DOUBLE, Literal(3.0)); + FieldType::BIGINT, Literal(3l)); expected_part_info.partition_filter = PredicateBuilder::And({greater_or_equal_new, equal_new}).value_or(nullptr); CheckPartitionInfo(mapping->partition_info.value(), expected_part_info); From e83c21e5867ea5bab32c045c590c90ce438f4d4e Mon Sep 17 00:00:00 2001 From: "jinli.zjw" Date: Thu, 23 Jul 2026 10:31:27 +0800 Subject: [PATCH 11/11] refactor: remove unused path-based reader builder --- include/paimon/format/reader_builder.h | 4 ---- .../common/reader/prefetch_file_batch_reader_impl_test.cpp | 4 ---- src/paimon/format/avro/avro_reader_builder.h | 4 ---- src/paimon/format/blob/blob_reader_builder.h | 4 ---- src/paimon/format/orc/orc_reader_builder.h | 4 ---- src/paimon/format/parquet/parquet_reader_builder.h | 4 ---- src/paimon/testing/mock/mock_format_reader_builder.h | 4 ---- 7 files changed, 28 deletions(-) diff --git a/include/paimon/format/reader_builder.h b/include/paimon/format/reader_builder.h index 91b604dd..ca2021c3 100644 --- a/include/paimon/format/reader_builder.h +++ b/include/paimon/format/reader_builder.h @@ -19,7 +19,6 @@ #pragma once #include -#include #include "paimon/memory/memory_pool.h" #include "paimon/reader/file_batch_reader.h" @@ -38,9 +37,6 @@ class PAIMON_EXPORT ReaderBuilder { /// Build a file batch reader based on the created `InputStream`. virtual Result> Build( const std::shared_ptr& path) const = 0; - - /// Build a file batch reader based on the file path. - virtual Result> Build(const std::string& path) const = 0; }; } // namespace paimon diff --git a/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp b/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp index e9832c19..d9b58218 100644 --- a/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp +++ b/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp @@ -100,10 +100,6 @@ class ControlledMockFormatReaderBuilder : public ReaderBuilder { data_, schema_, read_batch_size_, read_ranges_, need_prefetch_, set_read_ranges_status); } - Result> Build(const std::string& path) const override { - return Status::Invalid("do not support build reader with path in mock format"); - } - private: std::shared_ptr data_; std::shared_ptr schema_; diff --git a/src/paimon/format/avro/avro_reader_builder.h b/src/paimon/format/avro/avro_reader_builder.h index b4100d1e..bd1da1a2 100644 --- a/src/paimon/format/avro/avro_reader_builder.h +++ b/src/paimon/format/avro/avro_reader_builder.h @@ -47,10 +47,6 @@ class AvroReaderBuilder : public ReaderBuilder { return AvroFileBatchReader::Create(path, batch_size_, pool_); } - Result> Build(const std::string& path) const override { - return Status::Invalid("do not support build reader with path in avro format"); - } - private: const int32_t batch_size_; std::shared_ptr pool_; diff --git a/src/paimon/format/blob/blob_reader_builder.h b/src/paimon/format/blob/blob_reader_builder.h index 0bffbf31..85bebe96 100644 --- a/src/paimon/format/blob/blob_reader_builder.h +++ b/src/paimon/format/blob/blob_reader_builder.h @@ -48,10 +48,6 @@ class BlobReaderBuilder : public ReaderBuilder { return BlobFileBatchReader::Create(input_stream, batch_size_, blob_as_descriptor, pool_); } - Result> Build(const std::string& path) const override { - return Status::Invalid("do not support build reader with path in blob format"); - } - private: int32_t batch_size_; std::shared_ptr pool_; diff --git a/src/paimon/format/orc/orc_reader_builder.h b/src/paimon/format/orc/orc_reader_builder.h index 3fd88d87..e7ef5402 100644 --- a/src/paimon/format/orc/orc_reader_builder.h +++ b/src/paimon/format/orc/orc_reader_builder.h @@ -54,10 +54,6 @@ class OrcReaderBuilder : public ReaderBuilder { return OrcFileBatchReader::Create(std::move(input_stream), pool_, options_, batch_size_); } - Result> Build(const std::string& path) const override { - return Status::Invalid("do not support build reader with path in orc format"); - } - private: int32_t batch_size_ = -1; std::shared_ptr pool_; diff --git a/src/paimon/format/parquet/parquet_reader_builder.h b/src/paimon/format/parquet/parquet_reader_builder.h index 2a93696c..c0d99315 100644 --- a/src/paimon/format/parquet/parquet_reader_builder.h +++ b/src/paimon/format/parquet/parquet_reader_builder.h @@ -53,10 +53,6 @@ class ParquetReaderBuilder : public ReaderBuilder { batch_size_); } - Result> Build(const std::string& path) const override { - return Status::Invalid("do not support build reader with path in parquet format"); - } - private: int32_t batch_size_ = -1; std::shared_ptr pool_; diff --git a/src/paimon/testing/mock/mock_format_reader_builder.h b/src/paimon/testing/mock/mock_format_reader_builder.h index 0a7b471b..7f8906cd 100644 --- a/src/paimon/testing/mock/mock_format_reader_builder.h +++ b/src/paimon/testing/mock/mock_format_reader_builder.h @@ -54,10 +54,6 @@ class MockFormatReaderBuilder : public ReaderBuilder { return std::make_unique(data_, schema_, read_batch_size_); } - Result> Build(const std::string& path) const override { - return Status::Invalid("do not support build reader with path in mock format"); - } - private: std::shared_ptr data_; std::shared_ptr schema_;