fix cpp build after sync origin/main with common patches merged

taiyang-li · taiyang-li · commit d3f2bd0ffd56 · 2026-07-08T17:18:31.000+08:00
diff --git a/cpp/bolt/benchmarks/GenericBenchmark.cc b/cpp/bolt/benchmarks/GenericBenchmark.cc
@@ -397,7 +397,7 @@ auto BM_Generic = [](::benchmark::State& state,
   listener->updateLimit(FLAGS_memory_limit);
 
   auto* listenerPtr = listener.get();
-  auto* memoryManager = MemoryManager::create(kBoltBackendKind, std::move(listener), "bm-generic");
+  auto* memoryManager = MemoryManager::create(kBoltBackendKind, std::move(listener));
   auto runtime = runtimeFactory(memoryManager);
 
   auto plan = getPlanFromFile("Plan", planFile);
@@ -518,7 +518,7 @@ auto BM_ShuffleWriteRead = [](::benchmark::State& state,
   listener->updateLimit(FLAGS_memory_limit);
 
   auto* listenerPtr = listener.get();
-  auto* memoryManager = MemoryManager::create(kBoltBackendKind, std::move(listener), "bm-shuffle-write-read");
+  auto* memoryManager = MemoryManager::create(kBoltBackendKind, std::move(listener));
   auto runtime = runtimeFactory(memoryManager);
 
   const size_t dirIndex = std::hash<std::thread::id>{}(std::this_thread::get_id()) % localDirs.size();
diff --git a/cpp/bolt/compute/BoltBackend.cc b/cpp/bolt/compute/BoltBackend.cc
@@ -74,11 +74,8 @@ using namespace bytedance;
 namespace gluten {
 
 namespace {
-MemoryManager* boltMemoryManagerFactory(
-    const std::string& kind,
-    std::unique_ptr<AllocationListener> listener,
-    const std::string& name) {
-  return new BoltMemoryManager(kind, std::move(listener), *BoltBackend::get()->getBackendConf(), name);
+MemoryManager* boltMemoryManagerFactory(const std::string& kind, std::unique_ptr<AllocationListener> listener) {
+  return new BoltMemoryManager(kind, std::move(listener), *BoltBackend::get()->getBackendConf(), "");
 }
 
 void boltMemoryManagerReleaser(MemoryManager* memoryManager) {
@@ -110,12 +107,11 @@ Runtime* boltRuntimeFactory(
     const std::string& kind,
     MemoryManager* memoryManager,
     ThreadManager* threadManager,
-    const std::unordered_map<std::string, std::string>& sessionConf,
-    int64_t taskId) {
+    const std::unordered_map<std::string, std::string>& sessionConf) {
   auto* vmm = dynamic_cast<BoltMemoryManager*>(memoryManager);
   GLUTEN_CHECK(vmm != nullptr, "Not a Bolt memory manager");
   // new object every time
-  return new BoltRuntime(kind, vmm, threadManager, sessionConf, taskId);
+  return new BoltRuntime(kind, vmm, threadManager, sessionConf);
 }
 
 void boltRuntimeReleaser(Runtime* runtime) {
diff --git a/cpp/bolt/compute/BoltRuntime.cc b/cpp/bolt/compute/BoltRuntime.cc
@@ -30,6 +30,7 @@
 #include "compute/BoltPlanConverter.h"
 #include "config/BoltConfig.h"
 #include "config/GlutenConfig.h"
+#include "jni/TaskContextJniWrapper.h"
 #include "memory/BoltGlutenMemoryManager.h"
 #include "operators/serializer/BoltRowToColumnarConverter.h"
 #include "utils/ConfigExtractor.h"
@@ -66,18 +67,18 @@ BoltRuntime::BoltRuntime(
     const std::string& kind,
     BoltMemoryManager* vmm,
     ThreadManager* threadManager,
-    const std::unordered_map<std::string, std::string>& confMap,
-    int64_t taskId)
-    : Runtime(kind, vmm, threadManager, confMap, taskId) {
+    const std::unordered_map<std::string, std::string>& confMap)
+    : Runtime(kind, vmm, threadManager, confMap) {
   // Refresh session config.
   boltCfg_ =
       std::make_shared<bytedance::bolt::config::ConfigBase>(std::unordered_map<std::string, std::string>(confMap_));
 
   gluten::BoltGlutenMemoryManager::init(BoltBackend::getCombinedConf(boltCfg_)->rawConfigs());
 
   if (gluten::BoltGlutenMemoryManager::enabled()) {
+    const auto taskAttemptId = gluten::getCurrentSparkTaskAttemptId();
     auto holder = gluten::BoltGlutenMemoryManager::getMemoryManagerHolder(
-        memoryManager()->name(), taskId, reinterpret_cast<int64_t>(memoryManager()));
+        "", taskAttemptId, reinterpret_cast<int64_t>(memoryManager()));
     auto mm = holder->getManager();
     leafPool_ = mm->getLeafMemoryPool();
     aggregatePool_ = mm->getAggregateMemoryPool();
@@ -217,8 +218,9 @@ std::shared_ptr<ResultIterator> BoltRuntime::createResultIterator(
     std::weak_ptr<ResultIterator> weakAns = ans;
     auto spiller = std::make_shared<OperatorSpiller>(weakAns);
     auto genericSpiller = std::dynamic_pointer_cast<bytedance::bolt::memory::sparksql::Spiller>(spiller);
+    const auto taskAttemptId = gluten::getCurrentSparkTaskAttemptId();
     auto holder = gluten::BoltGlutenMemoryManager::getMemoryManagerHolder(
-        memoryManager()->name(), taskId(), reinterpret_cast<int64_t>(memoryManager()));
+        "", taskAttemptId, reinterpret_cast<int64_t>(memoryManager()));
     holder->appendSpiller(genericSpiller);
   }
 
@@ -368,8 +370,9 @@ std::shared_ptr<ShuffleWriterBase> BoltRuntime::createShuffleWriter(
     auto weakShuffleWriter = std::weak_ptr<ShuffleWriterBase>(shuffleWriter);
     auto spiller = std::make_shared<ShuffleSpiller>(weakShuffleWriter);
     auto genericSpiller = std::dynamic_pointer_cast<bytedance::bolt::memory::sparksql::Spiller>(spiller);
+    const auto taskAttemptId = gluten::getCurrentSparkTaskAttemptId();
     auto holder = gluten::BoltGlutenMemoryManager::getMemoryManagerHolder(
-        memoryManager()->name(), taskId(), reinterpret_cast<int64_t>(memoryManager()));
+        "", taskAttemptId, reinterpret_cast<int64_t>(memoryManager()));
     holder->appendSpiller(genericSpiller);
   }
   return shuffleWriter;
diff --git a/cpp/bolt/compute/BoltRuntime.h b/cpp/bolt/compute/BoltRuntime.h
@@ -46,8 +46,7 @@ class BoltRuntime final : public Runtime {
       const std::string& kind,
       BoltMemoryManager* vmm,
       ThreadManager* threadManager,
-      const std::unordered_map<std::string, std::string>& confMap,
-      int64_t taskId);
+      const std::unordered_map<std::string, std::string>& confMap);
 
   void setSparkTaskInfo(SparkTaskInfo taskInfo) override {
     static std::atomic<uint32_t> vtId{0};
diff --git a/cpp/bolt/compute/WholeStageResultIterator.cc b/cpp/bolt/compute/WholeStageResultIterator.cc
@@ -23,6 +23,7 @@
 #include "BoltBackend.h"
 #include "BoltRuntime.h"
 #include "config/BoltConfig.h"
+#include "jni/TaskContextJniWrapper.h"
 #include "memory/BoltMemoryManager.h"
 #include "memory/BoltGlutenMemoryManager.h"
 #include "bolt/connectors/hive/HiveConfig.h"
@@ -379,8 +380,9 @@ std::shared_ptr<bolt::core::QueryCtx> WholeStageResultIterator::createNewBoltQue
 
   memory::sparksql::BoltMemoryPoolPtr boltPool;
   if (gluten::BoltGlutenMemoryManager::enabled()) {
+    const auto taskAttemptId = gluten::getCurrentSparkTaskAttemptId();
     auto holder = gluten::BoltGlutenMemoryManager::getMemoryManagerHolder(
-        memoryManager_->name(), taskInfo_.taskId, reinterpret_cast<int64_t>(memoryManager_));
+        "", taskAttemptId, reinterpret_cast<int64_t>(memoryManager_));
     auto mm = holder->getManager();
     boltPool = mm->getAggregateMemoryPool();
   } else {
@@ -473,8 +475,9 @@ int64_t WholeStageResultIterator::spillFixedSize(int64_t size) {
   memory::sparksql::BoltMemoryPoolPtr pool;
   memory::sparksql::BoltMemoryManagerPtr manager;
   if (gluten::BoltGlutenMemoryManager::enabled()) {
+    const auto taskAttemptId = gluten::getCurrentSparkTaskAttemptId();
     auto holder = gluten::BoltGlutenMemoryManager::getMemoryManagerHolder(
-        memoryManager_->name(), taskInfo_.taskId, reinterpret_cast<int64_t>(memoryManager_));
+        "", taskAttemptId, reinterpret_cast<int64_t>(memoryManager_));
     manager = holder->getManager();
     pool = manager->getAggregateMemoryPool();
   } else {
diff --git a/cpp/bolt/memory/BoltMemoryManager.cc b/cpp/bolt/memory/BoltMemoryManager.cc
@@ -223,7 +223,7 @@ BoltMemoryManager::BoltMemoryManager(
     std::unique_ptr<AllocationListener> listener,
     const bytedance::bolt::config::ConfigBase& backendConf,
     const std::string& name)
-    : MemoryManager(kind, name), listener_(std::move(listener)) {
+    : MemoryManager(kind), listener_(std::move(listener)) {
   auto reservationBlockSize =
       backendConf.get<uint64_t>(kMemoryReservationBlockSize, kMemoryReservationBlockSizeDefault);
   blockListener_ = std::make_unique<BlockAllocationListener>(listener_.get(), reservationBlockSize);
diff --git a/cpp/bolt/tests/RuntimeTest.cc b/cpp/bolt/tests/RuntimeTest.cc
@@ -164,7 +164,7 @@ TEST(TestRuntime, CreateRuntime) {
 
 TEST(TestRuntime, CreateBoltRuntime) {
   BoltBackend::create(AllocationListener::noop(), {{kSparkOffHeapMemory, "7516192768"}});
-  auto mm = MemoryManager::create(kBoltBackendKind, AllocationListener::noop(), "test-bolt-runtime");
+  auto mm = MemoryManager::create(kBoltBackendKind, AllocationListener::noop());
   auto tm = ThreadManager::create(kBoltBackendKind, ThreadInitializer::noop());
   auto runtime = Runtime::create(kBoltBackendKind, mm, tm, {{kSparkOffHeapMemory, "7516192768"}}, 1);
   ASSERT_EQ(typeid(*runtime), typeid(BoltRuntime));
diff --git a/cpp/bolt/utils/ConfigExtractor.cc b/cpp/bolt/utils/ConfigExtractor.cc
@@ -228,20 +228,6 @@ void getAbfsHiveConfig(
 static const std::string kUseIcuRegex = "spark.gluten.sql.columnar.backend.bolt.useICURegex";
 static const std::string kUseIcuRegexDefault = "true";
 
-std::string getConfigValue(
-    const std::unordered_map<std::string, std::string>& confMap,
-    const std::string& key,
-    const std::optional<std::string>& fallbackValue) {
-  auto got = confMap.find(key);
-  if (got == confMap.end()) {
-    if (fallbackValue == std::nullopt) {
-      throw std::runtime_error("No such config key: " + key);
-    }
-    return fallbackValue.value();
-  }
-  return got->second;
-}
-
 std::shared_ptr<bytedance::bolt::config::ConfigBase> getHiveConfig(
     std::shared_ptr<bytedance::bolt::config::ConfigBase> conf,
     FileSystemType fsType) {
diff --git a/cpp/bolt/utils/ConfigExtractor.h b/cpp/bolt/utils/ConfigExtractor.h
@@ -25,16 +25,12 @@
 
 #include "config/GlutenConfig.h"
 #include "bolt/common/config/Config.h"
+#include "utils/ConfigResolver.h"
 
 namespace gluten {
 
 enum class FileSystemType : uint8_t { kHdfs, kS3, kAbfs, kGcs, kAll };
 
-std::string getConfigValue(
-    const std::unordered_map<std::string, std::string>& confMap,
-    const std::string& key,
-    const std::optional<std::string>& fallbackValue);
-
 std::shared_ptr<bytedance::bolt::config::ConfigBase> getHiveConfig(
     std::shared_ptr<bytedance::bolt::config::ConfigBase> conf,
     FileSystemType fsType = FileSystemType::kAll);
diff --git a/cpp/core/jni/JniCommon.h b/cpp/core/jni/JniCommon.h
@@ -165,10 +165,6 @@ class JniCommonState {
 
   void close();
 
-  JavaVM* getJavaVM() const {
-    return vm_;
-  }
-
   jmethodID runtimeAwareCtxHandle();
 
   JavaVM* getJavaVM() const {
diff --git a/cpp/core/jni/JniWrapper.cc b/cpp/core/jni/JniWrapper.cc
@@ -26,6 +26,7 @@
 #include "jni/JniCommon.h"
 #include "jni/JniError.h"
 #include "jni/JniWrapper.h"
+#include "jni/TaskContextJniWrapper.h"
 
 #include "memory/ColumnarBatch.h"
 #ifdef GLUTEN_ENABLE_BOLT
@@ -491,9 +492,9 @@ JNIEXPORT void JNICALL Java_org_apache_gluten_memory_NativeMemoryManagerJniWrapp
 
 #ifdef GLUTEN_ENABLE_BOLT
   if (gluten::BoltGlutenMemoryManager::enabled()) {
-    auto memoryManagerName = jStringToCString(env, jName);
+    const auto taskAttemptId = gluten::getCurrentSparkTaskAttemptId();
     auto holder = gluten::BoltGlutenMemoryManager::getMemoryManagerHolder(
-        memoryManagerName, taskAttemptId, reinterpret_cast<int64_t>(memoryManager));
+        "", taskAttemptId, reinterpret_cast<int64_t>(memoryManager));
     holder->hold();
   }
 #endif
@@ -511,6 +512,7 @@ JNIEXPORT void JNICALL Java_org_apache_gluten_memory_NativeMemoryManagerJniWrapp
 
 #ifdef GLUTEN_ENABLE_BOLT
   if (gluten::BoltGlutenMemoryManager::enabled()) {
+    const auto taskAttemptId = gluten::getCurrentSparkTaskAttemptId();
     gluten::BoltGlutenMemoryManager::destroy(taskAttemptId, nmmHandle);
   }
 #endif
diff --git a/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto b/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto
@@ -209,6 +209,20 @@ message ReadRel {
         bytes serialized_deletion_vector = 4;
       }
 
+      message PaimonReadOptions {
+        int32 bucket = 1;
+        int64 first_row_id = 2;
+        int64 max_sequence_number = 3;
+        int32 split_group = 4;
+        oneof file_format {
+          ParquetReadOptions parquet = 5;
+          OrcReadOptions orc = 6;
+        }
+        bool use_hive_split = 7;
+        bool raw_convertible = 8;
+        repeated string primary_keys = 9;
+      }
+
       // File reading options
       oneof file_format {
         ParquetReadOptions parquet = 9;
@@ -220,6 +234,7 @@ message ReadRel {
         JsonReadOptions json = 15;
         IcebergReadOptions iceberg = 16;
         DeltaReadOptions delta = 22;
+        PaimonReadOptions paimon = 23;
       }
 
       message partitionColumn {
