diff --git a/CMakeLists.txt b/CMakeLists.txt
index 084e6bf03..38c656ca1 100644
--- a/CMakeLists.txt
+++ b/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")
diff --git a/cmake_modules/DefineOptions.cmake b/cmake_modules/DefineOptions.cmake
index 21f653155..d865fd9b2 100644
--- a/cmake_modules/DefineOptions.cmake
+++ b/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"
                          ""
diff --git a/cmake_modules/ThirdpartyToolchain.cmake b/cmake_modules/ThirdpartyToolchain.cmake
index 776519104..78482f34d 100644
--- a/cmake_modules/ThirdpartyToolchain.cmake
+++ b/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)
diff --git a/docs/source/building.rst b/docs/source/building.rst
index 32c7e67cd..88dee2d75 100644
--- a/docs/source/building.rst
+++ b/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
diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
index 3de2b667e..bb47f23c0 100644
--- a/src/paimon/CMakeLists.txt
+++ b/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
diff --git a/src/paimon/common/utils/concurrent_bounded_queue.h b/src/paimon/common/utils/concurrent_bounded_queue.h
new file mode 100644
index 000000000..642d5b227
--- /dev/null
+++ b/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
diff --git a/src/paimon/common/utils/concurrent_bounded_queue_test.cpp b/src/paimon/common/utils/concurrent_bounded_queue_test.cpp
new file mode 100644
index 000000000..ed2479aa6
--- /dev/null
+++ b/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
diff --git a/src/paimon/common/utils/concurrent_hash_map.h b/src/paimon/common/utils/concurrent_hash_map.h
index 79a21f098..cedb49111 100644
--- a/src/paimon/common/utils/concurrent_hash_map.h
+++ b/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,23 +33,32 @@
 #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;
         }
@@ -54,13 +66,13 @@ class ConcurrentHashMap {
     }
 
     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
diff --git a/src/paimon/common/utils/concurrent_hash_map_test.cpp b/src/paimon/common/utils/concurrent_hash_map_test.cpp
index d3462afb5..249ddafeb 100644
--- a/src/paimon/common/utils/concurrent_hash_map_test.cpp
+++ b/src/paimon/common/utils/concurrent_hash_map_test.cpp
@@ -21,8 +21,10 @@
 
 #include <unistd.h>
 
+#include <atomic>
 #include <cstdlib>
 #include <functional>
+#include <optional>
 #include <thread>
 
 #include "gtest/gtest.h"
