feat: Add global config API and optimize Parquet read thread conf (#68)

* feat: Add global config API and optimize Parquet read thread configuration

- Add global_config.h/cpp for managing global configurations
- Change Parquet read thread configuration from thread count to boolean switch
- Update related test cases to adapt to the new configuration

* fix review

---------

Co-authored-by: lxy <38709059+lxy-9602@users.noreply.github.com>
Co-authored-by: Yonghao Fang <yonghao.fyh@alibaba-inc.com>

Author: 3 people

include/paimon/api.h

@@ -24,6 +24,7 @@
 #include "paimon/file_store_commit.h"        // IWYU pragma: export
 #include "paimon/file_store_write.h"         // IWYU pragma: export
 #include "paimon/fs/file_system_factory.h"   // IWYU pragma: export
+#include "paimon/global_config.h"            // IWYU pragma: export
 #include "paimon/memory/memory_pool.h"       // IWYU pragma: export
 #include "paimon/predicate/predicate.h"      // IWYU pragma: export
 #include "paimon/read_context.h"             // IWYU pragma: export

include/paimon/global_config.h

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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 "paimon/result.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+
+/// Get the capacity of the arrow's global thread pool
+/// This is a simple wrapper of arrow::GetCpuThreadPoolCapacity()
+///
+/// Return the number of worker threads in the thread pool to which
+/// Arrow dispatches various CPU-bound tasks.  This is an ideal number,
+/// not necessarily the exact number of threads at a given point in time.
+///
+/// You can change this number using SetArrowCpuThreadPoolCapacity().
+PAIMON_EXPORT int GetArrowCpuThreadPoolCapacity();
+
+/// Set the capacity of the arrow's global thread pool
+/// This is a simple wrapper of arrow::SetCpuThreadPoolCapacity()
+///
+/// Set the number of worker threads in the thread pool to which
+/// Arrow dispatches various CPU-bound tasks.
+///
+/// The current number is returned by GetArrowCpuThreadPoolCapacity().
+/// Currently, this capacity will significantly affect the performance
+/// of parquet file batch read.
+PAIMON_EXPORT Status SetArrowCpuThreadPoolCapacity(int threads);
+
+}  // namespace paimon

src/paimon/CMakeLists.txt

@@ -46,6 +46,7 @@ set(PAIMON_COMMON_SRCS
     common/fs/file_system.cpp
     common/fs/resolving_file_system.cpp
     common/fs/file_system_factory.cpp
+    common/global_config.cpp
     common/global_index/complete_index_score_batch_reader.cpp
     common/global_index/bitmap_vector_search_global_index_result.cpp
     common/global_index/bitmap_global_index_result.cpp

src/paimon/common/global_config.cpp

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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/global_config.h"
+
+#include "arrow/util/thread_pool.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+
+namespace paimon {
+PAIMON_EXPORT int GetArrowCpuThreadPoolCapacity() {
+    return arrow::GetCpuThreadPoolCapacity();
+}
+
+PAIMON_EXPORT Status SetArrowCpuThreadPoolCapacity(int threads) {
+    PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::SetCpuThreadPoolCapacity(threads));
+    return Status::OK();
+}
+}  // namespace paimon

src/paimon/format/parquet/parquet_file_batch_reader.cpp

@@ -33,7 +33,6 @@
 #include "arrow/record_batch.h"
 #include "arrow/type.h"
 #include "arrow/util/range.h"
-#include "arrow/util/thread_pool.h"
 #include "fmt/format.h"
 #include "paimon/common/metrics/metrics_impl.h"
 #include "paimon/common/utils/arrow/status_utils.h"
