Implement plan cloning with LCP

Summary:
Implement proper plan cloning using LCP instead of hacking it with
serialization.

depends on D1815

Reviewers: teon.banek, llugovic

Reviewed By: teon.banek

Subscribers: pullbot

Differential Revision: https://phabricator.memgraph.io/D1816

src/lisp/clone.lisp

@@ -61,7 +61,17 @@ Usage example:
 (defun copy-object (source-name dest-name)
   (format nil "~A = ~A;" dest-name source-name))
 
+;; TODO: This could be a common function in types.lisp if it were improved
+;; a bit. It probably won't be necessary once we refactor LCP to use uniform
+;; type designators.
+(defun get-type (type-designator)
+  (ctypecase type-designator
+    (lcp::cpp-type type-designator)
+    (string (lcp::parse-cpp-type-declaration type-designator))
+    (symbol (lcp::cpp-type type-designator))))
+
 (defun clone-by-copy-p (object-type)
+  (let ((object-type (get-type object-type)))
     (cond
       ((string= "vector" (lcp::cpp-type-name object-type))
        (clone-by-copy-p (car (lcp::cpp-type-type-args object-type))))
@@ -74,20 +84,31 @@ Usage example:
        (and (clone-by-copy-p (first (lcp::cpp-type-type-args object-type)))
             (clone-by-copy-p (second (lcp::cpp-type-type-args object-type)))))
       ((lcp::cpp-type-type-args object-type) nil)
-    (t (or
-        (lcp::find-cpp-enum (lcp::cpp-type-name object-type))
+      ((or (lcp::find-cpp-enum (lcp::cpp-type-name object-type))
            (typep object-type 'lcp::cpp-primitive-type)
            (string= "string" (lcp::cpp-type-name object-type))
-        (not (lcp::find-cpp-class (lcp::cpp-type-name object-type)))
-        (not (lcp::cpp-class-clone-opts
-              (lcp::find-cpp-class (lcp::cpp-type-name object-type))))))))
+           ;; TODO: We might want to forbid implicit copying of unknown types once
+           ;; there's a way to globally mark type as trivially copyable. Now it is
+           ;; too annoying to add (:clone :copy) option everywhere.
+           (not (lcp::find-cpp-class (lcp::cpp-type-name object-type))))
+       t)
+      (t
+       ;; We know now that we're dealing with a C++ class defined in
+       ;; LCP.  A class is cloneable by copy only if it doesn't have
+       ;; `Clone` function defined, all of its members are cloneable
+       ;; by copy and it is not a member of inheritance hierarchy.
+       (let ((cpp-class (lcp::find-cpp-class (lcp::cpp-type-name object-type))))
+         (assert cpp-class)
+         (and (not (lcp::cpp-class-clone-opts cpp-class))
+              (not (lcp::direct-subclasses-of cpp-class))
+              (not (lcp::cpp-class-super-classes cpp-class))
+              (every (lambda (member)
+                       (or (eq (lcp::cpp-member-clone member) :copy)
+                           (clone-by-copy-p (lcp::cpp-member-type member))))
+                     (lcp::cpp-class-members cpp-class))))))))
 
 (defun clone-object (object-type source-name dest-name &key args)
-  (let ((object-type
-         (ctypecase object-type
-           (lcp::cpp-type object-type)
-           (string (lcp::parse-cpp-type-declaration object-type))
-           (symbol (lcp::cpp-type object-type))))
+  (let ((object-type (get-type object-type))
         (arg-list (format nil "~{~A~^, ~}"
                           (mapcar (lambda (name-and-type)
                                     (lcp::cpp-variable-name (first name-and-type)))
@@ -115,8 +136,7 @@ Usage example:
       ((and (lcp::find-cpp-class (lcp::cpp-type-name object-type))
             (lcp::cpp-class-clone-opts (lcp::find-cpp-class (lcp::cpp-type-name object-type))))
        (format nil "~A = ~A.Clone(~A);" dest-name source-name arg-list))
-      (t
-       (format nil "static_assert(false, \"Don't know how to clone object of type ~A\");"
+      (t (clone-error "Don't know how to clone object of type ~A"
                       (lcp::cpp-type-decl object-type))))))
 
 (defun clone-vector (elem-type source-name dest-name &key args)

src/lisp/lcp-test.lisp

@@ -566,6 +566,24 @@
               'lcp.slk:slk-error)))
 
 (deftest "clone"
+  (macrolet ((single-member-test (member expected)
+               (let ((class-sym (gensym)))
+                 `(let ((,class-sym (lcp:define-class my-class ()
+                                      (,member)
+                                      (:clone))))
+                    ,(etypecase expected
+                       (string
+                        `(is-generated (lcp.clone:clone-function-definition-for-class ,class-sym)
+                                       (format nil "MyClass Clone() const {
+                                                     MyClass object;
+                                                     ~A
+                                                     return object;
+                                                   }"
+                                               ,expected)))
+                       (symbol
+                        (assert (eq expected 'lcp.clone:clone-error))
+                        `(is-error (lcp.clone:clone-function-definition-for-class ,class-sym)
+                                   ',expected)))))))
     (subtest "no inheritance"
       (undefine-cpp-types)
       (let ((tree-class (lcp:define-class tree ()
@@ -781,18 +799,6 @@
                      }")))
     (subtest "types"
       (undefine-cpp-types)
-    (macrolet ((single-member-test (member expected)
-                 (let ((class-sym (gensym)))
-                   `(let ((,class-sym (lcp:define-class my-class ()
-                                        (,member)
-                                        (:clone))))
-                      (is-generated (lcp.clone:clone-function-definition-for-class ,class-sym)
-                                    (format nil "MyClass Clone() const {
-                                                   MyClass object;
-                                                   ~A
-                                                   return object;
-                                                 }"
-                                            ,expected))))))
       (lcp:define-class klondike ()
         ()
         (:clone))
@@ -922,8 +928,7 @@
                                }
 
                                object.member_ = std::make_pair(std::move(first1), std::move(second1));
-                             }")
-        )
+                             }"))
       (subtest "pointers"
         (single-member-test (member "Klondike *")
                             "object.member_ = member_ ? member_->Clone() : nullptr;")
@@ -939,4 +944,25 @@
         (single-member-test (member :int32_t)
                             "object.member_ = member_;")
         (single-member-test (member :char)
-                            "object.member_ = member_;")))))
+                            "object.member_ = member_;")))
+    (subtest "class copying"
+      (undefine-cpp-types)
+      (lcp:define-class non-copyable-class-1 ()
+        ((counter "int32_t *")))
+      (lcp:define-class cloneable-class ()
+        ((uid :int32_t))
+        (:clone))
+      (lcp:define-class copyable-class-1 ()
+        ((counter "int32_t *" :clone :copy)))
+      (lcp:define-class copyable-class-2 ()
+        ((member :int32_t)))
+      (single-member-test (member "NonCopyableClass1")
+                          lcp.clone:clone-error)
+      (single-member-test (member "CloneableClass")
+                          "object.member_ = member_.Clone();")
+      (single-member-test (member "CopyableClass1")
+                          "object.member_ = member_;")
+      (single-member-test (member "CopyableClass2")
+                          "object.member_ = member_;")
+      (single-member-test (member "UnknownClass")
+                         "object.member_ = member_;"))))