@@ -30,6 +32,23 @@
 
 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) {
diff --git a/src/paimon/common/utils/std_concurrent_bounded_queue.h b/src/paimon/common/utils/std_concurrent_bounded_queue.h
new file mode 100644
index 000000000..5a009cff4
--- /dev/null
+++ b/src/paimon/common/utils/std_concurrent_bounded_queue.h
@@ -0,0 +1,92 @@
+/*
+ * 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
+
+#include <condition_variable>
+#include <cstddef>
+#include <limits>
+#include <mutex>
+#include <queue>
+#include <utility>
+
+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) {
+        {
+            std::unique_lock<std::mutex> lock(mutex_);
+            capacity_ =
+                capacity < 0 ? std::numeric_limits<size_t>::max() : static_cast<size_t>(capacity);
+        }
+        capacity_available_.notify_all();
+    }
+
+    void push(const T& value) {
+        T copied_value = value;
+        push(std::move(copied_value));
+    }
+
+    void push(T&& value) {
+        std::unique_lock<std::mutex> lock(mutex_);
+        capacity_available_.wait(lock, [this]() { return queue_.size() < capacity_; });
+        queue_.push(std::move(value));
+    }
+
+    bool try_pop(T& value) {
+        {
+            std::unique_lock<std::mutex> lock(mutex_);
+            if (queue_.empty()) {
+                return false;
+            }
+            value = std::move(queue_.front());
+            queue_.pop();
+        }
+        capacity_available_.notify_one();
+        return true;
+    }
+
+    bool empty() const {
+        std::unique_lock<std::mutex> lock(mutex_);
+        return queue_.empty();
+    }
+
+ private:
+    std::queue<T> queue_;
+    size_t capacity_ = std::numeric_limits<size_t>::max();
+    mutable std::mutex mutex_;
+    std::condition_variable capacity_available_;
+};
+
+}  // namespace paimon
diff --git a/src/paimon/common/utils/std_concurrent_hash_map.h b/src/paimon/common/utils/std_concurrent_hash_map.h
new file mode 100644
index 000000000..bd91d7f22
--- /dev/null
+++ b/src/paimon/common/utils/std_concurrent_hash_map.h
@@ -0,0 +1,146 @@
+/*
+ * 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
+
+#include <algorithm>
+#include <cstddef>
+#include <cstdint>
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <shared_mutex>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include "paimon/common/utils/murmurhash_utils.h"
+
+namespace paimon {
+
+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;
+    }
+};
+
+namespace detail {
+
+template <typename Key, typename HashCompare>
+class HashCompareHasher {
+ public:
+    explicit HashCompareHasher(std::shared_ptr<const HashCompare> hash_compare)
+        : hash_compare_(std::move(hash_compare)) {}
+
+    size_t operator()(const Key& key) const {
+        return hash_compare_->hash(key);
+    }
+
+ private:
+    std::shared_ptr<const HashCompare> hash_compare_;
+};
+
+template <typename Key, typename HashCompare>
+class HashCompareEqual {
+ public:
+    explicit HashCompareEqual(std::shared_ptr<const HashCompare> hash_compare)
+        : hash_compare_(std::move(hash_compare)) {}
+
+    bool operator()(const Key& lhs, const Key& rhs) const {
+        return hash_compare_->equal(lhs, rhs);
+    }
+
+ private:
+    std::shared_ptr<const HashCompare> hash_compare_;
+};
+
+}  // namespace detail
+
+template <typename Key, typename T, typename HashCompare = DefaultHashCompare<Key>>
+class ConcurrentHashMap {
+ private:
+    using HashMap = std::unordered_map<Key, T, detail::HashCompareHasher<Key, HashCompare>,
+                                       detail::HashCompareEqual<Key, HashCompare>>;
+
+ public:
+    ConcurrentHashMap()
+        : hash_compare_(std::make_shared<HashCompare>()),
+          hash_map_(0, detail::HashCompareHasher<Key, HashCompare>(hash_compare_),
+                    detail::HashCompareEqual<Key, HashCompare>(hash_compare_)) {}
+    ~ConcurrentHashMap() = default;
+
+    ConcurrentHashMap(const ConcurrentHashMap&) = delete;
+    void operator=(const ConcurrentHashMap&) = delete;
+    ConcurrentHashMap(ConcurrentHashMap&&) = delete;
+    ConcurrentHashMap& operator=(ConcurrentHashMap&&) = delete;
+
+    std::optional<T> Find(const Key& key) const {
+        std::shared_lock<std::shared_mutex> lock(mutex_);
+        typename HashMap::const_iterator iter = hash_map_.find(key);
+        if (iter != hash_map_.end()) {
+            return iter->second;
+        }
+        return std::nullopt;
+    }
+
+    void Insert(const Key& key, const T& value) {
+        std::unique_lock<std::shared_mutex> lock(mutex_);
+        hash_map_.insert_or_assign(key, value);
+    }
+
+    void Erase(const Key& key) {
+        std::unique_lock<std::shared_mutex> lock(mutex_);
+        hash_map_.erase(key);
+    }
+
+    size_t Size() const {
+        std::shared_lock<std::shared_mutex> lock(mutex_);
+        return hash_map_.size();
+    }
+
+ private:
+    std::shared_ptr<const HashCompare> hash_compare_;
+    HashMap hash_map_;
+    mutable std::shared_mutex mutex_;
+};
+
+class VectorStringHashCompare {
+ public:
+    size_t hash(const std::vector<std::string>& key) const {
+        int32_t ret = MurmurHashUtils::DEFAULT_SEED;
+        for (const auto& s : key) {
+            ret = MurmurHashUtils::HashUnsafeBytes(reinterpret_cast<const void*>(s.data()),
+                                                   /*offset=*/0, s.size(), ret);
+        }
+        return ret;
+    }
+
+    bool equal(const std::vector<std::string>& a, const std::vector<std::string>& b) const {
+        return a == b;
+    }
+};
+
+}  // namespace paimon
diff --git a/src/paimon/core/io/async_key_value_producer_and_consumer.cpp b/src/paimon/core/io/async_key_value_producer_and_consumer.cpp
index 1792b43cd..96e74f839 100644
--- a/src/paimon/core/io/async_key_value_producer_and_consumer.cpp
+++ b/src/paimon/core/io/async_key_value_producer_and_consumer.cpp
@@ -21,6 +21,7 @@
 #include <unistd.h>
 
 #include <chrono>