@@ -265,21 +264,15 @@ Result<::parquet::ReaderProperties> ParquetFileBatchReader::CreateReaderProperti
 Result<::parquet::ArrowReaderProperties> ParquetFileBatchReader::CreateArrowReaderProperties(
     const std::shared_ptr<arrow::MemoryPool>& pool,
     const std::map<std::string, std::string>& options, int32_t batch_size) {
-    PAIMON_ASSIGN_OR_RAISE(
-        uint32_t executor_thread_count,
-        OptionsUtils::GetValueFromMap<uint32_t>(options, PARQUET_READ_EXECUTOR_THREAD_COUNT,
-                                                DEFAULT_PARQUET_READ_EXECUTOR_THREAD_COUNT));
+    PAIMON_ASSIGN_OR_RAISE(bool use_threads,
+                           OptionsUtils::GetValueFromMap<bool>(options, PARQUET_READ_USE_THREADS,
+                                                               DEFAULT_PARQUET_READ_USE_THREADS));
 
     ::parquet::ArrowReaderProperties arrow_reader_props;
     // TODO(jinli.zjw): set more ArrowReaderProperties (compare with java)
     arrow_reader_props.set_pre_buffer(true);
     arrow_reader_props.set_batch_size(static_cast<int64_t>(batch_size));
-    if (executor_thread_count != 0) {
-        PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::SetCpuThreadPoolCapacity(executor_thread_count));
-        arrow_reader_props.set_use_threads(true);
-    } else {
-        arrow_reader_props.set_use_threads(false);
-    }
+    arrow_reader_props.set_use_threads(use_threads);
     PAIMON_ASSIGN_OR_RAISE(bool cache_lazy, OptionsUtils::GetValueFromMap<bool>(
                                                 options, PARQUET_READ_CACHE_OPTION_LAZY, false));
     PAIMON_ASSIGN_OR_RAISE(

src/paimon/format/parquet/parquet_file_batch_reader_test.cpp

@@ -41,6 +41,7 @@
 #include "paimon/format/parquet/parquet_input_stream_impl.h"
 #include "paimon/fs/file_system.h"
 #include "paimon/fs/local/local_file_system.h"
+#include "paimon/global_config.h"
 #include "paimon/memory/memory_pool.h"
 #include "paimon/predicate/literal.h"
 #include "paimon/predicate/predicate_builder.h"
@@ -394,25 +395,27 @@ TEST_F(ParquetFileBatchReaderTest, TestCreateArrowReaderProperties) {
         ASSERT_EQ(arrow_reader_properties.pre_buffer(), true);
         ASSERT_EQ(arrow_reader_properties.batch_size(), 1024);
         ASSERT_EQ(arrow_reader_properties.use_threads(), true);
-        ASSERT_EQ(arrow::GetCpuThreadPoolCapacity(), 3);
         ASSERT_EQ(arrow_reader_properties.cache_options(), arrow::io::CacheOptions::Defaults());
     }
     {
-        std::map<std::string, std::string> options = {{PARQUET_READ_EXECUTOR_THREAD_COUNT, "0"}};
+        std::map<std::string, std::string> options = {{PARQUET_READ_USE_THREADS, "false"}};
         int32_t batch_size = 1024;
         ASSERT_OK_AND_ASSIGN(
             auto arrow_reader_properties,
             ParquetFileBatchReader::CreateArrowReaderProperties(pool_, options, batch_size));
         ASSERT_EQ(arrow_reader_properties.use_threads(), false);
     }
     {
-        std::map<std::string, std::string> options = {{PARQUET_READ_EXECUTOR_THREAD_COUNT, "6"}};
+        int original_capacity = GetArrowCpuThreadPoolCapacity();
+        ASSERT_OK(SetArrowCpuThreadPoolCapacity(6));
+        std::map<std::string, std::string> options = {{PARQUET_READ_USE_THREADS, "true"}};
         int32_t batch_size = 1024;
         ASSERT_OK_AND_ASSIGN(
             auto arrow_reader_properties,
             ParquetFileBatchReader::CreateArrowReaderProperties(pool_, options, batch_size));
         ASSERT_EQ(arrow_reader_properties.use_threads(), true);
-        ASSERT_EQ(arrow::GetCpuThreadPoolCapacity(), 6);
+        ASSERT_EQ(GetArrowCpuThreadPoolCapacity(), 6);
+        ASSERT_OK(SetArrowCpuThreadPoolCapacity(original_capacity));
     }
 }
 

src/paimon/format/parquet/parquet_format_defs.h

@@ -36,9 +36,8 @@ static inline const char PARQUET_COMPRESSION_CODEC_ZLIB_LEVEL[] = "zlib.compress
 static inline const char PARQUET_COMPRESSION_CODEC_BROTLI_LEVEL[] = "compression.brotli.quality";
 
 // read
-static inline const char PARQUET_READ_EXECUTOR_THREAD_COUNT[] =
-    "parquet.read.executor.thread-count";
-static constexpr uint32_t DEFAULT_PARQUET_READ_EXECUTOR_THREAD_COUNT = 3;
+static inline const char PARQUET_READ_USE_THREADS[] = "parquet.read.use-threads";
+static inline const bool DEFAULT_PARQUET_READ_USE_THREADS = true;
 static inline const char PARQUET_READ_CACHE_OPTION_LAZY[] = "parquet.read.cache-option.lazy";
 static inline const char PARQUET_READ_CACHE_OPTION_PREFETCH_LIMIT[] =
     "parquet.read.cache-option.prefetch-limit";