This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 131b092 fix: sync missing changes after repository migration (#154)
131b092 is described below
commit 131b0921663a2f7384b863124657a9fcfc0f6f21
Author: Zhang Jiawei <[email protected]>
AuthorDate: Thu Jul 23 14:01:55 2026 +0800
fix: sync missing changes after repository migration (#154)
* fix(build): make package config relocatable (#306)
* fix: fix thread-safe problem for file store path factory (#323)
* feat(lumina): support null values in vector column during index building
(#310)
* feat: add CreateReader api with field_name & index_type (#391)
* test: add tests for executor (#396)
* fix: fix IsThreadSafe() in UnionGlobalIndexReader (#395)
* fix: release global index writer before reader to avoid mem issue when
write index failed (#398)
* 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.
* chore: remove remaining Lance references
* feat: remove support for FLOAT/DOUBLE type partition field (#333)
* refactor: remove unused path-based reader builder
---------
Co-authored-by: Yonghao Fang <[email protected]>
Co-authored-by: lxy <[email protected]>
Co-authored-by: lszskye <[email protected]>
---
CMakeLists.txt | 37 +---
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 ++-
include/paimon/format/reader_builder.h | 4 -
include/paimon/global_index/global_index_scan.h | 16 ++
src/paimon/common/data/binary_array_writer.cpp | 3 +-
src/paimon/common/data/binary_row_writer.h | 3 +-
.../common/executor/default_executor_test.cpp | 57 ++++++
.../global_index/union_global_index_reader.cpp | 11 +-
.../global_index/union_global_index_reader.h | 4 +-
.../union_global_index_reader_test.cpp | 27 ++-
.../prefetch_file_batch_reader_impl_test.cpp | 4 -
src/paimon/common/utils/file_type_test.cpp | 2 -
.../core/global_index/global_index_scan_impl.cpp | 33 +++-
.../core/global_index/global_index_scan_impl.h | 8 +-
.../core/global_index/global_index_write_task.cpp | 14 +-
src/paimon/core/io/field_mapping_reader.cpp | 10 --
src/paimon/core/io/field_mapping_reader_test.cpp | 71 +++-----
src/paimon/core/operation/abstract_split_read.cpp | 4 -
.../core/operation/data_evolution_split_read.cpp | 2 +-
.../core/operation/orphan_files_cleaner_impl.cpp | 2 +-
.../core/operation/orphan_files_cleaner_test.cpp | 2 -
src/paimon/core/operation/raw_file_split_read.cpp | 2 +-
src/paimon/core/schema/schema_validation.cpp | 16 +-
src/paimon/core/schema/schema_validation.h | 4 +-
src/paimon/core/schema/schema_validation_test.cpp | 57 +++---
src/paimon/core/stats/simple_stats_collector.cpp | 34 ----
.../core/stats/simple_stats_collector_test.cpp | 39 ++--
src/paimon/core/utils/field_mapping_test.cpp | 51 +++---
src/paimon/core/utils/file_store_path_factory.cpp | 28 ++-
src/paimon/core/utils/file_store_path_factory.h | 5 +-
.../core/utils/file_store_path_factory_test.cpp | 89 +++++++++
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 -
.../global_index/lumina/lumina_global_index.cpp | 71 +++++---
.../global_index/lumina/lumina_global_index.h | 2 +
.../lumina/lumina_global_index_test.cpp | 199 ++++++++++++++++++++-
.../testing/mock/mock_format_reader_builder.h | 4 -
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 | 124 ++++++++++---
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 +--
51 files changed, 822 insertions(+), 601 deletions(-)
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 886cd96..1726103 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -70,18 +70,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()
@@ -100,9 +94,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)
@@ -292,13 +286,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})
@@ -330,10 +317,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})
@@ -383,11 +366,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)
@@ -425,11 +403,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()
@@ -442,9 +420,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()
@@ -452,3 +427,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 e56d9d5..559417a 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 235434a..16e6fe0 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 c57336d..5f93160 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
"$<BUILD_INTERFACE:paimon_sanitizer_flags>")
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 6a61b68..8bcf714 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 082cf04..4fe7a8b 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)
diff --git a/include/paimon/format/reader_builder.h
b/include/paimon/format/reader_builder.h
index 91b604d..ca2021c 100644
--- a/include/paimon/format/reader_builder.h
+++ b/include/paimon/format/reader_builder.h
@@ -19,7 +19,6 @@
#pragma once
#include <memory>
-#include <string>
#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<std::unique_ptr<FileBatchReader>> Build(
const std::shared_ptr<InputStream>& path) const = 0;
-
- /// Build a file batch reader based on the file path.
- virtual Result<std::unique_ptr<FileBatchReader>> Build(const std::string&
path) const = 0;
};
} // namespace paimon
diff --git a/include/paimon/global_index/global_index_scan.h
b/include/paimon/global_index/global_index_scan.h
index d925e2f..d8ba799 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<RowRangeIndex>& row_range_index) const = 0;
+ /// Creates a `GlobalIndexReader` for a specific field with a specific
index type.
+ /// @param field_name Name of the indexed column.
+ /// @param index_type Name of index type.
+ /// @param row_range_index Optional row range that limits the scan to a
sub-range of row ids.
+ /// If not provided, the entire row range is
considered.
+ /// @return A `Result` that is:
+ /// - Successful with a reader(with global row id) if the index
exists and loads
+ /// correctly;
+ /// - Successful with nullptr if no index was built for the given
field with the given
+ /// index type;
+ /// - Error returns when loading fails (e.g., file corruption, I/O
error,
+ /// unsupported format).
+ virtual Result<std::shared_ptr<GlobalIndexReader>> CreateReader(
+ const std::string& field_name, const std::string& index_type,
+ const std::optional<RowRangeIndex>& row_range_index) const = 0;
+
/// Creates several `GlobalIndexReader`s for a specific field (looked up
by id),
/// @param field_id Field id of the indexed column.
/// @param row_range_index Optional row range that limits the scan to a
sub-range of row ids.
diff --git a/src/paimon/common/data/binary_array_writer.cpp
b/src/paimon/common/data/binary_array_writer.cpp
index 7845311..2a0961f 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<int64_t>(offset) << 32) |
size;
+ const auto offset_and_size =
+ static_cast<int64_t>((static_cast<uint64_t>(offset) << 32) |
static_cast<uint64_t>(size));
segment_.PutValue<int64_t>(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 06a5816..954ba1d 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<int64_t>(offset) << 32) |
size;
+ const auto offset_and_size =
static_cast<int64_t>((static_cast<uint64_t>(offset) << 32) |
+
static_cast<uint64_t>(size));
segment_.PutValue<int64_t>(GetFieldOffset(pos), offset_and_size);
}
diff --git a/src/paimon/common/executor/default_executor_test.cpp
b/src/paimon/common/executor/default_executor_test.cpp
index 26d84f8..2745a69 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<std::atomic<int32_t>> executed_slots(kTotalTaskCount);
+ for (auto& executed_slot : executed_slots) {
+ executed_slot.store(0);
+ }
+ std::vector<std::promise<void>> task_promises(kTotalTaskCount);
+ std::vector<std::future<void>> task_futures;
+ task_futures.reserve(kTotalTaskCount);
+ for (auto& task_promise : task_promises) {
+ task_futures.push_back(task_promise.get_future());
+ }
+ std::atomic<int32_t> ready_submitter_count = 0;
+ std::atomic<int32_t> executed_count = 0;
+ std::promise<void> start_signal;
+ std::shared_future<void> start_future = start_signal.get_future().share();
+ std::vector<std::thread> 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
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 7f4e5cc..a4de3b2 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<std::shared_ptr<GlobalIndexResult>>
UnionGlobalIndexReader::Union(ReaderA
return merged_result;
}
+bool UnionGlobalIndexReader::IsThreadSafe() const {
+ for (const auto& reader : readers_) {
+ if (!reader->IsThreadSafe()) {
+ return false;
+ }
+ }
+ return true;
+}
+
template <typename R>
std::vector<R> UnionGlobalIndexReader::ExecuteAllReaders(
const std::function<R(const std::shared_ptr<GlobalIndexReader>&)>& action)
{
@@ -202,7 +211,7 @@ std::vector<R> 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 526324e..efd42df 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<std::shared_ptr<GlobalIndexResult>> VisitFullTextSearch(
const std::shared_ptr<FullTextSearch>& 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 78efd34..8d55db4 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<int64_t> scored_row_ids_;
std::vector<float> scored_scores_;
bool has_scored_result_ = false;
+ bool thread_safe_ = true;
std::atomic<int32_t> 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<FakeReader>();
- std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader};
+TEST_F(UnionGlobalIndexReaderTest,
TestIsThreadSafeReturnsTrueWhenAllReadersAreSafe) {
+ auto reader1 = std::make_shared<FakeReader>();
+ auto reader2 = std::make_shared<FakeReader>();
+
+ std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
+ UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+
+ ASSERT_TRUE(union_reader.IsThreadSafe());
+}
+
+TEST_F(UnionGlobalIndexReaderTest,
TestIsThreadSafeReturnsFalseWhenAnyReaderIsNotSafe) {
+ auto reader1 = std::make_shared<FakeReader>();
+ auto reader2 = std::make_shared<FakeReader>();
+ reader2->SetThreadSafe(false);
+
+ std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
- // UnionGlobalIndexReader is not thread-safe regardless of inner readers
ASSERT_FALSE(union_reader.IsThreadSafe());
}
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 e9832c1..d9b5821 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<std::unique_ptr<FileBatchReader>> Build(const std::string& path)
const override {
- return Status::Invalid("do not support build reader with path in mock
format");
- }
-
private:
std::shared_ptr<arrow::Array> data_;
std::shared_ptr<arrow::DataType> schema_;
diff --git a/src/paimon/common/utils/file_type_test.cpp
b/src/paimon/common/utils/file_type_test.cpp
index a358b7a..b969813 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/global_index/global_index_scan_impl.cpp
b/src/paimon/core/global_index/global_index_scan_impl.cpp
index 40820a2..83f1239 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<std::shared_ptr<GlobalIndexEvaluator>>
GlobalIndexScanImpl::GetOrCreateIn
Result<std::vector<std::shared_ptr<GlobalIndexReader>>>
GlobalIndexScanImpl::CreateReaders(
int32_t field_id, const std::optional<RowRangeIndex>& row_range_index)
const {
PAIMON_ASSIGN_OR_RAISE(DataField field, table_schema_->GetField(field_id));
- return CreateReaders(field, row_range_index);
+ return CreateReaders(field, /*index_type=*/std::nullopt, row_range_index);
}
Result<std::vector<std::shared_ptr<GlobalIndexReader>>>
GlobalIndexScanImpl::CreateReaders(
const std::string& field_name, const std::optional<RowRangeIndex>&
row_range_index) const {
PAIMON_ASSIGN_OR_RAISE(DataField field,
table_schema_->GetField(field_name));
- return CreateReaders(field, row_range_index);
+ return CreateReaders(field, /*index_type=*/std::nullopt, row_range_index);
+}
+
+Result<std::shared_ptr<GlobalIndexReader>> GlobalIndexScanImpl::CreateReader(
+ const std::string& field_name, const std::string& index_type,
+ const std::optional<RowRangeIndex>& row_range_index) const {
+ PAIMON_ASSIGN_OR_RAISE(DataField field,
table_schema_->GetField(field_name));
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<std::shared_ptr<GlobalIndexReader>> readers,
+ CreateReaders(field, std::optional<std::string>(index_type),
row_range_index));
+ if (readers.empty()) {
+ return std::shared_ptr<GlobalIndexReader>();
+ }
+ if (readers.size() != 1) {
+ return Status::Invalid(
+ fmt::format("invalid global index reader size, expected 1, actual
{}", readers.size()));
+ }
+ return readers[0];
}
Result<std::vector<std::shared_ptr<GlobalIndexReader>>>
GlobalIndexScanImpl::CreateReaders(
- const DataField& field, const std::optional<RowRangeIndex>&
row_range_index) const {
+ const DataField& field, const std::optional<std::string>& index_type,
+ const std::optional<RowRangeIndex>& row_range_index) const {
auto field_iter = index_metas_.find(field.Id());
if (field_iter == index_metas_.end()) {
return std::vector<std::shared_ptr<GlobalIndexReader>>();
}
const auto& index_type_to_metas = field_iter->second;
std::vector<std::shared_ptr<GlobalIndexReader>> readers;
- readers.reserve(index_type_to_metas.size());
- for (const auto& [index_type, range_to_metas] : index_type_to_metas) {
+ readers.reserve(index_type ? 1 : index_type_to_metas.size());
+ for (const auto& [current_index_type, range_to_metas] :
index_type_to_metas) {
+ if (index_type && current_index_type != index_type.value()) {
+ continue;
+ }
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<GlobalIndexer> indexer,
- GlobalIndexerFactory::Get(index_type,
options_.ToMap()));
+ GlobalIndexerFactory::Get(current_index_type,
options_.ToMap()));
if (!indexer) {
continue;
}
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 d5f10d0..a7921d7 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 <map>
#include <memory>
+#include <optional>
#include <string>
#include <vector>
@@ -50,6 +51,10 @@ class GlobalIndexScanImpl : public GlobalIndexScan {
const std::string& field_name,
const std::optional<RowRangeIndex>& row_range_index) const override;
+ Result<std::shared_ptr<GlobalIndexReader>> CreateReader(
+ const std::string& field_name, const std::string& index_type,
+ const std::optional<RowRangeIndex>& row_range_index) const override;
+
Result<std::vector<std::shared_ptr<GlobalIndexReader>>> CreateReaders(
int32_t field_id, const std::optional<RowRangeIndex>& row_range_index)
const override;
@@ -68,7 +73,8 @@ class GlobalIndexScanImpl : public GlobalIndexScan {
Result<std::shared_ptr<GlobalIndexEvaluator>> GetOrCreateIndexEvaluator();
Result<std::vector<std::shared_ptr<GlobalIndexReader>>> CreateReaders(
- const DataField& field, const std::optional<RowRangeIndex>&
row_range_index) const;
+ const DataField& field, const std::optional<std::string>& index_type,
+ const std::optional<RowRangeIndex>& row_range_index) const;
std::vector<GlobalIndexIOMeta> ToGlobalIndexIOMetas(
const std::vector<std::shared_ptr<IndexFileMeta>>& metas) const;
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 1d2932f..a42f39a 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<std::shared_ptr<CommitMessage>>
GlobalIndexWriteTask::WriteIndex(
std::shared_ptr<GlobalIndexFileManager> index_file_manager,
CreateGlobalIndexFileManager(table_path, table_schema, core_options,
pool));
+ // create batch reader
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<BatchReader> 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<GlobalIndexWriter> global_index_writer,
CreateGlobalIndexWriter(index_type, field, index_file_manager,
core_options, pool));
- // create batch reader
- PAIMON_ASSIGN_OR_RAISE(
- std::unique_ptr<BatchReader> 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/src/paimon/core/io/field_mapping_reader.cpp
b/src/paimon/core/io/field_mapping_reader.cpp
index afd5a5e..148b62d 100644
--- a/src/paimon/core/io/field_mapping_reader.cpp
+++ b/src/paimon/core/io/field_mapping_reader.cpp
@@ -220,16 +220,6 @@ Result<std::shared_ptr<arrow::Array>>
FieldMappingReader::GenerateSinglePartitio
scalar = std::make_shared<arrow::Int64Scalar>(value);
break;
}
- case arrow::Type::type::FLOAT: {
- float value =
partition_.GetFloat(partition_info_.value().idx_in_partition[idx]);
- scalar = std::make_shared<arrow::FloatScalar>(value);
- break;
- }
- case arrow::Type::type::DOUBLE: {
- double value =
partition_.GetDouble(partition_info_.value().idx_in_partition[idx]);
- scalar = std::make_shared<arrow::DoubleScalar>(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 52775fc..4b64a0b 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<FieldMapping>(partition_info,
non_part_info,
/*non_exist_field_info=*/std::nullopt);
auto partition = BinaryRowGenerator::GenerateRow(
{false, static_cast<int8_t>(1), static_cast<int16_t>(2),
static_cast<int32_t>(3),
- static_cast<int64_t>(4), static_cast<float>(5.1), 6.21,
std::string("7"),
- std::make_shared<Bytes>("8", pool_.get()), 100},
+ static_cast<int64_t>(4), std::string("5"),
std::make_shared<Bytes>("6", pool_.get()), 100},
pool_.get());
auto mapping_reader = std::make_unique<FieldMappingReader>(
- /*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<arrow::Date32Array*>(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<arrow::BinaryArray*>(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<arrow::StringArray*>(p7_array.get())->Value(0),
- "7");
+ /*idx=*/0, /*batch_size=*/2));
+ ASSERT_EQ(p7_array->length(), 2);
+
ASSERT_EQ(arrow::internal::checked_cast<arrow::Date32Array*>(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<arrow::NumericArray<arrow::DoubleType>*>(p6_array.get())
- ->Value(0),
- static_cast<double>(6.21));
+ /*idx=*/1, /*batch_size=*/2));
+ ASSERT_EQ(p6_array->length(), 2);
+
ASSERT_EQ(arrow::internal::checked_cast<arrow::BinaryArray*>(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<arrow::NumericArray<arrow::FloatType>*>(p5_array.get())
- ->Value(0),
- static_cast<float>(5.1));
+ /*idx=*/2, /*batch_size=*/1));
+ ASSERT_EQ(p5_array->length(), 1);
+
ASSERT_EQ(arrow::internal::checked_cast<arrow::StringArray*>(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<arrow::NumericArray<arrow::Int64Type>*>(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<arrow::NumericArray<arrow::Int32Type>*>(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<arrow::NumericArray<arrow::Int16Type>*>(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<arrow::NumericArray<arrow::Int8Type>*>(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<arrow::BooleanArray*>(p0_array.get())->Value(0),
false);
}
diff --git a/src/paimon/core/operation/abstract_split_read.cpp
b/src/paimon/core/operation/abstract_split_read.cpp
index fb57c0e..30b2215 100644
--- a/src/paimon/core/operation/abstract_split_read.cpp
+++ b/src/paimon/core/operation/abstract_split_read.cpp
@@ -148,10 +148,6 @@ Result<std::unique_ptr<FileBatchReader>>
AbstractSplitRead::CreateFileBatchReade
const std::shared_ptr<DataFileMeta>& 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 5e01768..81c19c6 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<std::unique_ptr<FileBatchReader>>
DataEvolutionSplitRead::ApplyIndexAndDv
std::unique_ptr<FileBatchReader> 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<ApplyBitmapIndexBatchReader>(
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 5790d30..e582ae4 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<std::string> supported_formats = {".orc", ".parquet",
".avro", ".lance"};
+ static std::vector<std::string> 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 4fb94f8..3773104 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 c351254..c06fb05 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<std::unique_ptr<FileBatchReader>>
RawFileSplitRead::ApplyIndexAndDvReader
std::unique_ptr<FileBatchReader> 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<ApplyBitmapIndexBatchReader>(std::move(file_reader),
std::move(actual_selection).value());
diff --git a/src/paimon/core/schema/schema_validation.cpp
b/src/paimon/core/schema/schema_validation.cpp
index 28949a4..17037ed 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<DataField>& fields, const std::vector<std::string>&
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 af73ee6..778f8a3 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<DataField>& fields,
const
std::vector<std::string>& field_names,
const std::string&
error_message_intro);
- static Status ValidateNotContainComplexType(const std::vector<DataField>&
fields,
- const
std::vector<std::string>& field_names);
+ static Status ValidateNotContainSpecificType(const std::vector<DataField>&
fields,
+ const
std::vector<std::string>& 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 0a08efe..9dc7472 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<std::string> primary_keys = {"f0", "f1"};
- std::vector<std::string> partition_keys = {"f1"};
- ASSERT_OK_AND_ASSIGN(
- std::shared_ptr<TableSchema> 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<std::string> primary_keys = {"f0", "f1"};
+ std::vector<std::string> partition_keys = {"f1"};
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<TableSchema> 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<std::string> primary_keys = {"f0", "f1"};
+ std::vector<std::string> partition_keys = {"f1"};
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<TableSchema> 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<TableSchema> 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<std::string> primary_keys = {"f0", "f1"};
- std::vector<std::string> partition_keys = {"f2"};
+ std::vector<std::string> primary_keys = {"f0", "f2"};
+ std::vector<std::string> partition_keys = {"f1"};
std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
{Options::BUCKET_KEY,
"f0"}};
ASSERT_OK_AND_ASSIGN(
@@ -524,7 +539,7 @@ TEST(SchemaValidationTest, ValidateBucket) {
}
{
std::vector<std::string> primary_keys = {};
- std::vector<std::string> partition_keys = {"f2"};
+ std::vector<std::string> partition_keys = {"f1"};
std::map<std::string, std::string> options = {{Options::BUCKET, "2"}};
ASSERT_OK_AND_ASSIGN(
std::shared_ptr<TableSchema> table_schema,
@@ -533,7 +548,7 @@ TEST(SchemaValidationTest, ValidateBucket) {
"You should define a 'bucket-key' for bucketed
append mode");
}
{
- std::vector<std::string> partition_keys = {"f2"};
+ std::vector<std::string> partition_keys = {"f1"};
std::map<std::string, std::string> options =
{{"full-compaction.delta-commits", "2"}};
ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
TableSchema::Create(/*schema_id=*/0, schema,
partition_keys,
@@ -633,11 +648,11 @@ TEST(SchemaValidationTest, ValidateSequenceField) {
std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
{Options::SEQUENCE_FIELD, "f0,f1,f2"}};
ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> 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 7cd0440..c0f2c8f 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<FloatColumnStats*>(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<DoubleColumnStats*>(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 bac30b4..d3a860a 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<int8_t>(1), static_cast<int16_t>(1),
static_cast<int32_t>(1),
- static_cast<int64_t>(1), static_cast<float>(3.0),
static_cast<double>(3.0),
- std::string("abc"), 2025},
+ static_cast<int64_t>(1), std::string("abc"), 2025},
pool.get())));
ASSERT_OK(collector.Collect(BinaryRowGenerator::GenerateRow(
{false, static_cast<int8_t>(2), static_cast<int16_t>(2),
static_cast<int32_t>(2),
- static_cast<int64_t>(2), static_cast<float>(6.0),
static_cast<double>(6.0),
- std::string("bcd"), 2026},
+ static_cast<int64_t>(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<int8_t>(1), static_cast<int16_t>(1),
static_cast<int32_t>(1),
- static_cast<int64_t>(1), static_cast<float>(3.0),
static_cast<double>(3.0),
- std::string("abc"), 2025},
+ static_cast<int64_t>(1), std::string("abc"), 2025},
{true, static_cast<int8_t>(2), static_cast<int16_t>(2),
static_cast<int32_t>(2),
- static_cast<int64_t>(2), static_cast<float>(6.0),
static_cast<double>(6.0),
- std::string("bcd"), 2026},
- std::vector<int64_t>({0, 0, 0, 0, 0, 0, 0, 0, 0}),
GetDefaultPool().get());
+ static_cast<int64_t>(2), std::string("bcd"), 2026},
+ std::vector<int64_t>({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<int64_t> expected,
stats.NullCounts().ToLongArray());
- ASSERT_EQ(expected, std::vector<int64_t>({1l, 1l, 1l, 1l, 1l, 1l, 1l, 1l,
1l, 0l}));
+ ASSERT_EQ(expected, std::vector<int64_t>({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 ed1b056..153932d 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<std::string> partition_keys = {"f0", "f1", "f2", "f3"};
+ std::vector<DataField> 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<arrow::Schema> schema =
DataField::ConvertDataFieldsToArrowSchema(fields);
+ std::vector<std::string> 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<PredicateFilter>(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<DataField> 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<DataField> 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<DataField> 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<DataField> 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<DataField> 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<DataField> 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<DataField> 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);
diff --git a/src/paimon/core/utils/file_store_path_factory.cpp
b/src/paimon/core/utils/file_store_path_factory.cpp
index 97a3bea..d17b223 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<std::string>
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<std::shared_mutex> 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<std::pair<std::string, std::string>> 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<std::shared_mutex> 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<BinaryRow> FileStorePathFactory::ToBinaryRow(
const std::map<std::string, std::string>& partition) const {
+ {
+ std::shared_lock<std::shared_mutex> 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<std::shared_mutex> 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 8e07bce..1890f2b 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 <cstdint>
#include <map>
#include <memory>
+#include <mutex>
+#include <shared_mutex>
#include <string>
#include <unordered_map>
#include <utility>
@@ -99,7 +101,7 @@ class FileStorePathFactory : public
std::enable_shared_from_this<FileStorePathFa
return global_index_external_path_;
}
- /// @note This method is NOT THREAD SAFE.
+ /// Thread-safe for concurrent calls. Cache accesses are protected by
internal locks.
Result<std::string> GetPartitionString(const BinaryRow& partition) const;
std::string NewManifestFile() const {
return PathUtil::JoinPath(
@@ -169,6 +171,7 @@ class FileStorePathFactory : public
std::enable_shared_from_this<FileStorePathFa
mutable std::map<std::map<std::string, std::string>, BinaryRow>
map_to_row_cache_;
mutable std::unordered_map<BinaryRow, std::string> 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 f76f836..41a9635 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 <atomic>
+#include <mutex>
#include <optional>
+#include <thread>
#include <variant>
#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<FileStorePathFactory> 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<std::string>(),
+
/*global_index_external_path=*/std::nullopt,
+ /*index_file_in_data_file_dir=*/false,
mem_pool_));
+
+ const std::vector<std::pair<std::map<std::string, std::string>,
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<bool> has_error = false;
+ std::mutex error_mutex;
+ std::string first_error;
+
+ constexpr int32_t kThreadNum = 8;
+ constexpr int32_t kIterationsPerThread = 500;
+ std::vector<std::thread> 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<std::mutex> 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<std::mutex> 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<std::mutex> 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
diff --git a/src/paimon/format/avro/avro_reader_builder.h
b/src/paimon/format/avro/avro_reader_builder.h
index b4100d1..bd1da1a 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<std::unique_ptr<FileBatchReader>> 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<MemoryPool> pool_;
diff --git a/src/paimon/format/blob/blob_reader_builder.h
b/src/paimon/format/blob/blob_reader_builder.h
index 0bffbf3..85bebe9 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<std::unique_ptr<FileBatchReader>> 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<MemoryPool> pool_;
diff --git a/src/paimon/format/orc/orc_reader_builder.h
b/src/paimon/format/orc/orc_reader_builder.h
index 3fd88d8..e7ef540 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<std::unique_ptr<FileBatchReader>> 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<MemoryPool> pool_;
diff --git a/src/paimon/format/parquet/parquet_reader_builder.h
b/src/paimon/format/parquet/parquet_reader_builder.h
index 2a93696..c0d9931 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<std::unique_ptr<FileBatchReader>> 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<MemoryPool> pool_;
diff --git a/src/paimon/global_index/lumina/lumina_global_index.cpp
b/src/paimon/global_index/lumina/lumina_global_index.cpp
index 5976acd..2d13aa1 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<std::shared_ptr<GlobalIndexReader>>
LuminaGlobalIndex::CreateReader(
class LuminaDataset : public ::lumina::api::Dataset {
public:
LuminaDataset(int64_t element_count, uint32_t dimension,
- const std::vector<std::shared_ptr<arrow::FloatArray>>&
array_vec)
- : element_count_(element_count), dimension_(dimension),
array_vec_(array_vec) {}
+ const std::vector<std::shared_ptr<arrow::FloatArray>>&
array_vec,
+ const std::vector<int64_t>& 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<std::shared_ptr<arrow::FloatArray>> array_vec_;
+ std::vector<int64_t> 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<arrow::ListArray>(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<arrow::FloatArray>(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<arrow::FloatArray>(
+ 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<int64_t>(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<std::vector<GlobalIndexIOMeta>> LuminaIndexWriter::Finish() {
+ if (indexed_count_ == 0) {
+ return std::vector<GlobalIndexIOMeta>();
+ }
::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<std::shared_ptr<arrow::FloatArray>>().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 281f518..572361a 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<LuminaMemoryPool> pool_;
std::string field_name_;
std::shared_ptr<arrow::DataType> arrow_type_;
@@ -102,6 +103,7 @@ class LuminaIndexWriter : public GlobalIndexWriter {
::lumina::api::IOOptions io_options_;
std::map<std::string, std::string> lumina_options_;
std::vector<std::shared_ptr<arrow::FloatArray>> array_vec_;
+ std::vector<int64_t> 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 d0bda5f..0f31477 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<MemoryPool> pool_ = GetDefaultPool();
std::shared_ptr<FileSystem> fs_ = std::make_shared<LocalFileSystem>();
std::map<std::string, std::string> 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<arrow::Array> 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<VectorSearch>(
+ /*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<arrow::Array> 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<VectorSearch>(
+ /*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<arrow::Array> all_null_array =
+ arrow::ipc::internal::json::ArrayFromJSON(data_type_,
+ R"([
+ [null],
+ [null],
+ [null]
+ ])")
+ .ValueOrDie();
+
+ auto global_index = std::make_shared<LuminaGlobalIndex>(options_);
+ auto path_factory = std::make_shared<FakeIndexPathFactory>(test_root);
+ auto file_writer = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<GlobalIndexWriter> 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<int64_t> 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<arrow::Array> 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<VectorSearch>(
+ /*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<arrow::Array> 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<arrow::Array> 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<LuminaGlobalIndex>(options_);
+ auto path_factory = std::make_shared<FakeIndexPathFactory>(test_root);
+ auto file_writer = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<GlobalIndexWriter> 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<int64_t> 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<int64_t> 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<VectorSearch>(
+ /*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/src/paimon/testing/mock/mock_format_reader_builder.h
b/src/paimon/testing/mock/mock_format_reader_builder.h
index 0a7b471..7f8906c 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<MockFileBatchReader>(data_, schema_,
read_batch_size_);
}
- Result<std::unique_ptr<FileBatchReader>> Build(const std::string& path)
const override {
- return Status::Invalid("do not support build reader with path in mock
format");
- }
-
private:
std::shared_ptr<arrow::Array> data_;
std::shared_ptr<arrow::DataType> schema_;
diff --git a/test/inte/append_compaction_inte_test.cpp
b/test/inte/append_compaction_inte_test.cpp
index eda4d6c..88d8115 100644
--- a/test/inte/append_compaction_inte_test.cpp
+++ b/test/inte/append_compaction_inte_test.cpp
@@ -149,9 +149,6 @@ std::vector<std::string>
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 eb3ba04..ca6df4f 100644
--- a/test/inte/blob_table_inte_test.cpp
+++ b/test/inte/blob_table_inte_test.cpp
@@ -250,9 +250,6 @@ std::vector<std::string>
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::StructArray>(
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::StructArray>(
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::StructArray>(
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::StructArray>(
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<std::string> partition_keys = {"f0"};
std::map<std::string, std::string> 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 efcd7b0..4c040fc 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::StructArray>(
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::StructArray>(
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::StructArray>(
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::StructArray>(
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::StructArray>(
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::StructArray>(
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::StructArray>(
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::StructArray>(
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::StructArray>(
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<std::string> 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<std::string>
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 dce39d8..3091dc4 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<std::string, std::string> 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<std::string, std::string> 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<std::string> 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");
@@ -428,7 +463,7 @@ TEST_P(GlobalIndexTest, TestWriteIndexWithPartition) {
#endif
TEST_P(GlobalIndexTest, TestScanIndex) {
- if (file_format_ == "lance" || file_format_ == "avro") {
+ if (file_format_ == "avro") {
return;
}
@@ -597,7 +632,7 @@ TEST_P(GlobalIndexTest, TestScanIndex) {
}
TEST_P(GlobalIndexTest, TestScanIndexWithSpecificSnapshot) {
- if (file_format_ == "lance" || file_format_ == "avro") {
+ if (file_format_ == "avro") {
return;
}
@@ -646,7 +681,7 @@ TEST_P(GlobalIndexTest, TestScanIndexWithSpecificSnapshot) {
}
TEST_P(GlobalIndexTest, TestScanIndexWithSpecificSnapshotWithNoIndex) {
- if (file_format_ == "lance" || file_format_ == "avro") {
+ if (file_format_ == "avro") {
return;
}
@@ -671,7 +706,7 @@ TEST_P(GlobalIndexTest,
TestScanIndexWithSpecificSnapshotWithNoIndex) {
}
TEST_P(GlobalIndexTest, TestScanIndexWithRange) {
- if (file_format_ == "lance" || file_format_ == "avro") {
+ if (file_format_ == "avro") {
return;
}
@@ -707,7 +742,7 @@ TEST_P(GlobalIndexTest, TestScanIndexWithRange) {
}
TEST_P(GlobalIndexTest, TestScanIndexWithPartition) {
- if (file_format_ == "lance" || file_format_ == "avro") {
+ if (file_format_ == "avro") {
return;
}
@@ -762,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();
@@ -991,8 +1026,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 +1041,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<Range>& read_row_ranges,
const std::shared_ptr<arrow::Array>&
expected_array,
@@ -1022,8 +1059,8 @@ TEST_P(GlobalIndexTest,
TestWriteCommitScanReadIndexWithScore) {
result_fields.insert(result_fields.begin(),
SpecialFields::ValueKind().ArrowField());
result_fields.push_back(SpecialFields::IndexScore().ArrowField());
std::map<int64_t, float> 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 +1075,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 +1087,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 +1095,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 +1106,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<GlobalIndexScan> 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<float> query = {1.0f, 1.0f, 1.0f, 1.1f};
+ auto vector_search = std::make_shared<VectorSearch>(
+ "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<BitmapScoredGlobalIndexResult>(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
@@ -1672,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())};
@@ -2633,6 +2692,33 @@ TEST_P(GlobalIndexTest, TestBTreeAndBitmapCoexist) {
ASSERT_OK_AND_ASSIGN(auto index_readers,
global_index_scan->CreateReaders("f0", std::nullopt));
ASSERT_EQ(index_readers.size(), 2u);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexReader> btree_reader,
+ global_index_scan->CreateReader("f0", "btree",
std::nullopt));
+ ASSERT_TRUE(btree_reader);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexResult> btree_result,
+ btree_reader->VisitEqual(Literal(FieldType::STRING,
"Bob", 3)));
+ ASSERT_TRUE(btree_result);
+ ASSERT_EQ(btree_result->ToString(), "{1,2}");
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexResult>
btree_less_than_result,
+
btree_reader->VisitLessThan(Literal(FieldType::STRING, "Emily", 5)));
+ ASSERT_TRUE(btree_less_than_result);
+ ASSERT_EQ(btree_less_than_result->ToString(), "{0,1,2}");
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexReader> bitmap_reader,
+ global_index_scan->CreateReader("f0", "bitmap",
std::nullopt));
+ ASSERT_TRUE(bitmap_reader);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexResult> bitmap_result,
+ bitmap_reader->VisitEqual(Literal(FieldType::STRING,
"Bob", 3)));
+ ASSERT_TRUE(bitmap_result);
+ ASSERT_EQ(bitmap_result->ToString(), "{1,2}");
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexResult>
bitmap_less_than_result,
+
bitmap_reader->VisitLessThan(Literal(FieldType::STRING, "Emily", 5)));
+ ASSERT_FALSE(bitmap_less_than_result);
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexReader> missing_reader,
+ global_index_scan->CreateReader("f0", "lucene",
std::nullopt));
+ ASSERT_FALSE(missing_reader);
+
// Each reader individually should return the same result for Equal("Bob")
for (const auto& index_reader : index_readers) {
ASSERT_OK_AND_ASSIGN(auto result,
@@ -2722,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 =
@@ -3024,10 +3110,6 @@ std::vector<ParamType> 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 6e2bb87..b996220 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<std::string>
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 572d402..ad4c66f 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<std::string, std::string> 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<RecordBatch> 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::ChunkedArray>(
- 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 603e32d..3f9fc62 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<std::pair<std::string, std::string>>
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 236c637..62c61b7 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<DataFileMeta> ReconstructDataFileMeta(
const std::shared_ptr<DataFileMeta>& 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<DataFileMeta>(
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<std::string> 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()),