+#include <thread>
 #include <type_traits>
 
 #include "arrow/c/abi.h"
@@ -143,14 +144,35 @@ template <typename T, typename R>
 void AsyncKeyValueProducerAndConsumer<T, R>::CleanUp() {
     consume_finished_ = true;
     next_batch_finished_ = true;
+
+    // A producer or consumer may already be blocked in a bounded queue push. Keep draining both
+    // queues until every worker has stopped producing, otherwise a single drain can miss an item
+    // pushed by a worker after it is woken up.
+    while (true) {
+        bool producer_finished =
+            !producer_future_.valid() ||
+            producer_future_.wait_for(std::chrono::microseconds(0)) == std::future_status::ready;
+        bool consumers_finished = true;
+        for (const std::unique_ptr<AsyncKeyValueConsumer<T, R>>& consumer : consumers_) {
+            if (!consumer->IsFinished()) {
+                consumers_finished = false;
+                break;
+            }
+        }
+        if (producer_finished && consumers_finished) {
+            break;
+        }
+        CleanUpQueue();
+        std::this_thread::yield();
+    }
     CleanUpQueue();
+
     if (producer_future_.valid()) {
         [[maybe_unused]] Status status = producer_future_.get();
     }
     for (auto& consumer : consumers_) {
         consumer->CleanUp();
     }
-    CleanUpQueue();
 }
 
 template <typename T, typename R>
@@ -180,8 +202,8 @@ template <typename T, typename R>
 AsyncKeyValueConsumer<T, R>::AsyncKeyValueConsumer(
     std::unique_ptr<RowToArrowArrayConverter<T, R>>&& key_value_consumer,
     std::atomic<bool>& consume_finished, std::atomic<int32_t>& consumer_finished_count,
-    tbb::concurrent_bounded_queue<std::vector<KeyValue>>& kv_queue,
-    tbb::concurrent_bounded_queue<R>& result_queue)
+    ConcurrentBoundedQueue<std::vector<KeyValue>>& kv_queue,
+    ConcurrentBoundedQueue<R>& result_queue)
     : key_value_consumer_(std::move(key_value_consumer)),
       consume_finished_(consume_finished),
       consumer_finished_count_(consumer_finished_count),
@@ -202,6 +224,12 @@ Status AsyncKeyValueConsumer<T, R>::GetStatus() const {
     return Status::OK();
 }
 
