feat(build): make TBB dependency configurable

CMakeLists.txt

@@ -58,6 +58,7 @@ option(PAIMON_BUILD_BENCHMARKS "Build benchmarks" OFF)
 option(PAIMON_USE_ASAN "Use Address Sanitizer" OFF)
 option(PAIMON_USE_UBSAN "Use Undefined Behavior Sanitizer" OFF)
 option(PAIMON_USE_CXX11_ABI "Use C++11 ABI" ON)
+option(PAIMON_USE_TBB "Use oneTBB concurrent containers" ON)
 option(PAIMON_ENABLE_AVRO "Whether to enable avro file format" ON)
 option(PAIMON_ENABLE_ORC "Whether to enable orc file format" ON)
 option(PAIMON_ENABLE_JINDO "Whether to enable jindo file system" OFF)
@@ -104,6 +105,9 @@ else()
     endif()
     add_definitions(-D_GLIBCXX_USE_CXX11_ABI=0)
 endif()
+if(PAIMON_USE_TBB)
+    add_definitions(-DPAIMON_USE_TBB)
+endif()
 if(PAIMON_ENABLE_LUMINA)
     add_definitions(-DPAIMON_ENABLE_LUMINA)
 endif()
@@ -360,7 +364,11 @@ include_directories("${CMAKE_SOURCE_DIR}/third_party/roaring_bitmap")
 include_directories("${CMAKE_SOURCE_DIR}/third_party/xxhash")
 
 include_directories(SYSTEM ${ARROW_INCLUDE_DIR})
-include_directories(SYSTEM ${TBB_INCLUDE_DIR})
+set(PAIMON_TBB_LIBS)
+if(PAIMON_USE_TBB)
+    include_directories(SYSTEM ${TBB_INCLUDE_DIR})
+    list(APPEND PAIMON_TBB_LIBS tbb)
+endif()
 
 include_directories(SYSTEM ${GLOG_INCLUDE_DIR})
 add_compile_definitions("GLOG_USE_GLOG_EXPORT")

cmake_modules/DefineOptions.cmake

@@ -173,6 +173,8 @@ if("${CMAKE_SOURCE_DIR}" STREQUAL "${CMAKE_CURRENT_SOURCE_DIR}")
     define_option(PAIMON_DEPENDENCY_USE_SHARED
                   "Prefer shared libraries for system third-party packages" OFF)
 
