[native]Advance velox and update memory part

presto-native-execution/presto_cpp/main/PeriodicServiceInventoryManager.cpp

@@ -31,7 +31,7 @@ PeriodicServiceInventoryManager::PeriodicServiceInventoryManager(
       ciphers_(std::move(ciphers)),
       id_(std::move(id)),
       frequencyMs_(frequencyMs),
-      pool_(velox::memory::addDefaultLeafMemoryPool(id_)),
+      pool_(velox::memory::deprecatedAddDefaultLeafMemoryPool(id_)),
       eventBaseThread_(false /*autostart*/) {}
 
 void PeriodicServiceInventoryManager::start() {

presto-native-execution/presto_cpp/main/PeriodicTaskManager.cpp

@@ -75,7 +75,7 @@ PeriodicTaskManager::PeriodicTaskManager(
       taskManager_(taskManager),
       memoryAllocator_(memoryAllocator),
       asyncDataCache_(asyncDataCache),
-      arbitrator_(velox::memory::defaultMemoryManager().arbitrator()),
+      arbitrator_(velox::memory::deprecatedDefaultMemoryManager().arbitrator()),
       connectors_(connectors) {}
 
 void PeriodicTaskManager::start() {

presto-native-execution/presto_cpp/main/PrestoServer.cpp

@@ -346,7 +346,8 @@ void PrestoServer::run() {
   facebook::velox::exec::ExchangeSource::registerFactory(
       operators::BroadcastExchangeSource::createExchangeSource);
 
-  pool_ = velox::memory::addDefaultLeafMemoryPool();
+  pool_ =
+      velox::memory::MemoryManager::getInstance()->addLeafPool("PrestoServer");
   taskManager_ = std::make_unique<TaskManager>(
       driverExecutor_.get(), httpSrvCpuExecutor_.get(), spillerExecutor_.get());
 
@@ -656,9 +657,10 @@ void PrestoServer::initializeVeloxMemory() {
         systemConfig->memoryPoolTransferCapacity();
     options.arbitrationStateCheckCb = velox::exec::memoryArbitrationStateCheck;
   }
-  const auto& manager = memory::MemoryManager::getInstance(options);
-  PRESTO_STARTUP_LOG(INFO) << "Memory manager has been setup: "
-                           << manager.toString();
+  memory::MemoryManager::initialize(options);
+  PRESTO_STARTUP_LOG(INFO)
+      << "Memory manager has been setup: "
+      << memory::deprecatedDefaultMemoryManager().toString();
 }
 
 void PrestoServer::stop() {

presto-native-execution/presto_cpp/main/QueryContextManager.cpp

@@ -138,7 +138,7 @@ std::shared_ptr<core::QueryCtx> QueryContextManager::findOrCreateQueryCtx(
   // though the query ctx has been evicted out of the cache. The query ctx cache
   // is still indexed by the query id.
   static std::atomic_uint64_t poolId{0};
-  auto pool = memory::defaultMemoryManager().addRootPool(
+  auto pool = memory::MemoryManager::getInstance()->addRootPool(
       fmt::format("{}_{}", queryId, poolId++),
       queryConfig.queryMaxMemoryPerNode() != 0
           ? queryConfig.queryMaxMemoryPerNode()

presto-native-execution/presto_cpp/main/http/tests/HttpJwtTest.cpp

@@ -33,6 +33,10 @@ class HttpJwtTestSuite : public ::testing::TestWithParam<bool> {
   }
 
  protected:
+  static void SetUpTestCase() {
+    memory::MemoryManager::testingSetInstance({});
+  }
+
   std::unique_ptr<Config> jwtSystemConfig(
       const std::unordered_map<std::string, std::string> configOverride = {})
       const {
@@ -67,7 +71,8 @@ class HttpJwtTestSuite : public ::testing::TestWithParam<bool> {
       std::unordered_map<std::string, std::string> serverSystemConfigOverride =
           {},
       const uint64_t sendDelayMs = 500) {
-    auto memoryPool = defaultMemoryManager().addLeafPool("HttpJwtTestSuite");
+    auto memoryPool =
+        memory::MemoryManager::getInstance()->addLeafPool("HttpJwtTestSuite");
     auto clientConfig = jwtSystemConfig(clientSystemConfigOverride);
     auto systemConfig = SystemConfig::instance();
     systemConfig->initialize(std::move(clientConfig));

presto-native-execution/presto_cpp/main/http/tests/HttpTest.cpp

@@ -19,10 +19,15 @@ int main(int argc, char** argv) {
   return RUN_ALL_TESTS();
 }
 
-class HttpsBasicTest : public ::testing::Test {};
+class HttpsBasicTest : public ::testing::Test {
+ protected:
+  static void SetUpTestCase() {
+    memory::MemoryManager::testingSetInstance({});
+  }
+};
 
 TEST_F(HttpsBasicTest, ssl) {
-  auto memoryPool = defaultMemoryManager().addLeafPool("ssl");
+  auto memoryPool = memory::MemoryManager::getInstance()->addLeafPool("ssl");
 
   std::string certPath = getCertsPath("test_cert1.pem");
   std::string keyPath = getCertsPath("test_key1.pem");
@@ -67,7 +72,7 @@ class HttpTestSuite : public ::testing::TestWithParam<bool> {
 };
 
 TEST_P(HttpTestSuite, basic) {
-  auto memoryPool = defaultMemoryManager().addLeafPool("basic");
+  auto memoryPool = memory::MemoryManager::getInstance()->addLeafPool("basic");
 
   const bool useHttps = GetParam();
   auto server = getServer(useHttps);
@@ -128,7 +133,8 @@ TEST_P(HttpTestSuite, basic) {
 
 TEST_P(HttpTestSuite, httpResponseAllocationFailure) {
   const int64_t memoryCapBytes = 1 << 10;
-  auto rootPool = defaultMemoryManager().addRootPool("", memoryCapBytes);
+  auto rootPool =
+      memory::MemoryManager::getInstance()->addRootPool("", memoryCapBytes);
   auto leafPool = rootPool->addLeafChild("httpResponseAllocationFailure");
 
   const bool useHttps = GetParam();
@@ -160,7 +166,8 @@ TEST_P(HttpTestSuite, httpResponseAllocationFailure) {
 }
 
 TEST_P(HttpTestSuite, serverRestart) {
-  auto memoryPool = defaultMemoryManager().addLeafPool("serverRestart");
+  auto memoryPool =
+      memory::MemoryManager::getInstance()->addLeafPool("serverRestart");
 
   const bool useHttps = GetParam();
   auto ioPool = std::make_shared<folly::IOThreadPoolExecutor>(
@@ -206,7 +213,8 @@ TEST_P(HttpTestSuite, serverRestart) {
 }
 
 TEST_P(HttpTestSuite, asyncRequests) {
-  auto memoryPool = defaultMemoryManager().addLeafPool("asyncRequests");
+  auto memoryPool =
+      memory::MemoryManager::getInstance()->addLeafPool("asyncRequests");
 
   const bool useHttps = GetParam();
   auto server = getServer(useHttps);
@@ -244,7 +252,8 @@ TEST_P(HttpTestSuite, asyncRequests) {
 }
 
 TEST_P(HttpTestSuite, timedOutRequests) {
-  auto memoryPool = defaultMemoryManager().addLeafPool("timedOutRequests");
+  auto memoryPool =
+      memory::MemoryManager::getInstance()->addLeafPool("timedOutRequests");
 
   const bool useHttps = GetParam();
   auto server = getServer(useHttps);
@@ -288,7 +297,8 @@ TEST_P(HttpTestSuite, httpConnectTimeout) {
     // handled in kernel.
     return;
   }
-  auto memoryPool = defaultMemoryManager().addLeafPool("httpTimeouts");
+  auto memoryPool =
+      memory::MemoryManager::getInstance()->addLeafPool("httpTimeouts");
 
   auto ioPool = std::make_shared<folly::IOThreadPoolExecutor>(
       1, std::make_shared<folly::NamedThreadFactory>("HTTPSrvIO"));
@@ -365,7 +375,8 @@ TEST_P(HttpTestSuite, httpConnectTimeout) {
 }
 
 TEST_P(HttpTestSuite, httpRequestTimeout) {
-  auto memoryPool = defaultMemoryManager().addLeafPool("httpRequestTimeout");
+  auto memoryPool =
+      memory::MemoryManager::getInstance()->addLeafPool("httpRequestTimeout");
 
   const bool useHttps = GetParam();
   auto ioPool = std::make_shared<folly::IOThreadPoolExecutor>(
@@ -410,8 +421,8 @@ TEST_P(HttpTestSuite, httpRequestTimeout) {
 // TODO: Enabled it when fixed.
 // Disabled it, while we are investigating and fixing this test failure.
 TEST_P(HttpTestSuite, DISABLED_outstandingRequests) {
-  auto memoryPool =
-      defaultMemoryManager().addLeafPool("DISABLED_outstandingRequests");
+  auto memoryPool = memory::MemoryManager::getInstance()->addLeafPool(
+      "DISABLED_outstandingRequests");
 
   const bool useHttps = GetParam();
   auto server = getServer(useHttps);
@@ -449,7 +460,8 @@ TEST_P(HttpTestSuite, DISABLED_outstandingRequests) {
 
 TEST_P(HttpTestSuite, testReportOnBodyStatsFunc) {
   std::atomic<int> reportedCount = 0;
-  auto memoryPool = defaultMemoryManager().addLeafPool("asyncRequests");
+  auto memoryPool =
+      memory::MemoryManager::getInstance()->addLeafPool("asyncRequests");
 
   const bool useHttps = GetParam();
   auto server = getServer(useHttps);

presto-native-execution/presto_cpp/main/operators/tests/PlanNodeSerdeTest.cpp

@@ -29,6 +29,10 @@ namespace facebook::velox::exec::test {
 class PlanNodeSerdeTest : public testing::Test,
                           public velox::test::VectorTestBase {
  protected:
+  static void SetUpTestCase() {
+    memory::MemoryManager::testingSetInstance({});
+  }
+
   PlanNodeSerdeTest() {
     functions::prestosql::registerAllScalarFunctions();
     aggregate::prestosql::registerAllAggregateFunctions();

presto-native-execution/presto_cpp/main/tests/PrestoExchangeSourceTest.cpp

@@ -391,7 +391,7 @@ struct Params {
 class PrestoExchangeSourceTest : public ::testing::TestWithParam<Params> {
  public:
   void SetUp() override {
-    pool_ = memory::addDefaultLeafMemoryPool();
+    pool_ = memory::deprecatedAddDefaultLeafMemoryPool();
 
     memory::MmapAllocator::Options options;
     options.capacity = 1L << 30;
@@ -767,7 +767,8 @@ DEBUG_ONLY_TEST_P(
   for (bool persistentError : {false, true}) {
     SCOPED_TRACE(fmt::format("persistentError: {}", persistentError));
 
-    auto rootPool = defaultMemoryManager().addRootPool("", memoryCapBytes);
+    auto rootPool =
+        deprecatedDefaultMemoryManager().addRootPool("", memoryCapBytes);
     const std::string leafPoolName("exceedingMemoryCapacityForHttpResponse");
     auto leafPool = rootPool->addLeafChild(leafPoolName);
 
@@ -830,7 +831,7 @@ TEST_P(PrestoExchangeSourceTest, memoryAllocationAndUsageCheck) {
     SCOPED_TRACE(fmt::format("resetPeak {}", resetPeak));
 
     PrestoExchangeSource::testingClearMemoryUsage();
-    auto rootPool = defaultMemoryManager().addRootPool();
+    auto rootPool = memory::MemoryManager::getInstance()->addRootPool();
     auto leafPool = rootPool->addLeafChild("memoryAllocationAndUsageCheck");
 
     const bool useHttps = GetParam().useHttps;

presto-native-execution/presto_cpp/main/tests/PrestoQueryRunner.h

@@ -17,7 +17,7 @@
 #include <folly/io/async/EventBaseThread.h>
 #include "presto_cpp/main/http/HttpClient.h"
 #include "velox/common/memory/Memory.h"
-#include "velox/exec/tests/utils/ReferenceQueryRunner.h"
+#include "velox/exec/fuzzer/ReferenceQueryRunner.h"
 #include "velox/vector/ComplexVector.h"
 
 namespace facebook::presto::test {
@@ -82,7 +82,7 @@ class PrestoQueryRunner : public velox::exec::test::ReferenceQueryRunner {
   const std::string user_;
   folly::EventBaseThread eventBaseThread_{false};
   std::shared_ptr<velox::memory::MemoryPool> rootPool_{
-      velox::memory::defaultMemoryManager().addRootPool()};
+      velox::memory::deprecatedDefaultMemoryManager().addRootPool()};
   std::shared_ptr<velox::memory::MemoryPool> pool_{
       rootPool_->addLeafChild("leaf")};
 };

presto-native-execution/presto_cpp/main/tests/TaskManagerTest.cpp

@@ -179,9 +179,10 @@ class TaskManagerTest : public testing::Test {
                 kHiveConnectorId, std::make_shared<core::MemConfig>());
     connector::registerConnector(hiveConnector);
 
-    rootPool_ =
-        memory::defaultMemoryManager().addRootPool("TaskManagerTest.root");
-    leafPool_ = memory::addDefaultLeafMemoryPool("TaskManagerTest.leaf");
+    rootPool_ = memory::deprecatedDefaultMemoryManager().addRootPool(
+        "TaskManagerTest.root");
+    leafPool_ =
+        memory::deprecatedAddDefaultLeafMemoryPool("TaskManagerTest.leaf");
     rowType_ = ROW({"c0", "c1"}, {INTEGER(), VARCHAR()});
 
     taskManager_ = std::make_unique<TaskManager>(

presto-native-execution/presto_cpp/main/types/tests/CMakeLists.txt

@@ -61,6 +61,7 @@ target_link_libraries(
   velox_presto_serializer
   velox_serialization
   velox_type
+  velox_type_parser
   Boost::filesystem
   ${RE2}
   ${FOLLY_WITH_DEPENDENCIES}

presto-native-execution/presto_cpp/main/types/tests/PlanConverterTest.cpp

@@ -62,7 +62,7 @@ std::shared_ptr<const core::PlanNode> assertToVeloxQueryPlan(
   std::string fragment = slurp(getDataPath(fileName));
 
   protocol::PlanFragment prestoPlan = json::parse(fragment);
-  auto pool = memory::addDefaultLeafMemoryPool();
+  auto pool = memory::deprecatedAddDefaultLeafMemoryPool();
 
   auto queryCtx = std::make_shared<core::QueryCtx>();
   VeloxInteractiveQueryPlanConverter converter(queryCtx.get(), pool.get());
@@ -80,7 +80,7 @@ std::shared_ptr<const core::PlanNode> assertToBatchVeloxQueryPlan(
   const std::string fragment = slurp(getDataPath(fileName));
 
   protocol::PlanFragment prestoPlan = json::parse(fragment);
-  auto pool = memory::addDefaultLeafMemoryPool();
+  auto pool = memory::deprecatedAddDefaultLeafMemoryPool();
   auto queryCtx = std::make_shared<core::QueryCtx>();
   VeloxBatchQueryPlanConverter converter(
       shuffleName,
@@ -95,7 +95,12 @@ std::shared_ptr<const core::PlanNode> assertToBatchVeloxQueryPlan(
 }
 } // namespace
 
-class PlanConverterTest : public ::testing::Test {};
+class PlanConverterTest : public ::testing::Test {
+ protected:
+  static void SetUpTestCase() {
+    memory::MemoryManager::testingSetInstance({});
+  }
+};
 
 // Leaf stage plan for select regionkey, sum(1) from nation group by 1
 // Scan + Partial Agg + Repartitioning

presto-native-execution/presto_cpp/main/types/tests/RowExpressionTest.cpp

@@ -25,8 +25,12 @@ using namespace facebook::velox::core;
 
 class RowExpressionTest : public ::testing::Test {
  public:
+  static void SetUpTestCase() {
+    memory::MemoryManager::testingSetInstance({});
+  }
+
   void SetUp() override {
-    pool_ = memory::addDefaultLeafMemoryPool();
+    pool_ = memory::MemoryManager::getInstance()->addLeafPool();
     converter_ =
         std::make_unique<VeloxExprConverter>(pool_.get(), &typeParser_);
   }

presto-native-execution/presto_cpp/main/types/tests/ValuesPipeTest.cpp

@@ -64,7 +64,7 @@ TEST_F(TestValues, valuesRowVector) {
 
   testJsonRoundtrip(j, p);
 
-  auto pool = memory::addDefaultLeafMemoryPool();
+  auto pool = memory::deprecatedAddDefaultLeafMemoryPool();
   auto queryCtx = std::make_shared<core::QueryCtx>();
   VeloxInteractiveQueryPlanConverter converter(queryCtx.get(), pool.get());
   auto values = std::dynamic_pointer_cast<const core::ValuesNode>(
@@ -104,7 +104,7 @@ TEST_F(TestValues, valuesPlan) {
 
   testJsonRoundtrip(j, p);
 
-  auto pool = memory::addDefaultLeafMemoryPool();
+  auto pool = memory::deprecatedAddDefaultLeafMemoryPool();
   auto queryCtx = std::make_shared<core::QueryCtx>();
   VeloxInteractiveQueryPlanConverter converter(queryCtx.get(), pool.get());
   auto values = converter.toVeloxQueryPlan(

presto-native-execution/presto_cpp/presto_protocol/tests/Base64Test.cpp

@@ -24,7 +24,7 @@ using namespace facebook::velox;
 class Base64Test : public ::testing::Test {
  public:
   void SetUp() override {
-    pool_ = memory::addDefaultLeafMemoryPool();
+    pool_ = memory::deprecatedAddDefaultLeafMemoryPool();
   }
 
   std::shared_ptr<memory::MemoryPool> pool_;