+template <typename T, typename R>
+bool AsyncKeyValueConsumer<T, R>::IsFinished() const {
+    return !consumer_future_.valid() ||
+           consumer_future_.wait_for(std::chrono::microseconds(0)) == std::future_status::ready;
+}
+
 template <typename T, typename R>
 Status AsyncKeyValueConsumer<T, R>::ConsumeLoop() {
     while (!consume_finished_) {
diff --git a/src/paimon/core/io/async_key_value_producer_and_consumer.h b/src/paimon/core/io/async_key_value_producer_and_consumer.h
index af8bbed68..bdfa8f0a0 100644
--- a/src/paimon/core/io/async_key_value_producer_and_consumer.h
+++ b/src/paimon/core/io/async_key_value_producer_and_consumer.h
@@ -27,12 +27,12 @@
 #include <vector>
 
 #include "arrow/api.h"
+#include "paimon/common/utils/concurrent_bounded_queue.h"
 #include "paimon/core/io/row_to_arrow_array_converter.h"
 #include "paimon/core/key_value.h"
 #include "paimon/core/mergetree/compact/sort_merge_reader.h"
 #include "paimon/result.h"
 #include "paimon/status.h"
-#include "tbb/concurrent_queue.h"
 
 namespace paimon {
 template <typename T, typename R>
@@ -94,8 +94,8 @@ class AsyncKeyValueProducerAndConsumer {
     std::shared_future<Status> producer_future_;
     std::vector<std::unique_ptr<AsyncKeyValueConsumer<T, R>>> consumers_;
     std::atomic<int32_t> consumer_finished_count_ = 0;
-    tbb::concurrent_bounded_queue<std::vector<KeyValue>> kv_queue_;
-    tbb::concurrent_bounded_queue<R> result_queue_;
+    ConcurrentBoundedQueue<std::vector<KeyValue>> kv_queue_;
+    ConcurrentBoundedQueue<R> result_queue_;
 };
 
 template <typename T, typename R>
@@ -104,8 +104,8 @@ class AsyncKeyValueConsumer {
     AsyncKeyValueConsumer(std::unique_ptr<RowToArrowArrayConverter<T, R>>&& key_value_consumer,
                           std::atomic<bool>& consume_finished,
                           std::atomic<int32_t>& consumer_finished_count,
-                          tbb::concurrent_bounded_queue<std::vector<KeyValue>>& kv_queue,
-                          tbb::concurrent_bounded_queue<R>& result_queue);
+                          ConcurrentBoundedQueue<std::vector<KeyValue>>& kv_queue,
+                          ConcurrentBoundedQueue<R>& result_queue);
 
     ~AsyncKeyValueConsumer() {
         CleanUp();
@@ -115,6 +115,9 @@ class AsyncKeyValueConsumer {
     void CleanUp();
 
  private:
+    friend class AsyncKeyValueProducerAndConsumer<T, R>;
+
+    bool IsFinished() const;
     Status ConsumeLoop();
 
  private:
@@ -122,8 +125,8 @@ class AsyncKeyValueConsumer {
     std::shared_future<Status> consumer_future_;
     std::atomic<bool>& consume_finished_;
     std::atomic<int32_t>& consumer_finished_count_;
-    tbb::concurrent_bounded_queue<std::vector<KeyValue>>& kv_queue_;
-    tbb::concurrent_bounded_queue<R>& result_queue_;
+    ConcurrentBoundedQueue<std::vector<KeyValue>>& kv_queue_;
+    ConcurrentBoundedQueue<R>& result_queue_;
 };
 
 }  // namespace paimon
diff --git a/src/paimon/core/io/async_key_value_producer_and_consumer_test.cpp b/src/paimon/core/io/async_key_value_producer_and_consumer_test.cpp
new file mode 100644
index 000000000..fc2f7d340
--- /dev/null
+++ b/src/paimon/core/io/async_key_value_producer_and_consumer_test.cpp
@@ -0,0 +1,132 @@
+/*
+ * 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/core/io/async_key_value_producer_and_consumer.h"
+
+#include <condition_variable>
+#include <cstddef>
+#include <memory>
+#include <mutex>
+#include <utility>
+#include <vector>
+
+#include "gtest/gtest.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+namespace {
+
+class ConsumerBarrier {
+ public:
+    explicit ConsumerBarrier(size_t consumer_count) : consumer_count_(consumer_count) {}
+
+    void Wait() {
+        std::unique_lock<std::mutex> lock(mutex_);
+        ++arrived_count_;
+        if (arrived_count_ == consumer_count_) {
+            condition_.notify_all();
+        } else {
+            condition_.wait(lock, [this]() { return arrived_count_ >= consumer_count_; });
+        }
+    }
+
+ private:
+    const size_t consumer_count_;
+    size_t arrived_count_ = 0;
+    std::mutex mutex_;
+    std::condition_variable condition_;
+};
+
+class TestKeyValueConsumer : public RowToArrowArrayConverter<KeyValue, KeyValueBatch> {
+ public:
+    explicit TestKeyValueConsumer(std::shared_ptr<ConsumerBarrier> barrier)
+        : RowToArrowArrayConverter(/*reserve_count=*/0, std::vector<AppendValueFunc>(), nullptr,
+                                   nullptr),
+          barrier_(std::move(barrier)) {}
+
+    Result<KeyValueBatch> NextBatch(const std::vector<KeyValue>&) override {
+        barrier_->Wait();
+        return KeyValueBatch();
+    }
+
+ private:
+    std::shared_ptr<ConsumerBarrier> barrier_;
+};
+
+class TestSortMergeReader : public SortMergeReader {
+ public:
+    explicit TestSortMergeReader(size_t row_count) : rows_(row_count) {}
+
+    class Iterator : public SortMergeReader::Iterator {
+     public:
+        explicit Iterator(TestSortMergeReader* reader) : reader_(reader) {}
+
+        Result<bool> HasNext() override {
+            return reader_->next_row_ < reader_->rows_.size();
+        }
+
+        KeyValue&& Next() override {
+            return std::move(reader_->rows_[reader_->next_row_++]);
+        }
+
+     private:
+        TestSortMergeReader* reader_;
+    };
+
+    Result<std::unique_ptr<SortMergeReader::Iterator>> NextBatch() override {
+        if (iterator_created_) {
+            return std::unique_ptr<SortMergeReader::Iterator>();
+        }
+        iterator_created_ = true;
+        return std::make_unique<Iterator>(this);
+    }
+
+    void Close() override {}
+
+    std::shared_ptr<Metrics> GetReaderMetrics() const override {
+        return nullptr;
+    }
+
+ private:
+    std::vector<KeyValue> rows_;
+    size_t next_row_ = 0;
+    bool iterator_created_ = false;
+};
+
+}  // namespace
+
+TEST(AsyncKeyValueProducerAndConsumerTest, TestEarlyCloseWithBlockedConsumers) {
+    constexpr int32_t kConsumerCount = 32;
+    std::shared_ptr<ConsumerBarrier> barrier = std::make_shared<ConsumerBarrier>(kConsumerCount);
+    auto create_consumer =
+        [barrier]() -> Result<std::unique_ptr<RowToArrowArrayConverter<KeyValue, KeyValueBatch>>> {
+        std::unique_ptr<RowToArrowArrayConverter<KeyValue, KeyValueBatch>> consumer =
+            std::make_unique<TestKeyValueConsumer>(barrier);
+        return consumer;
+    };
+
+    AsyncKeyValueProducerAndConsumer<KeyValue, KeyValueBatch> producer_and_consumer(
+        std::make_unique<TestSortMergeReader>(kConsumerCount * 4), create_consumer,
+        /*batch_size=*/1, kConsumerCount, /*pool=*/nullptr);
+
+    ASSERT_OK_AND_ASSIGN(KeyValueBatch first_batch, producer_and_consumer.NextBatch());
+    ASSERT_FALSE(first_batch.batch);
+    producer_and_consumer.Close();
+}
+
+}  // namespace paimon::test
diff --git a/src/paimon/format/avro/CMakeLists.txt b/src/paimon/format/avro/CMakeLists.txt
index 1c4256741..9b61c0ecc 100644
--- a/src/paimon/format/avro/CMakeLists.txt
+++ b/src/paimon/format/avro/CMakeLists.txt
@@ -39,7 +39,7 @@ if(PAIMON_ENABLE_AVRO)
                    glog
                    fmt
                    avro
-                   tbb
+                   ${PAIMON_TBB_LIBS}
                    Threads::Threads
                    SHARED_LINK_LIBS
                    paimon_shared
diff --git a/src/paimon/format/orc/CMakeLists.txt b/src/paimon/format/orc/CMakeLists.txt
index d86750ea7..7641686ef 100644
--- a/src/paimon/format/orc/CMakeLists.txt
+++ b/src/paimon/format/orc/CMakeLists.txt
@@ -39,7 +39,7 @@ if(PAIMON_ENABLE_ORC)
                    glog
                    fmt
                    orc::orc
-                   tbb
+                   ${PAIMON_TBB_LIBS}
                    Threads::Threads
                    SHARED_LINK_LIBS
                    paimon_shared