fix: prevent zero-thread DefaultExecutor and validate FileSystem/SchemeMap conflict

1. DefaultExecutor: guard against thread_count==0 by falling back to 1
2. GetGlobalDefaultExecutor: handle hardware_concurrency()==0 case
3. ReadContextBuilder::Finish: validate prefetch_max_parallel_num > 0
4. ReadContextBuilder::Finish: reject simultaneous WithFileSystem() and
   WithFileSystemSchemeToIdentifierMap() as documented in API contract
5. GlobalIndexScanImpl::Create: guard against global-index.thread-num <= 0
6. Update read_context_test to match new validation rules

Author: zjw1111

src/paimon/common/executor/executor.cpp

@@ -49,7 +49,8 @@ class DefaultExecutor : public Executor {
     int32_t active_tasks_ = 0;
 };
 
-DefaultExecutor::DefaultExecutor(uint32_t thread_count) : thread_count_(thread_count) {
+DefaultExecutor::DefaultExecutor(uint32_t thread_count)
+    : thread_count_(thread_count == 0 ? 1 : thread_count) {
     for (uint32_t i = 0; i < thread_count_; ++i) {
         workers_.emplace_back(&DefaultExecutor::WorkerThread, this);
     }
@@ -130,7 +131,7 @@ void DefaultExecutor::WorkerThread() {
 PAIMON_EXPORT std::shared_ptr<Executor> GetGlobalDefaultExecutor() {
     static uint32_t all_cores = std::thread::hardware_concurrency();
     static std::shared_ptr<Executor> internal =
-        std::make_shared<DefaultExecutor>(/*thread_count=*/all_cores);
+        std::make_shared<DefaultExecutor>(/*thread_count=*/all_cores > 0 ? all_cores : 1);
     return internal;
 }
 

src/paimon/core/global_index/global_index_scan_impl.cpp

@@ -107,6 +107,9 @@ Result<std::unique_ptr<GlobalIndexScanImpl>> GlobalIndexScanImpl::Create(
             uint32_t cpu_count = std::thread::hardware_concurrency();
             thread_num = cpu_count > 0 ? static_cast<int32_t>(cpu_count) : 1;
         }
+        if (thread_num.value() <= 0) {
+            thread_num = 1;
+        }
         final_executor = CreateDefaultExecutor(static_cast<uint32_t>(thread_num.value()));
     }
     return std::unique_ptr<GlobalIndexScanImpl>(new GlobalIndexScanImpl(

src/paimon/core/operation/read_context.cpp

@@ -222,6 +222,9 @@ Result<std::unique_ptr<ReadContext>> ReadContextBuilder::Finish() {
     if (impl_->path_.empty()) {
         return Status::Invalid("cannot read with empty table path");
     }
+    if (impl_->enable_prefetch_ && impl_->prefetch_max_parallel_num_ == 0) {
+        return Status::Invalid("prefetch max parallel num should be greater than 0");
+    }
     if (impl_->enable_prefetch_ && impl_->prefetch_batch_count_ <= 0) {
         return Status::Invalid("prefetch batch count should be greater than 0");
     }
@@ -230,6 +233,10 @@ Result<std::unique_ptr<ReadContext>> ReadContextBuilder::Finish() {
         return Status::Invalid(
             "prefetch batch count should be greater than or equal to prefetch max parallel num");
     }
+    if (impl_->specific_file_system_ && !impl_->fs_scheme_to_identifier_map_.empty()) {
+        return Status::Invalid(
+            "WithFileSystem() and WithFileSystemSchemeToIdentifierMap() cannot be used together");
+    }
     if (!impl_->executor_) {
         // If the user do not set executor, create default executor by prefetch batch count
         uint32_t thread_count = impl_->enable_prefetch_ ? impl_->prefetch_max_parallel_num_ : 1;

src/paimon/core/operation/read_context_test.cpp

@@ -75,7 +75,6 @@ TEST(ReadContextTest, TestSetContent) {
     builder.SetTableSchema("table-schema-json");
     builder.WithBranch("rt");
     builder.WithCacheConfig(cache_config);
-    builder.WithFileSystemSchemeToIdentifierMap({{"file", "local"}});
     auto fs = std::make_shared<MockFileSystem>();
     builder.WithFileSystem(fs);
     ASSERT_OK_AND_ASSIGN(auto ctx, builder.Finish());
@@ -103,8 +102,7 @@ TEST(ReadContextTest, TestSetContent) {
     ASSERT_EQ(512U, ctx->GetCacheConfig().GetRangeSizeLimit());
     ASSERT_EQ(128U, ctx->GetCacheConfig().GetHoleSizeLimit());
     ASSERT_EQ(2048U, ctx->GetCacheConfig().GetPreBufferLimit());
-    std::map<std::string, std::string> expected_fs_map = {{"file", "local"}};
-    ASSERT_EQ(expected_fs_map, ctx->GetFileSystemSchemeToIdentifierMap());
+    ASSERT_TRUE(ctx->GetFileSystemSchemeToIdentifierMap().empty());
     std::map<std::string, std::string> expected_options = {{"key", "value"}};
     ASSERT_EQ(expected_options, ctx->GetOptions());
     ASSERT_EQ(ctx->GetSpecificFileSystem(), fs);
@@ -121,4 +119,30 @@ TEST(ReadContextTest, TestSetOptionsOverridesAddedOptions) {
     ASSERT_EQ(expected_options, ctx->GetOptions());
 }
 
+TEST(ReadContextTest, TestFileSystemAndSchemeMapConflict) {
+    ReadContextBuilder builder("table_root_path");
+    auto fs = std::make_shared<MockFileSystem>();
+    builder.WithFileSystem(fs);
+    builder.WithFileSystemSchemeToIdentifierMap({{"file", "local"}});
+    ASSERT_NOK_WITH_MSG(
+        builder.Finish(),
+        "WithFileSystem() and WithFileSystemSchemeToIdentifierMap() cannot be used together");
+}
+
+TEST(ReadContextTest, TestSchemeMapWithoutFileSystem) {
+    ReadContextBuilder builder("table_root_path");
+    builder.WithFileSystemSchemeToIdentifierMap({{"file", "local"}});
+    ASSERT_OK_AND_ASSIGN(auto ctx, builder.Finish());
+    std::map<std::string, std::string> expected_fs_map = {{"file", "local"}};
+    ASSERT_EQ(expected_fs_map, ctx->GetFileSystemSchemeToIdentifierMap());
+    ASSERT_FALSE(ctx->GetSpecificFileSystem());
+}
+
+TEST(ReadContextTest, TestPrefetchMaxParallelNumZero) {
+    ReadContextBuilder builder("table_root_path");
+    builder.EnablePrefetch(true);
+    builder.SetPrefetchMaxParallelNum(0);
+    ASSERT_NOK_WITH_MSG(builder.Finish(), "prefetch max parallel num should be greater than 0");
+}
+
 }  // namespace paimon::test