fix(plan): Propagate memory pool context during deserialization (#12029)

Summary:
Pull Request resolved: #12029

While deserializing query plans, we need to ensure that the
void* context (which is internally cast as a MemoryPool*) is properly
propagated when deserializing lambda expressions.

Reviewed By: xiaoxmeng, PatrickKan, hongtaoy

Differential Revision: D67887156

fbshipit-source-id: 69331743573b94b3d85f56c12778b5b28676334a

velox/core/Expressions.cpp

@@ -237,7 +237,7 @@ folly::dynamic LambdaTypedExpr::serialize() const {
 // static
 TypedExprPtr LambdaTypedExpr::create(const folly::dynamic& obj, void* context) {
   auto signature = ISerializable::deserialize<Type>(obj["signature"]);
-  auto body = ISerializable::deserialize<ITypedExpr>(obj["body"]);
+  auto body = ISerializable::deserialize<ITypedExpr>(obj["body"], context);
 
   return std::make_shared<LambdaTypedExpr>(
       asRowType(signature), std::move(body));

velox/core/PlanNode.cpp

@@ -2120,7 +2120,8 @@ PlanNodePtr TableWriteNode::create(const folly::dynamic& obj, void* context) {
   std::shared_ptr<AggregationNode> aggregationNode;
   if (obj.count("aggregationNode") != 0) {
     aggregationNode = std::const_pointer_cast<AggregationNode>(
-        ISerializable::deserialize<AggregationNode>(obj["aggregationNode"]));
+        ISerializable::deserialize<AggregationNode>(
+            obj["aggregationNode"], context));
   }
   auto connectorId = obj["connectorId"].asString();
   auto connectorInsertTableHandle =

velox/core/tests/TypedExprSerdeTest.cpp

@@ -139,16 +139,18 @@ TEST_F(TypedExprSerDeTest, concat) {
 }
 
 TEST_F(TypedExprSerDeTest, lambda) {
-  // x -> (x > 10)
+  // x -> (x in (1, ..., 5))
   auto expression = std::make_shared<LambdaTypedExpr>(
       ROW({"x"}, {BIGINT()}),
       std::make_shared<CallTypedExpr>(
           BOOLEAN(),
           std::vector<TypedExprPtr>{
               std::make_shared<FieldAccessTypedExpr>(BIGINT(), "x"),
-              std::make_shared<ConstantTypedExpr>(BIGINT(), 10LL),
+              std::make_shared<ConstantTypedExpr>(makeArrayVector<int64_t>({
+                  {1, 2, 3, 4, 5},
+              })),
           },
-          "gt"));
+          "in"));
   testSerde(expression);
 }
 

velox/vector/VectorSaver.cpp

@@ -648,6 +648,8 @@ void saveStringToFile(const std::string& content, const char* filePath) {
 }
 
 VectorPtr restoreVector(std::istream& in, memory::MemoryPool* pool) {
+  VELOX_CHECK_NOT_NULL(pool);
+
   // Encoding.
   auto encoding = readEncoding(in);