From 20becc2489002a2d67ef957f36e9869e6f76892d Mon Sep 17 00:00:00 2001 From: Chris Pride Date: Tue, 18 Aug 2026 17:25:39 -0700 Subject: [PATCH 1/3] GH-601: Fix TOCTOU race in Gandiva's LLVM object cache read Projector::Make() and Filter::Make() each read the shared expression cache once to decide is_cached, then independently call SetLLVMObjectCache(), which performed its own second, unsynchronized read of the same cache key via GandivaObjectCache::getObject() before eagerly pre-loading a cached object into the LLJIT. Between the two reads, another thread compiling the identical (schema, expressions, selection vector mode, configuration) tuple can finish and insert into the shared cache. The first thread then loads that now-available cached object (defining e.g. symbol "expr_0_0" in its own JITDylib) via the second read, but -- since its own is_cached snapshot from the *first* read was already false -- proceeds to build and add its own freshly compiled IR module defining the same symbol name in the same JITDylib. LLVM ORC's duplicate-symbol detection then fires, crashing with "Duplicate definition of symbol 'expr_0_0'" (the GH-601 report). This is the root cause underlying arrow-java's later, coarser fix of wrapping Projector.make()/Filter.make() in a class-level `synchronized` -- serializing all concurrent compilation in the process, including unrelated queries, in order to close a race that only actually involves the same cache key. Fix: thread the single cache-lookup result (prev_cached_obj) that already determined is_cached directly into SetLLVMObjectCache, instead of letting it perform an independent second lookup. Engine::SetLLVMObjectCache and LLVMGenerator::SetLLVMObjectCache now take the already-resolved shared_ptr rather than a GandivaObjectCache reference, so the eager cache-preload decision can never end up inconsistent with the is_cached snapshot that decides whether Build() also compiles fresh IR for the same symbol. No new locking -- this removes the window entirely rather than serializing around it. Companion arrow-java change (dremio/arrow-java): replaces the blanket `synchronized` on Projector.make()/Filter.make() with a lock striped by a hash of the same cache key, since some concurrent-build serialization is still needed until this fix is vendored. Not yet build-verified in this checkout (no local LLVM/vcpkg toolchain available); needs a real Gandiva C++ build + the existing ProjectorTest::testMakeProjectorParallel-style concurrency stress test before merging. Co-Authored-By: Claude Sonnet 5 --- cpp/src/gandiva/engine.cc | 11 ++++++++--- cpp/src/gandiva/engine.h | 8 ++++++-- cpp/src/gandiva/filter.cc | 7 +++++-- cpp/src/gandiva/llvm_generator.cc | 5 +++-- cpp/src/gandiva/llvm_generator.h | 7 +++++-- cpp/src/gandiva/projector.cc | 7 +++++-- 6 files changed, 32 insertions(+), 13 deletions(-) diff --git a/cpp/src/gandiva/engine.cc b/cpp/src/gandiva/engine.cc index 894ee7e10e5d..13939981b98c 100644 --- a/cpp/src/gandiva/engine.cc +++ b/cpp/src/gandiva/engine.cc @@ -275,9 +275,14 @@ arrow::Status VerifyAndLinkModule( } // namespace -Status Engine::SetLLVMObjectCache(GandivaObjectCache& object_cache) { - auto cached_buffer = object_cache.getObject(nullptr); - if (cached_buffer) { +Status Engine::SetLLVMObjectCache( + const std::shared_ptr& prev_cached_obj) { + if (prev_cached_obj) { + // Copy rather than hand over the cache's own buffer -- addObjectFile takes + // ownership and may free it, but the shared cache map holds a reference to the + // same object too. + auto cached_buffer = prev_cached_obj->getMemBufferCopy( + prev_cached_obj->getBuffer(), prev_cached_obj->getBufferIdentifier()); auto error = lljit_->addObjectFile(std::move(cached_buffer)); if (error) { return Status::CodeGenError("Failed to add cached object file to LLJIT: ", diff --git a/cpp/src/gandiva/engine.h b/cpp/src/gandiva/engine.h index 20165787cb66..23fc366fe15a 100644 --- a/cpp/src/gandiva/engine.h +++ b/cpp/src/gandiva/engine.h @@ -74,8 +74,12 @@ class GANDIVA_EXPORT Engine { /// Optimise and compile the module. Status FinalizeModule(); - /// Set LLVM ObjectCache. - Status SetLLVMObjectCache(GandivaObjectCache& object_cache); + /// Pre-load the LLJIT with an already-compiled object for the current module, if + /// one was found by the caller's own cache lookup. Takes the lookup result directly + /// rather than re-querying the cache, so this can never disagree with the \p cached + /// flag passed to \p Make -- avoiding the load being based on a newer cache state + /// than the one that decided whether Make() would still build (and load) fresh IR. + Status SetLLVMObjectCache(const std::shared_ptr& prev_cached_obj); /// Get the compiled function corresponding to the irfunction. Result CompiledFunction(const std::string& function); diff --git a/cpp/src/gandiva/filter.cc b/cpp/src/gandiva/filter.cc index 8a270cfdc06f..be67c12254f9 100644 --- a/cpp/src/gandiva/filter.cc +++ b/cpp/src/gandiva/filter.cc @@ -76,8 +76,11 @@ Status Filter::Make(SchemaPtr schema, ConditionPtr condition, ARROW_RETURN_NOT_OK(expr_validator.Validate(condition)); } - // Set the object cache for LLVM - ARROW_RETURN_NOT_OK(llvm_gen->SetLLVMObjectCache(obj_cache)); + // Pre-load the already-compiled object from the cache lookup above, if there was + // one. Reuses prev_cached_obj instead of letting SetLLVMObjectCache re-query the + // shared cache, so this can never disagree with the is_cached snapshot that + // decided whether Build() below still compiles fresh IR (see GH-601). + ARROW_RETURN_NOT_OK(llvm_gen->SetLLVMObjectCache(prev_cached_obj)); ARROW_RETURN_NOT_OK(llvm_gen->Build({condition}, SelectionVector::Mode::MODE_NONE)); diff --git a/cpp/src/gandiva/llvm_generator.cc b/cpp/src/gandiva/llvm_generator.cc index bf5207efe193..5ec014341652 100644 --- a/cpp/src/gandiva/llvm_generator.cc +++ b/cpp/src/gandiva/llvm_generator.cc @@ -65,8 +65,9 @@ LLVMGenerator::GetCache() { return shared_cache; } -Status LLVMGenerator::SetLLVMObjectCache(GandivaObjectCache& object_cache) { - return engine_->SetLLVMObjectCache(object_cache); +Status LLVMGenerator::SetLLVMObjectCache( + const std::shared_ptr& prev_cached_obj) { + return engine_->SetLLVMObjectCache(prev_cached_obj); } Status LLVMGenerator::Add(const ExpressionPtr expr, const FieldDescriptorPtr output) { diff --git a/cpp/src/gandiva/llvm_generator.h b/cpp/src/gandiva/llvm_generator.h index 1e7882d1bb2b..0528157b0143 100644 --- a/cpp/src/gandiva/llvm_generator.h +++ b/cpp/src/gandiva/llvm_generator.h @@ -58,8 +58,11 @@ class GANDIVA_EXPORT LLVMGenerator { static std::shared_ptr>> GetCache(); - /// \brief Set LLVM ObjectCache. - Status SetLLVMObjectCache(GandivaObjectCache& object_cache); + /// \brief Pre-load the engine with an already-compiled object for the current + /// module, if the caller's own cache lookup found one. See the equivalent Engine + /// method for why this takes the lookup result directly instead of re-querying + /// the cache. + Status SetLLVMObjectCache(const std::shared_ptr& prev_cached_obj); /// \brief Resolve the precompiled function name, remapping to a TimestampIR variant /// when the function's params include a non-millisecond timestamp argument. diff --git a/cpp/src/gandiva/projector.cc b/cpp/src/gandiva/projector.cc index ec0302146fff..eb52477ff6d6 100644 --- a/cpp/src/gandiva/projector.cc +++ b/cpp/src/gandiva/projector.cc @@ -94,8 +94,11 @@ Status Projector::Make(SchemaPtr schema, const ExpressionVector& exprs, } } - // Set the object cache for LLVM - ARROW_RETURN_NOT_OK(llvm_gen->SetLLVMObjectCache(obj_cache)); + // Pre-load the already-compiled object from the cache lookup above, if there was + // one. Reuses prev_cached_obj instead of letting SetLLVMObjectCache re-query the + // shared cache, so this can never disagree with the is_cached snapshot that + // decided whether Build() below still compiles fresh IR (see GH-601). + ARROW_RETURN_NOT_OK(llvm_gen->SetLLVMObjectCache(prev_cached_obj)); ARROW_RETURN_NOT_OK(llvm_gen->Build(exprs, selection_vector_mode)); From 0be394b7e1f871b1c8b9e2adbd9451dcfdea7db7 Mon Sep 17 00:00:00 2001 From: "logan.riggs@gmail.com" Date: Fri, 18 Sep 2026 00:32:07 +0000 Subject: [PATCH 2/3] more test --- cpp/src/gandiva/tests/CMakeLists.txt | 1 + cpp/src/gandiva/tests/concurrent_make_test.cc | 205 ++++++++++++++++++ 2 files changed, 206 insertions(+) create mode 100644 cpp/src/gandiva/tests/concurrent_make_test.cc diff --git a/cpp/src/gandiva/tests/CMakeLists.txt b/cpp/src/gandiva/tests/CMakeLists.txt index be635e16d3ad..52d32cb7f01f 100644 --- a/cpp/src/gandiva/tests/CMakeLists.txt +++ b/cpp/src/gandiva/tests/CMakeLists.txt @@ -19,6 +19,7 @@ add_gandiva_test(projector-test SOURCES binary_test.cc boolean_expr_test.cc + concurrent_make_test.cc date_time_test.cc decimal_alignment_test.cc decimal_single_test.cc diff --git a/cpp/src/gandiva/tests/concurrent_make_test.cc b/cpp/src/gandiva/tests/concurrent_make_test.cc new file mode 100644 index 000000000000..f6bb7e805345 --- /dev/null +++ b/cpp/src/gandiva/tests/concurrent_make_test.cc @@ -0,0 +1,205 @@ +// 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. + +// Regression tests for GH-601: concurrent Projector::Make()/Filter::Make() calls for the +// *same* expression cache key. +// +// Before the fix, Make() read the shared object cache once to decide `is_cached`, then +// SetLLVMObjectCache() performed a second, independent read of the same key. A thread that +// saw a miss on the first read and a hit on the second one would both pre-load the cached +// object (defining e.g. "expr_0_0" in its JITDylib) *and* add its own freshly compiled IR +// module defining the same symbol, so LLVM ORC's duplicate-symbol detection fired and +// Make() returned "Failed to add IR module to LLJIT: Duplicate definition of symbol". +// +// Note that this surfaces as a returned Status rather than a crash, so every Make() below +// must be asserted on -- that is precisely what the pre-existing Java-side repro +// (ProjectorTest#testMakeProjectorParallel) failed to do. + +#include + +#include +#include +#include +#include +#include +#include + +#include "arrow/memory_pool.h" +#include "gandiva/filter.h" +#include "gandiva/projector.h" +#include "gandiva/tests/test_util.h" +#include "gandiva/tree_expr_builder.h" + +namespace gandiva { + +using arrow::int32; + +namespace { + +// Releases every waiting thread at once, so the racing Make() calls actually overlap +// instead of trickling in as each thread is spawned. Hand-rolled rather than std::latch +// to avoid depending on the standard library's C++20 support. +class StartGate { + public: + void Wait() { + std::unique_lock lock(mutex_); + cv_.wait(lock, [this] { return open_; }); + } + + void Open() { + { + std::lock_guard lock(mutex_); + open_ = true; + } + cv_.notify_all(); + } + + private: + std::mutex mutex_; + std::condition_variable cv_; + bool open_ = false; +}; + +int NumThreads() { + auto hw = static_cast(std::thread::hardware_concurrency()); + return std::max(8, 2 * std::max(1, hw)); +} + +// Each iteration needs a cache key that has never been built before, so that every thread +// in that iteration genuinely starts from a miss. Note that TestConfiguration() returns +// the *same* shared Configuration instance every call, and ExpressionCacheKey compares +// configurations by pointer identity, so varying the configuration would not produce a +// miss -- vary the schema instead. +constexpr int kIterations = 25; + +} // namespace + +class TestConcurrentMake : public ::testing::Test { + public: + void SetUp() { pool_ = arrow::default_memory_pool(); } + + protected: + arrow::MemoryPool* pool_; +}; + +TEST_F(TestConcurrentMake, TestProjectorMakeSameCacheKey) { + const int num_threads = NumThreads(); + + for (int iter = 0; iter < kIterations; ++iter) { + // Unique per iteration => guaranteed cache miss for the first thread to get there. + auto field0 = field("f0_" + std::to_string(iter), int32()); + auto field1 = field("f1_" + std::to_string(iter), int32()); + auto schema = arrow::schema({field0, field1}); + auto field_sum = field("add_" + std::to_string(iter), int32()); + auto sum_expr = TreeExprBuilder::MakeExpression("add", {field0, field1}, field_sum); + auto configuration = TestConfiguration(); + + StartGate gate; + std::vector threads; + std::vector statuses(num_threads); + std::vector> projectors(num_threads); + + threads.reserve(num_threads); + for (int i = 0; i < num_threads; ++i) { + threads.emplace_back([&, i] { + gate.Wait(); + statuses[i] = Projector::Make(schema, {sum_expr}, configuration, &projectors[i]); + }); + } + gate.Open(); + for (auto& thread : threads) { + thread.join(); + } + + // Create a row-batch with some sample data. + int num_records = 4; + auto array0 = MakeArrowArrayInt32({1, 2, 3, 4}, {true, true, true, true}); + auto array1 = MakeArrowArrayInt32({11, 13, 15, 17}, {true, true, true, true}); + auto exp_sum = MakeArrowArrayInt32({12, 15, 18, 21}, {true, true, true, true}); + auto in_batch = arrow::RecordBatch::Make(schema, num_records, {array0, array1}); + + for (int i = 0; i < num_threads; ++i) { + ASSERT_OK(statuses[i]) << "iteration " << iter << ", thread " << i; + ASSERT_NE(projectors[i], nullptr) << "iteration " << iter << ", thread " << i; + + // Every projector must evaluate correctly, including the ones built from the cache + // hit path. No pre-existing test evaluates a cache-hit projector, so a + // corrupt-but-non-null cached module would otherwise go unnoticed. + arrow::ArrayVector outputs; + ASSERT_OK(projectors[i]->Evaluate(*in_batch, pool_, &outputs)) + << "iteration " << iter << ", thread " << i; + EXPECT_ARROW_ARRAY_EQUALS(exp_sum, outputs.at(0)); + } + } +} + +TEST_F(TestConcurrentMake, TestFilterMakeSameCacheKey) { + const int num_threads = NumThreads(); + + for (int iter = 0; iter < kIterations; ++iter) { + auto field0 = field("g0_" + std::to_string(iter), int32()); + auto field1 = field("g1_" + std::to_string(iter), int32()); + auto schema = arrow::schema({field0, field1}); + + // Condition: f0 + f1 < 10 + auto node_f0 = TreeExprBuilder::MakeField(field0); + auto node_f1 = TreeExprBuilder::MakeField(field1); + auto sum_func = + TreeExprBuilder::MakeFunction("add", {node_f0, node_f1}, arrow::int32()); + auto literal_10 = TreeExprBuilder::MakeLiteral(static_cast(10)); + auto less_than_10 = TreeExprBuilder::MakeFunction("less_than", {sum_func, literal_10}, + arrow::boolean()); + auto condition = TreeExprBuilder::MakeCondition(less_than_10); + auto configuration = TestConfiguration(); + + StartGate gate; + std::vector threads; + std::vector statuses(num_threads); + std::vector> filters(num_threads); + + threads.reserve(num_threads); + for (int i = 0; i < num_threads; ++i) { + threads.emplace_back([&, i] { + gate.Wait(); + statuses[i] = Filter::Make(schema, condition, configuration, &filters[i]); + }); + } + gate.Open(); + for (auto& thread : threads) { + thread.join(); + } + + int num_records = 5; + auto array0 = MakeArrowArrayInt32({1, 2, 3, 4, 6}, {true, true, true, false, true}); + auto array1 = MakeArrowArrayInt32({5, 9, 6, 17, 3}, {true, true, false, true, true}); + auto exp = MakeArrowArrayUint16({0, 4}); + auto in_batch = arrow::RecordBatch::Make(schema, num_records, {array0, array1}); + + for (int i = 0; i < num_threads; ++i) { + ASSERT_OK(statuses[i]) << "iteration " << iter << ", thread " << i; + ASSERT_NE(filters[i], nullptr) << "iteration " << iter << ", thread " << i; + + std::shared_ptr selection_vector; + ASSERT_OK(SelectionVector::MakeInt16(num_records, pool_, &selection_vector)); + ASSERT_OK(filters[i]->Evaluate(*in_batch, selection_vector)) + << "iteration " << iter << ", thread " << i; + EXPECT_ARROW_ARRAY_EQUALS(exp, selection_vector->ToArray()); + } + } +} + +} // namespace gandiva From 80309f9b8d2e70a85c6ca423358d1a3fe24a0511 Mon Sep 17 00:00:00 2001 From: "logan.riggs@gmail.com" Date: Tue, 6 Oct 2026 20:55:11 +0000 Subject: [PATCH 3/3] Cleanup tests and limit max concurrnecy --- cpp/src/gandiva/tests/concurrent_make_test.cc | 53 +++++++++++-------- 1 file changed, 30 insertions(+), 23 deletions(-) diff --git a/cpp/src/gandiva/tests/concurrent_make_test.cc b/cpp/src/gandiva/tests/concurrent_make_test.cc index f6bb7e805345..db50fb299af9 100644 --- a/cpp/src/gandiva/tests/concurrent_make_test.cc +++ b/cpp/src/gandiva/tests/concurrent_make_test.cc @@ -19,15 +19,13 @@ // *same* expression cache key. // // Before the fix, Make() read the shared object cache once to decide `is_cached`, then -// SetLLVMObjectCache() performed a second, independent read of the same key. A thread that -// saw a miss on the first read and a hit on the second one would both pre-load the cached -// object (defining e.g. "expr_0_0" in its JITDylib) *and* add its own freshly compiled IR -// module defining the same symbol, so LLVM ORC's duplicate-symbol detection fired and -// Make() returned "Failed to add IR module to LLJIT: Duplicate definition of symbol". +// SetLLVMObjectCache() performed a second, independent read of the same key. A thread +// that saw a miss on the first read and a hit on the second one would both pre-load the +// cached object (defining e.g. "expr_0_0" in its JITDylib) *and* add its own freshly +// compiled IR module defining the same symbol, so LLVM ORC's duplicate-symbol detection +// fired and Make() returned "Failed to add IR module to LLJIT: Duplicate definition of +// symbol". // -// Note that this surfaces as a returned Status rather than a crash, so every Make() below -// must be asserted on -- that is precisely what the pre-existing Java-side repro -// (ProjectorTest#testMakeProjectorParallel) failed to do. #include @@ -50,9 +48,7 @@ using arrow::int32; namespace { -// Releases every waiting thread at once, so the racing Make() calls actually overlap -// instead of trickling in as each thread is spawned. Hand-rolled rather than std::latch -// to avoid depending on the standard library's C++20 support. +// Releases every waiting thread at once, so the racing Make() calls overlap. class StartGate { public: void Wait() { @@ -74,23 +70,37 @@ class StartGate { bool open_ = false; }; -int NumThreads() { - auto hw = static_cast(std::thread::hardware_concurrency()); - return std::max(8, 2 * std::max(1, hw)); +int HardwareConcurrency() { + return std::max(1, static_cast(std::thread::hardware_concurrency())); } -// Each iteration needs a cache key that has never been built before, so that every thread -// in that iteration genuinely starts from a miss. Note that TestConfiguration() returns -// the *same* shared Configuration instance every call, and ExpressionCacheKey compares -// configurations by pointer identity, so varying the configuration would not produce a -// miss -- vary the schema instead. +int NumThreads() { return std::min(32, std::max(8, 2 * HardwareConcurrency())); } + +// This test only works when the threads are oversubscribed relative to the CPUs: the race +// window is between two reads of the shared cache inside one Make(). +// So if the cap in NumThreads() has pulled the count under 2x, this test +// cannot fail and must not report success. +bool HasEnoughOversubscription() { return NumThreads() >= 2 * HardwareConcurrency(); } + constexpr int kIterations = 25; } // namespace class TestConcurrentMake : public ::testing::Test { public: - void SetUp() { pool_ = arrow::default_memory_pool(); } + void SetUp() { + if (!HasEnoughOversubscription()) { + GTEST_SKIP() + << "Needs >= 2x thread oversubscription to observe the GH-601 race, but " + "NumThreads() is " + << NumThreads() << " against hardware_concurrency() " << HardwareConcurrency() + << ". Skipping rather than passing vacuously -- see the note on " + "HasEnoughOversubscription(). Note that hardware_concurrency() is not " + "cgroup-aware, so in a CPU-limited container it reports the host's " + "cores and this skip may be unnecessary."; + } + pool_ = arrow::default_memory_pool(); + } protected: arrow::MemoryPool* pool_; @@ -136,9 +146,6 @@ TEST_F(TestConcurrentMake, TestProjectorMakeSameCacheKey) { ASSERT_OK(statuses[i]) << "iteration " << iter << ", thread " << i; ASSERT_NE(projectors[i], nullptr) << "iteration " << iter << ", thread " << i; - // Every projector must evaluate correctly, including the ones built from the cache - // hit path. No pre-existing test evaluates a cache-hit projector, so a - // corrupt-but-non-null cached module would otherwise go unnoticed. arrow::ArrayVector outputs; ASSERT_OK(projectors[i]->Evaluate(*in_batch, pool_, &outputs)) << "iteration " << iter << ", thread " << i;