src/query/frontend/ast/ast.lcp

@@ -332,7 +332,14 @@ class AstStorage {
 cpp<#
 
 (lcp:define-class tree ()
-  ((uid :int32_t :scope :public))
+  ((uid :int32_t :scope :public
+        :clone (lambda (source dest)
+                 #>cpp
+                 ${dest} = ${source};
+                 // TODO(mtomic): This is a hack. Adopting existing UIDs breaks everything
+                 // because `AstStorage` keeps a counter for generating when `Create` is called.
+                 storage->max_existing_uid_ = std::max(storage->max_existing_uid_, ${source});
+                 cpp<#)))
   (:abstractp t)
   (:public
     #>cpp

src/query/plan/distributed.cpp

@@ -18,18 +18,9 @@ namespace {
 
 std::pair<std::unique_ptr<LogicalOperator>, AstStorage> Clone(
     const LogicalOperator &original_plan) {
-  // TODO: Add a proper Clone method to LogicalOperator
-  ::capnp::MallocMessageBuilder message;
-  {
-    auto builder = message.initRoot<query::plan::capnp::LogicalOperator>();
-    LogicalOperator::SaveHelper helper;
-    Save(original_plan, &builder, &helper);
-  }
-  auto reader = message.getRoot<query::plan::capnp::LogicalOperator>();
-  std::unique_ptr<LogicalOperator> plan_copy;
-  LogicalOperator::LoadHelper helper;
-  Load(&plan_copy, reader, &helper);
-  return std::make_pair(std::move(plan_copy), std::move(helper.ast_storage));
+  AstStorage storage;
+  std::unique_ptr<LogicalOperator> root_copy = original_plan.Clone(&storage);
+  return std::make_pair(std::move(root_copy), std::move(storage));
 }
 
 int64_t AddWorkerPlan(DistributedPlan &distributed_plan,

src/query/plan/distributed_ops.lcp

@@ -84,7 +84,8 @@ time on data transfer. It gives no guarantees on result order.")
      input_ = input;
    }
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (defun slk-load-pull-remote (member)
   #>cpp
@@ -119,7 +120,16 @@ time on data transfer. It gives no guarantees on result order.")
                 :slk-save #'slk-save-operator-pointer
                 :slk-load #'slk-load-pull-remote
                 :capnp-save #'save-operator-pointer
-                :capnp-load #'load-pull-remote)
+                :capnp-load #'load-pull-remote
+                :clone (lambda (source dest)
+                         #>cpp
+                         if (${source}) {
+                           std::shared_ptr<LogicalOperator> tmp = ${source}->Clone(storage);
+                           ${dest} = std::static_pointer_cast<PullRemote>(tmp);
+                         } else {
+                           ${dest} = nullptr;
+                         }
+                         cpp<#))
    (advance-command :bool :initval "false" :scope :public))
   (:documentation
    "Operator used to synchronize stages of plan execution between the master and
@@ -164,7 +174,8 @@ Logic of the synchronize operator is:
      input_ = input;
    }
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class pull-remote-order-by (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -209,7 +220,8 @@ by having only one result from each worker.")
      input_ = input;
    }
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class distributed-expand (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -242,7 +254,8 @@ by having only one result from each worker.")
      input_ = input;
    }
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class distributed-expand-bfs (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -312,7 +325,8 @@ by having only one result from each worker.")
      input_ = input;
    }
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class distributed-create-node (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -348,7 +362,8 @@ by having only one result from each worker.")
      input_ = input;
    }
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class distributed-create-expand (logical-operator)
   ((node-info "NodeCreationInfo" :scope :public
@@ -395,7 +410,8 @@ by having only one result from each worker.")
      input_ = input;
    }
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:pop-namespace) ;; plan
 (lcp:pop-namespace) ;; query

src/query/plan/operator.lcp

@@ -229,7 +229,9 @@ can serve as inputs to others and thus a sequence of operations is formed.")
          :load-args '((helper "query::plan::LogicalOperator::SlkLoadHelper *")))
    (:capnp :base t
            :save-args '((helper "LogicalOperator::SaveHelper *"))
-           :load-args '((helper "LogicalOperator::LoadHelper *")))))
+           :load-args '((helper "LogicalOperator::LoadHelper *"))))
+  (:clone :args '((storage "AstStorage *"))
+          :base t))
 
 (defun slk-save-ast-pointer (member)
   #>cpp
@@ -371,7 +373,8 @@ and false on every following Pull.")
      bool did_pull_{false};
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (defun slk-save-properties (member)
   #>cpp
@@ -431,7 +434,8 @@ and false on every following Pull.")
   (:serialize (:slk :save-args '((helper "query::plan::LogicalOperator::SaveHelper *"))
                     :load-args '((helper "query::plan::LogicalOperator::SlkLoadHelper *")))
               (:capnp :save-args '((helper "LogicalOperator::SaveHelper *"))
-                      :load-args '((helper "LogicalOperator::LoadHelper *")))))
+                      :load-args '((helper "LogicalOperator::LoadHelper *"))))
+  (:clone :args '((storage "AstStorage *"))))
 
 (lcp:define-class create-node (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -494,7 +498,8 @@ a preceeding `MATCH`), or multiple nodes (`MATCH ... CREATE` or
      const std::unique_ptr<Cursor> input_cursor_;
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-struct edge-creation-info ()
   ((symbol "Symbol")
@@ -516,7 +521,8 @@ a preceeding `MATCH`), or multiple nodes (`MATCH ... CREATE` or
   (:serialize (:slk :save-args '((helper "query::plan::LogicalOperator::SaveHelper *"))
                     :load-args '((helper "query::plan::LogicalOperator::SlkLoadHelper *")))
               (:capnp :save-args '((helper "LogicalOperator::SaveHelper *"))
-                      :load-args '((helper "LogicalOperator::LoadHelper *")))))
+                      :load-args '((helper "LogicalOperator::LoadHelper *"))))
+  (:clone :args '((storage "AstStorage *"))))
 
 (lcp:define-class create-expand (logical-operator)
   ((node-info "NodeCreationInfo" :scope :public
@@ -621,7 +627,8 @@ chained in cases when longer paths need creating.
                      ExpressionEvaluator &evaluator);
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class scan-all (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -675,7 +682,8 @@ with a constructor argument.
      input_ = input;
    }
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class scan-all-by-label (scan-all)
   ((label "storage::Label" :scope :public))
@@ -696,7 +704,8 @@ given label.
    std::unique_ptr<Cursor> MakeCursor(
        database::GraphDbAccessor &db) const override;
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (defun slk-save-optional-bound (member)
   #>cpp
@@ -774,20 +783,35 @@ given label.
                                       load-bound)
              reader member capnp-name)))
 
+(defun clone-optional-bound (source dest)
+  #>cpp
+  if (${source}) {
+    ${dest}.emplace(utils::Bound<Expression *>(
+                    ${source}->value()->Clone(storage),
+                    ${source}->type()));
+  } else {
+    ${dest} = std::experimental::nullopt;
+  }
+  cpp<#)
+
 (lcp:define-class scan-all-by-label-property-range (scan-all)
   ((label "storage::Label" :scope :public)
    (property "storage::Property" :scope :public)
    (property-name "std::string" :scope :public)
    (lower-bound "std::experimental::optional<Bound>" :scope :public
-                :capnp-type "Utils.Optional(Utils.Bound(Ast.Tree))"
                 :slk-save #'slk-save-optional-bound
                 :slk-load #'slk-load-optional-bound
-                :capnp-save #'save-optional-bound :capnp-load #'load-optional-bound)
+                :capnp-save #'save-optional-bound
+                :capnp-load #'load-optional-bound
+                :capnp-type "Utils.Optional(Utils.Bound(Ast.Tree))"
+                :clone #'clone-optional-bound)
    (upper-bound "std::experimental::optional<Bound>" :scope :public
                 :slk-save #'slk-save-optional-bound
                 :slk-load #'slk-load-optional-bound
-                :capnp-save #'save-optional-bound :capnp-load #'load-optional-bound
-                :capnp-type "Utils.Optional(Utils.Bound(Ast.Tree))"))
+                :capnp-save #'save-optional-bound
+                :capnp-load #'load-optional-bound
+                :capnp-type "Utils.Optional(Utils.Bound(Ast.Tree))"
+                :clone #'clone-optional-bound))
   (:documentation
    "Behaves like @c ScanAll, but produces only vertices with given label and
 property value which is inside a range (inclusive or exlusive).
@@ -826,7 +850,8 @@ property value which is inside a range (inclusive or exlusive).
    std::unique_ptr<Cursor> MakeCursor(
        database::GraphDbAccessor &db) const override;
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class scan-all-by-label-property-value (scan-all)
   ((label "storage::Label" :scope :public)
@@ -868,7 +893,8 @@ property value.
    std::unique_ptr<Cursor> MakeCursor(
        database::GraphDbAccessor &db) const override;
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-struct expand-common ()
   (
@@ -989,7 +1015,8 @@ pulled.")
      bool InitEdges(Frame &, ExecutionContext &);
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-struct expansion-lambda ()
   ((inner-edge-symbol "Symbol" :documentation "Currently expanded edge symbol.")
@@ -1028,7 +1055,8 @@ pulled.")
               (:capnp
                :save-args '((saved-ast-uids "std::vector<int32_t> *"))
                :load-args '((ast-storage "AstStorage *")
-                            (loaded-ast-uids "std::vector<int32_t> *")))))
+                            (loaded-ast-uids "std::vector<int32_t> *"))))
+  (:clone :args '((storage "AstStorage *"))))
 
 (lcp:define-class expand-variable (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -1189,7 +1217,8 @@ pulled.")
    friend class ExpandVariableCursor;
    friend class ExpandWeightedShortestPathCursor;
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class construct-named-path (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -1223,7 +1252,8 @@ pulled.")
      input_ = input;
    }
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class filter (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -1273,7 +1303,8 @@ a boolean value.")
      const std::unique_ptr<Cursor> input_cursor_;
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class produce (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -1329,7 +1360,8 @@ RETURN clause) the Produce's pull succeeds exactly once.")
      const std::unique_ptr<Cursor> input_cursor_;
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class delete (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -1382,7 +1414,8 @@ Has a flag for using DETACH DELETE when deleting vertices.")
      const std::unique_ptr<Cursor> input_cursor_;
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class set-property (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -1438,7 +1471,8 @@ can be stored (a TypedValue that can be converted to PropertyValue).")
      const std::unique_ptr<Cursor> input_cursor_;
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class set-properties (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -1511,7 +1545,8 @@ that the old properties are discarded and replaced with new ones.")
      void Set(TRecordAccessor &record, const TypedValue &rhs) const;
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class set-labels (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -1558,7 +1593,8 @@ It does NOT remove labels that are already set on that Vertex.")
      const std::unique_ptr<Cursor> input_cursor_;
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class remove-property (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -1606,7 +1642,8 @@ It does NOT remove labels that are already set on that Vertex.")
      const std::unique_ptr<Cursor> input_cursor_;
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class remove-labels (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -1653,7 +1690,8 @@ If a label does not exist on a Vertex, nothing happens.")
      const std::unique_ptr<Cursor> input_cursor_;
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class edge-uniqueness-filter (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -1712,7 +1750,8 @@ edge lists).")
      const std::unique_ptr<Cursor> input_cursor_;
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class accumulate (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -1786,7 +1825,8 @@ has been cached will be reconstructed before Pull returns.
      bool pulled_all_input_{false};
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 #>cpp
 /**
@@ -1881,7 +1921,8 @@ elements are in an undefined state after aggregation.")
                        :load-args '((helper "query::plan::LogicalOperator::SlkLoadHelper *")))
                  (:capnp
                   :save-args '((helper "LogicalOperator::SaveHelper *"))
-                  :load-args '((helper "LogicalOperator::LoadHelper *")))))
+                  :load-args '((helper "LogicalOperator::LoadHelper *"))))
+     (:clone :args '((storage "AstStorage *"))))
    #>cpp
    Aggregate() = default;
    Aggregate(const std::shared_ptr<LogicalOperator> &input,
@@ -1981,7 +2022,8 @@ elements are in an undefined state after aggregation.")
      void EnsureOkForAvgSum(const TypedValue &value) const;
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class skip (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -2041,7 +2083,8 @@ operator's implementation does not expect this.")
      int skipped_{0};
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class limit (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -2104,7 +2147,8 @@ input should be performed).")
      int pulled_{0};
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class order-by (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -2176,7 +2220,8 @@ are valid for usage after the OrderBy operator.")
      decltype(cache_.begin()) cache_it_ = cache_.begin();
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class merge (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -2249,7 +2294,8 @@ documentation.")
      bool pull_input_{true};
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class optional (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -2311,7 +2357,8 @@ and returns true, once.")
      bool pull_input_{true};
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class unwind (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -2368,7 +2415,8 @@ Input is optional (unwind can be the first clause in a query).")
      std::vector<TypedValue>::iterator input_value_it_ = input_value_.end();
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class distinct (logical-operator)
   ((input "std::shared_ptr<LogicalOperator>" :scope :public
@@ -2427,7 +2475,8 @@ This implementation maintains input ordering.")
          seen_rows_;
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class union (logical-operator)
   ((left-op "std::shared_ptr<LogicalOperator>" :scope :public
@@ -2488,7 +2537,8 @@ of symbols used by each of the inputs.")
      const std::unique_ptr<Cursor> left_cursor_, right_cursor_;
    };
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 ;; TODO: We should probably output this operator in regular planner, not just
 ;; distributed planner.
@@ -2533,12 +2583,13 @@ of symbols used by each of the inputs.")
     std::shared_ptr<LogicalOperator> input() const override;
     void set_input(std::shared_ptr<LogicalOperator>) override;
     cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:define-class output-table (logical-operator)
   ((output-symbols "std::vector<Symbol>" :scope :public :dont-save t)
    (callback "std::function<std::vector<std::vector<TypedValue>>(Frame *, ExecutionContext *)>"
-             :scope :public :dont-save t))
+             :scope :public :dont-save t :clone :copy))
   (:documentation "An operator that outputs a table, producing a single row on each pull")
   (:public
    #>cpp
@@ -2567,7 +2618,8 @@ of symbols used by each of the inputs.")
    std::shared_ptr<LogicalOperator> input() const override;
    void set_input(std::shared_ptr<LogicalOperator> input) override;
    cpp<#)
-  (:serialize (:slk) (:capnp)))
+  (:serialize (:slk) (:capnp))
+  (:clone))
 
 (lcp:pop-namespace) ;; plan
 (lcp:pop-namespace) ;; query

tests/unit/bfs_common.hpp

@@ -127,6 +127,11 @@ class Yield : public query::plan::LogicalOperator {
     LOG(FATAL) << "Please go away, visitor!";
   }
 
+  std::unique_ptr<LogicalOperator> Clone(
+      query::AstStorage *storage) const override {
+    LOG(FATAL) << "Don't clone Yield operator!";
+  }
+
   std::shared_ptr<query::plan::LogicalOperator> input_;
   std::vector<query::Symbol> modified_symbols_;
   std::vector<std::vector<query::TypedValue>> values_;