+    define_option(PAIMON_USE_TBB "Use oneTBB concurrent containers" ON)
+
     define_option_string(Arrow_SOURCE
                          "Dependency source for Apache Arrow; SYSTEM is unsupported"
                          ""

cmake_modules/ThirdpartyToolchain.cmake

@@ -1950,7 +1950,9 @@ resolve_dependency(ZLIB)
 resolve_dependency(LZ4)
 resolve_dependency(Arrow)
 paimon_warn_if_mixed_arrow_dependencies()
-resolve_dependency(TBB)
+if(PAIMON_USE_TBB)
+    resolve_dependency(TBB)
+endif()
 resolve_dependency(glog)
 
 if(PAIMON_ENABLE_AVRO)

docs/source/building.rst

@@ -184,6 +184,10 @@ boolean flags to ``cmake``.
 * ``-DPAIMON_ENABLE_TANTIVY=ON``: Enable the experimental Tantivy full-text index Rust FFI.
 * ``-DPAIMON_ENABLE_REST=ON``: Support for the REST catalog
   (``metastore=rest``), requires the libcurl and OpenSSL development packages.
+* ``-DPAIMON_USE_TBB=ON``: Use oneTBB for the internal concurrent hash map and
+  bounded queue implementations. This is enabled by default. Set it to ``OFF``
+  to use the C++17 standard-library implementations without resolving, building,
+  or linking TBB.
 
 Third-party dependency source
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -217,6 +221,9 @@ require project-specific patches, so their supported source values are
      -Dfmt_ROOT=/opt/fmt \
      -Dzstd_SOURCE=BUNDLED
 
+``TBB_SOURCE`` is only considered when ``PAIMON_USE_TBB=ON``. To build without
+any TBB dependency, configure with ``-DPAIMON_USE_TBB=OFF``.
+
 Use ``PAIMON_PACKAGE_PREFIX`` to provide one common prefix for dependencies
 whose own ``<Package>_ROOT`` variable is not set. Because the patched Arrow and
 ORC dependencies cannot be resolved from the system, a global ``SYSTEM`` build

src/paimon/CMakeLists.txt

@@ -467,7 +467,7 @@ add_paimon_lib(paimon
                ${PAIMON_CORE_SRCS}
                DEPENDENCIES
                arrow
-               tbb
+               ${PAIMON_TBB_LIBS}
                glog
                fmt
                roaring_bitmap
@@ -479,7 +479,7 @@ add_paimon_lib(paimon
                DataSketches
                STATIC_LINK_LIBS
                arrow
-               tbb
+               ${PAIMON_TBB_LIBS}
                glog
                fmt
                roaring_bitmap
@@ -632,6 +632,7 @@ if(PAIMON_BUILD_TESTS)
                     common/utils/arrow/mem_utils_test.cpp
                     common/utils/arrow/status_utils_test.cpp
                     common/utils/concurrent_hash_map_test.cpp
+                    common/utils/concurrent_bounded_queue_test.cpp
                     common/utils/projected_row_test.cpp
                     common/utils/projected_array_test.cpp
                     common/utils/bit_set_test.cpp
@@ -755,6 +756,7 @@ if(PAIMON_BUILD_TESTS)
                     core/index/pk/primary_key_index_definitions_test.cpp
                     core/index/pksorted/pk_sorted_bucket_index_state_test.cpp
                     core/index/index_file_handler_test.cpp
+                    core/io/async_key_value_producer_and_consumer_test.cpp
                     core/io/compact_increment_test.cpp
                     core/io/infer_shredding_file_writer_test.cpp
                     core/io/concat_key_value_record_reader_test.cpp

src/paimon/common/utils/concurrent_bounded_queue.h

@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+#ifdef PAIMON_USE_TBB
+
+#include <cstddef>
+#include <utility>
+
+#include "tbb/concurrent_queue.h"
+
+namespace paimon {
+
+template <typename T>
+class ConcurrentBoundedQueue {
+ public:
+    using size_type = std::ptrdiff_t;
+    using value_type = T;
+    using reference = T&;
+    using const_reference = const T&;
+
+    ConcurrentBoundedQueue() = default;
+    ~ConcurrentBoundedQueue() = default;
+
+    ConcurrentBoundedQueue(const ConcurrentBoundedQueue&) = delete;
+    ConcurrentBoundedQueue& operator=(const ConcurrentBoundedQueue&) = delete;
+    ConcurrentBoundedQueue(ConcurrentBoundedQueue&&) = delete;
+    ConcurrentBoundedQueue& operator=(ConcurrentBoundedQueue&&) = delete;
+
+    void set_capacity(size_type capacity) {
+        queue_.set_capacity(capacity);
+    }
+
+    void push(const T& value) {
+        queue_.push(value);
+    }
+
+    void push(T&& value) {
+        queue_.push(std::move(value));
+    }
+
+    bool try_pop(T& value) {
+        return queue_.try_pop(value);
+    }
+
+    bool empty() const {
+        return queue_.empty();
+    }
+
+ private:
+    tbb::concurrent_bounded_queue<T> queue_;
+};
+
+}  // namespace paimon
+
+#else
+
+#include "paimon/common/utils/std_concurrent_bounded_queue.h"
+
+#endif

src/paimon/common/utils/concurrent_bounded_queue_test.cpp

@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "paimon/common/utils/concurrent_bounded_queue.h"
+
+#include <chrono>
+#include <cstdint>
+#include <future>
+
+#include "gtest/gtest.h"
+
+namespace paimon::test {
+
+TEST(ConcurrentBoundedQueueTest, TestPushAndTryPop) {
+    ConcurrentBoundedQueue<int32_t> queue;
+    queue.set_capacity(2);
+    ASSERT_TRUE(queue.empty());
+
+    queue.push(1);
+    queue.push(2);
+    ASSERT_FALSE(queue.empty());
+
+    int32_t value = 0;
+    ASSERT_TRUE(queue.try_pop(value));
+    ASSERT_EQ(value, 1);
+    ASSERT_TRUE(queue.try_pop(value));
+    ASSERT_EQ(value, 2);
+    ASSERT_FALSE(queue.try_pop(value));
+    ASSERT_TRUE(queue.empty());
+}
+
+TEST(ConcurrentBoundedQueueTest, TestPushWaitsForCapacity) {
+    ConcurrentBoundedQueue<int32_t> queue;
+    queue.set_capacity(1);
+    queue.push(1);
+
+    std::future<void> push_future = std::async(std::launch::async, [&queue]() { queue.push(2); });
+    std::future_status initial_status = push_future.wait_for(std::chrono::milliseconds(50));
+
+    int32_t value = 0;
+    ASSERT_TRUE(queue.try_pop(value));
+    ASSERT_EQ(value, 1);
+    ASSERT_EQ(initial_status, std::future_status::timeout);
+    ASSERT_EQ(push_future.wait_for(std::chrono::seconds(1)), std::future_status::ready);
+    ASSERT_TRUE(queue.try_pop(value));
+    ASSERT_EQ(value, 2);
+}
+
+}  // namespace paimon::test

src/paimon/common/utils/concurrent_hash_map.h

@@ -19,9 +19,12 @@
 
 #pragma once
 
+#ifdef PAIMON_USE_TBB
+
 #include <algorithm>
 #include <cstddef>
 #include <cstdint>
+#include <functional>
 #include <optional>
 #include <string>
 #include <vector>
@@ -30,37 +33,46 @@
 #include "tbb/concurrent_hash_map.h"
 
 namespace paimon {
-template <typename Key, typename T, typename HashCompare = tbb::tbb_hash_compare<Key>>
-class ConcurrentHashMap {
- private:
-    using HashMap = tbb::concurrent_hash_map<Key, T, HashCompare>;
 
+template <typename Key>
+class DefaultHashCompare {
+ public:
+    size_t hash(const Key& key) const {
+        return std::hash<Key>{}(key);
+    }
+
+    bool equal(const Key& lhs, const Key& rhs) const {
+        return lhs == rhs;
+    }
+};
+
+template <typename Key, typename T, typename HashCompare = DefaultHashCompare<Key>>
+class ConcurrentHashMap {
  public:
     ConcurrentHashMap() = default;
     ~ConcurrentHashMap() = default;
 
-    // No copying allowed
     ConcurrentHashMap(const ConcurrentHashMap&) = delete;
     void operator=(const ConcurrentHashMap&) = delete;
     ConcurrentHashMap(ConcurrentHashMap&&) = delete;
     ConcurrentHashMap& operator=(ConcurrentHashMap&&) = delete;
 
     std::optional<T> Find(const Key& key) const {
-        typename HashMap::const_accessor accessor;
+        typename tbb::concurrent_hash_map<Key, T, HashCompare>::const_accessor accessor;
         if (hash_map_.find(accessor, key)) {
             return accessor->second;
         }
         return std::nullopt;
     }
 
     void Insert(const Key& key, const T& value) {
-        typename HashMap::accessor accessor;
+        typename tbb::concurrent_hash_map<Key, T, HashCompare>::accessor accessor;
         hash_map_.insert(accessor, key);
         accessor->second = value;
     }
 
     void Erase(const Key& key) {
-        typename HashMap::accessor accessor;
+        typename tbb::concurrent_hash_map<Key, T, HashCompare>::accessor accessor;
         if (hash_map_.find(accessor, key)) {
             hash_map_.erase(accessor);
         }
@@ -71,7 +83,7 @@ class ConcurrentHashMap {
     }
 
  private:
-    HashMap hash_map_;
+    tbb::concurrent_hash_map<Key, T, HashCompare> hash_map_;
 };
 
 class VectorStringHashCompare {
@@ -89,4 +101,11 @@ class VectorStringHashCompare {
         return a == b;
     }
 };
+
 }  // namespace paimon
+
+#else
+
+#include "paimon/common/utils/std_concurrent_hash_map.h"
+
+#endif

src/paimon/common/utils/concurrent_hash_map_test.cpp

@@ -21,15 +21,34 @@
 
 #include <unistd.h>
 
+#include <atomic>
 #include <cstdlib>
 #include <functional>
+#include <optional>
 #include <thread>
 
 #include "gtest/gtest.h"
 #include "paimon/testing/utils/testharness.h"
 
 namespace paimon::test {
 
+class StatefulHashCompare {
+ public:
+    StatefulHashCompare() : seed_(next_seed_++) {}
+
+    size_t hash(const int32_t& key) const {
+        return std::hash<int32_t>{}(key) + seed_;
+    }
+
+    bool equal(const int32_t& lhs, const int32_t& rhs) const {
+        return lhs == rhs;
+    }
+
+ private:
+    inline static std::atomic<size_t> next_seed_ = 0;
+    size_t seed_;
+};
+
 TEST(ConcurrentHashMapTest, TestSimple) {
     ConcurrentHashMap<int32_t, std::string> hash_map;
     ASSERT_EQ(hash_map.Find(10), std::nullopt);
@@ -71,6 +90,14 @@ TEST(ConcurrentHashMapTest, TestVectorStringHashCompare) {
     ASSERT_EQ(hash_map.Size(), 4);
 }
 
+TEST(ConcurrentHashMapTest, TestStatefulHashCompare) {
+    ConcurrentHashMap<int32_t, std::string, StatefulHashCompare> hash_map;
+    hash_map.Insert(1, "a");
+    ASSERT_EQ(hash_map.Find(1).value(), "a");
+    hash_map.Erase(1);
+    ASSERT_EQ(hash_map.Find(1), std::nullopt);
+}
+
 TEST(ConcurrentHashMapTest, TestMultiThreadInsertAndFindAndDelete) {
     int32_t map_size = 1000;
     auto insert_task = [&](ConcurrentHashMap<int32_t, std::string>& hash_map) {