diff --git a/src/binder/bind/bind_ddl.cpp b/src/binder/bind/bind_ddl.cpp
index bde38536402..10aeceac381 100644
--- a/src/binder/bind/bind_ddl.cpp
+++ b/src/binder/bind/bind_ddl.cpp
@@ -22,7 +22,7 @@ unique_ptr<BoundStatement> Binder::bindCreateNodeClause(const Statement& stateme
     auto primaryKeyIdx = bindPrimaryKey(
         createNodeClause.getPKColName(), createNodeClause.getPropertyNameDataTypes());
     return make_unique<BoundCreateNodeClause>(
-        tableName, move(boundPropertyNameDataTypes), primaryKeyIdx);
+        tableName, std::move(boundPropertyNameDataTypes), primaryKeyIdx);
 }
 
 unique_ptr<BoundStatement> Binder::bindCreateRelClause(const Statement& statement) {
@@ -40,7 +40,7 @@ unique_ptr<BoundStatement> Binder::bindCreateRelClause(const Statement& statemen
         srcDstTableIDs.emplace_back(bindNodeTableID(srcTableName), bindNodeTableID(dstTableName));
     }
     return make_unique<BoundCreateRelClause>(
-        tableName, move(propertyNameDataTypes), relMultiplicity, srcDstTableIDs);
+        tableName, std::move(propertyNameDataTypes), relMultiplicity, srcDstTableIDs);
 }
 
 unique_ptr<BoundStatement> Binder::bindDropTable(const Statement& statement) {
diff --git a/src/include/binder/bound_statement_result.h b/src/include/binder/bound_statement_result.h
index 9cdb6084364..52270a036f1 100644
--- a/src/include/binder/bound_statement_result.h
+++ b/src/include/binder/bound_statement_result.h
@@ -43,6 +43,12 @@ class BoundStatementResult {
         return result;
     }
 
+    inline shared_ptr<Expression> getSingleExpressionToCollect() {
+        auto expressionsToCollect = getExpressionsToCollect();
+        assert(expressionsToCollect.size() == 1);
+        return expressionsToCollect[0];
+    }
+
     inline unique_ptr<BoundStatementResult> copy() {
         return make_unique<BoundStatementResult>(columns, expressionsToCollectPerColumn);
     }
diff --git a/src/include/planner/logical_plan/logical_operator/logical_create_node_table.h b/src/include/planner/logical_plan/logical_operator/logical_create_node_table.h
index 321ddd9f8a4..3343e00fe04 100644
--- a/src/include/planner/logical_plan/logical_operator/logical_create_node_table.h
+++ b/src/include/planner/logical_plan/logical_operator/logical_create_node_table.h
@@ -8,15 +8,16 @@ namespace planner {
 class LogicalCreateNodeTable : public LogicalCreateTable {
 public:
     LogicalCreateNodeTable(string tableName, vector<PropertyNameDataType> propertyNameDataTypes,
-        uint32_t primaryKeyIdx)
+        uint32_t primaryKeyIdx, shared_ptr<Expression> outputExpression)
         : LogicalCreateTable{LogicalOperatorType::CREATE_NODE_TABLE, std::move(tableName),
-              std::move(propertyNameDataTypes)},
+              std::move(propertyNameDataTypes), std::move(outputExpression)},
           primaryKeyIdx{primaryKeyIdx} {}
 
     inline uint32_t getPrimaryKeyIdx() const { return primaryKeyIdx; }
 
     inline unique_ptr<LogicalOperator> copy() override {
-        return make_unique<LogicalCreateNodeTable>(tableName, propertyNameDataTypes, primaryKeyIdx);
+        return make_unique<LogicalCreateNodeTable>(
+            tableName, propertyNameDataTypes, primaryKeyIdx, outputExpression);
     }
 
 private:
diff --git a/src/include/planner/logical_plan/logical_operator/logical_create_rel_table.h b/src/include/planner/logical_plan/logical_operator/logical_create_rel_table.h
index af0932711c5..793987fc4dc 100644
--- a/src/include/planner/logical_plan/logical_operator/logical_create_rel_table.h
+++ b/src/include/planner/logical_plan/logical_operator/logical_create_rel_table.h
@@ -9,9 +9,10 @@ namespace planner {
 class LogicalCreateRelTable : public LogicalCreateTable {
 public:
     LogicalCreateRelTable(string tableName, vector<PropertyNameDataType> propertyNameDataTypes,
-        RelMultiplicity relMultiplicity, vector<pair<table_id_t, table_id_t>> srcDstTableIDs)
+        RelMultiplicity relMultiplicity, vector<pair<table_id_t, table_id_t>> srcDstTableIDs,
+        shared_ptr<Expression> outputExpression)
         : LogicalCreateTable{LogicalOperatorType::CREATE_REL_TABLE, std::move(tableName),
-              std::move(propertyNameDataTypes)},
+              std::move(propertyNameDataTypes), std::move(outputExpression)},
           relMultiplicity{relMultiplicity}, srcDstTableIDs{std::move(srcDstTableIDs)} {}
 
     inline RelMultiplicity getRelMultiplicity() const { return relMultiplicity; }
@@ -20,7 +21,7 @@ class LogicalCreateRelTable : public LogicalCreateTable {
 
     inline unique_ptr<LogicalOperator> copy() override {
         return make_unique<LogicalCreateRelTable>(
-            tableName, propertyNameDataTypes, relMultiplicity, srcDstTableIDs);
+            tableName, propertyNameDataTypes, relMultiplicity, srcDstTableIDs, outputExpression);
     }
 
 private:
diff --git a/src/include/planner/logical_plan/logical_operator/logical_create_table.h b/src/include/planner/logical_plan/logical_operator/logical_create_table.h
index 9a0d4b639b9..a78af92a18d 100644
--- a/src/include/planner/logical_plan/logical_operator/logical_create_table.h
+++ b/src/include/planner/logical_plan/logical_operator/logical_create_table.h
@@ -8,9 +8,9 @@ namespace planner {
 class LogicalCreateTable : public LogicalDDL {
 public:
     LogicalCreateTable(LogicalOperatorType operatorType, string tableName,
-        vector<PropertyNameDataType> propertyNameDataTypes)
-        : LogicalDDL{operatorType, std::move(tableName)}, propertyNameDataTypes{
-                                                              std::move(propertyNameDataTypes)} {}
+        vector<PropertyNameDataType> propertyNameDataTypes, shared_ptr<Expression> outputExpression)
+        : LogicalDDL{operatorType, std::move(tableName), std::move(outputExpression)},
+          propertyNameDataTypes{std::move(propertyNameDataTypes)} {}
 
     inline vector<PropertyNameDataType> getPropertyNameDataTypes() const {
         return propertyNameDataTypes;
diff --git a/src/include/planner/logical_plan/logical_operator/logical_ddl.h b/src/include/planner/logical_plan/logical_operator/logical_ddl.h
index 4d45668bb4c..0aff5b51f31 100644
--- a/src/include/planner/logical_plan/logical_operator/logical_ddl.h
+++ b/src/include/planner/logical_plan/logical_operator/logical_ddl.h
@@ -7,17 +7,26 @@ namespace planner {
 
 class LogicalDDL : public LogicalOperator {
 public:
-    LogicalDDL(LogicalOperatorType operatorType, string tableName)
-        : LogicalOperator{operatorType}, tableName{std::move(tableName)} {}
+    LogicalDDL(
+        LogicalOperatorType operatorType, string tableName, shared_ptr<Expression> outputExpression)
+        : LogicalOperator{operatorType}, tableName{std::move(tableName)},
+          outputExpression{std::move(outputExpression)} {}
 
     inline string getTableName() const { return tableName; }
+    inline shared_ptr<Expression> getOutputExpression() const { return outputExpression; }
 
     inline string getExpressionsForPrinting() const override { return tableName; }
 
-    inline void computeSchema() override { schema = make_unique<Schema>(); }
+    inline void computeSchema() override {
+        schema = make_unique<Schema>();
+        auto groupPos = schema->createGroup();
+        schema->insertToGroupAndScope(outputExpression, groupPos);
+        schema->setGroupAsSingleState(groupPos);
+    }
 
 protected:
     string tableName;
+    shared_ptr<Expression> outputExpression;
 };
 
 } // namespace planner
diff --git a/src/include/planner/logical_plan/logical_operator/logical_drop_property.h b/src/include/planner/logical_plan/logical_operator/logical_drop_property.h
index cc531000103..f913d8e2371 100644
--- a/src/include/planner/logical_plan/logical_operator/logical_drop_property.h
+++ b/src/include/planner/logical_plan/logical_operator/logical_drop_property.h
@@ -6,18 +6,19 @@ namespace kuzu {
 namespace planner {
 
 class LogicalDropProperty : public LogicalDDL {
-
 public:
-    explicit LogicalDropProperty(table_id_t tableID, property_id_t propertyID, string tableName)
-        : LogicalDDL{LogicalOperatorType::DROP_PROPERTY, std::move(tableName)}, tableID{tableID},
-          propertyID{propertyID} {}
+    explicit LogicalDropProperty(table_id_t tableID, property_id_t propertyID, string tableName,
+        shared_ptr<Expression> outputExpression)
+        : LogicalDDL{LogicalOperatorType::DROP_PROPERTY, std::move(tableName),
+              std::move(outputExpression)},
+          tableID{tableID}, propertyID{propertyID} {}
 
     inline table_id_t getTableID() const { return tableID; }
 
     inline property_id_t getPropertyID() const { return propertyID; }
 
     inline unique_ptr<LogicalOperator> copy() override {
-        return make_unique<LogicalDropProperty>(tableID, propertyID, tableName);
+        return make_unique<LogicalDropProperty>(tableID, propertyID, tableName, outputExpression);
     }
 
 private:
diff --git a/src/include/planner/logical_plan/logical_operator/logical_drop_table.h b/src/include/planner/logical_plan/logical_operator/logical_drop_table.h
index 265542f246a..42d998f257d 100644
--- a/src/include/planner/logical_plan/logical_operator/logical_drop_table.h
+++ b/src/include/planner/logical_plan/logical_operator/logical_drop_table.h
@@ -6,15 +6,17 @@ namespace kuzu {
 namespace planner {
 
 class LogicalDropTable : public LogicalDDL {
-
 public:
-    explicit LogicalDropTable(table_id_t tableID, string tableName)
-        : LogicalDDL{LogicalOperatorType::DROP_TABLE, std::move(tableName)}, tableID{tableID} {}
+    explicit LogicalDropTable(
+        table_id_t tableID, string tableName, shared_ptr<Expression> outputExpression)
+        : LogicalDDL{LogicalOperatorType::DROP_TABLE, std::move(tableName),
+              std::move(outputExpression)},
+          tableID{tableID} {}
 
     inline table_id_t getTableID() const { return tableID; }
 
     inline unique_ptr<LogicalOperator> copy() override {
-        return make_unique<LogicalDropTable>(tableID, tableName);
+        return make_unique<LogicalDropTable>(tableID, tableName, outputExpression);
     }
 
 private:
diff --git a/src/include/processor/mapper/plan_mapper.h b/src/include/processor/mapper/plan_mapper.h
index bec4fb2ce98..835d8c74306 100644
--- a/src/include/processor/mapper/plan_mapper.h
+++ b/src/include/processor/mapper/plan_mapper.h
@@ -1,6 +1,7 @@
 #pragma once
 
 #include "binder/expression/node_expression.h"
+#include "common/statement_type.h"
 #include "planner/logical_plan/logical_plan.h"
 #include "processor/mapper/expression_mapper.h"
 #include "processor/operator/result_collector.h"
diff --git a/src/include/processor/operator/ddl/create_node_table.h b/src/include/processor/operator/ddl/create_node_table.h
index d86f37e6289..6bf145f611e 100644
--- a/src/include/processor/operator/ddl/create_node_table.h
+++ b/src/include/processor/operator/ddl/create_node_table.h
@@ -9,23 +9,25 @@ namespace processor {
 class CreateNodeTable : public CreateTable {
 public:
     CreateNodeTable(Catalog* catalog, string tableName,
-        vector<PropertyNameDataType> propertyNameDataTypes, uint32_t primaryKeyIdx, uint32_t id,
-        const string& paramsString, NodesStatisticsAndDeletedIDs* nodesStatisticsAndDeletedIDs)
+        vector<PropertyNameDataType> propertyNameDataTypes, uint32_t primaryKeyIdx,
+        const DataPos& outputPos, uint32_t id, const string& paramsString,
+        NodesStatisticsAndDeletedIDs* nodesStatistics)
         : CreateTable{PhysicalOperatorType::CREATE_NODE_TABLE, catalog, std::move(tableName),
-              std::move(propertyNameDataTypes), id, paramsString},
-          primaryKeyIdx{primaryKeyIdx}, nodesStatisticsAndDeletedIDs{nodesStatisticsAndDeletedIDs} {
-    }
+              std::move(propertyNameDataTypes), outputPos, id, paramsString},
+          primaryKeyIdx{primaryKeyIdx}, nodesStatistics{nodesStatistics} {}
+
+    void executeDDLInternal() override;
 
-    string execute() override;
+    std::string getOutputMsg() override;
 
     unique_ptr<PhysicalOperator> clone() override {
         return make_unique<CreateNodeTable>(catalog, tableName, propertyNameDataTypes,
-            primaryKeyIdx, id, paramsString, nodesStatisticsAndDeletedIDs);
+            primaryKeyIdx, outputPos, id, paramsString, nodesStatistics);
     }
 
 private:
     uint32_t primaryKeyIdx;
-    NodesStatisticsAndDeletedIDs* nodesStatisticsAndDeletedIDs;
+    NodesStatisticsAndDeletedIDs* nodesStatistics;
 };
 
 } // namespace processor
diff --git a/src/include/processor/operator/ddl/create_rel_table.h b/src/include/processor/operator/ddl/create_rel_table.h
index 7ac374daed2..f7dd6e6c832 100644
--- a/src/include/processor/operator/ddl/create_rel_table.h
+++ b/src/include/processor/operator/ddl/create_rel_table.h
@@ -10,18 +10,20 @@ class CreateRelTable : public CreateTable {
 public:
     CreateRelTable(Catalog* catalog, string tableName,
         vector<PropertyNameDataType> propertyNameDataTypes, RelMultiplicity relMultiplicity,
-        vector<pair<table_id_t, table_id_t>> srcDstTableIDs, uint32_t id,
+        vector<pair<table_id_t, table_id_t>> srcDstTableIDs, const DataPos& outputPos, uint32_t id,
         const string& paramsString, RelsStatistics* relsStatistics)
         : CreateTable{PhysicalOperatorType::CREATE_REL_TABLE, catalog, std::move(tableName),
-              std::move(propertyNameDataTypes), id, paramsString},
+              std::move(propertyNameDataTypes), outputPos, id, paramsString},
           relMultiplicity{relMultiplicity}, srcDstTableIDs{std::move(srcDstTableIDs)},
           relsStatistics{relsStatistics} {}
 
-    string execute() override;
+    void executeDDLInternal() override;
+
+    std::string getOutputMsg() override;
 
     unique_ptr<PhysicalOperator> clone() override {
         return make_unique<CreateRelTable>(catalog, tableName, propertyNameDataTypes,
-            relMultiplicity, srcDstTableIDs, id, paramsString, relsStatistics);
+            relMultiplicity, srcDstTableIDs, outputPos, id, paramsString, relsStatistics);
     }
 
 private:
diff --git a/src/include/processor/operator/ddl/create_table.h b/src/include/processor/operator/ddl/create_table.h
index 0868bd75b9d..d8d372c09d6 100644
--- a/src/include/processor/operator/ddl/create_table.h
+++ b/src/include/processor/operator/ddl/create_table.h
@@ -8,12 +8,17 @@ namespace processor {
 class CreateTable : public DDL {
 public:
     CreateTable(PhysicalOperatorType operatorType, Catalog* catalog, string tableName,
-        vector<PropertyNameDataType> propertyNameDataTypes, uint32_t id, const string& paramsString)
-        : DDL{operatorType, catalog, id, paramsString}, tableName{std::move(tableName)},
-          propertyNameDataTypes{move(propertyNameDataTypes)} {}
+        vector<PropertyNameDataType> propertyNameDataTypes, const DataPos& outputPos, uint32_t id,
+        const string& paramsString)
+        : DDL{operatorType, catalog, outputPos, id, paramsString}, tableName{std::move(tableName)},
+          propertyNameDataTypes{std::move(propertyNameDataTypes)} {}
 
     virtual ~CreateTable() override = default;
 
+    void executeDDLInternal() override = 0;
+
+    std::string getOutputMsg() override = 0;
+
 protected:
     string tableName;
     vector<PropertyNameDataType> propertyNameDataTypes;
diff --git a/src/include/processor/operator/ddl/ddl.h b/src/include/processor/operator/ddl/ddl.h
index 10ce132ce78..47aaaf97451 100644
--- a/src/include/processor/operator/ddl/ddl.h
+++ b/src/include/processor/operator/ddl/ddl.h
@@ -12,21 +12,28 @@ namespace processor {
 
 class DDL : public PhysicalOperator {
 public:
-    DDL(PhysicalOperatorType operatorType, Catalog* catalog, uint32_t id,
+    DDL(PhysicalOperatorType operatorType, Catalog* catalog, const DataPos& outputPos, uint32_t id,
         const string& paramsString)
-        : PhysicalOperator{operatorType, id, paramsString}, catalog{catalog} {}
+        : PhysicalOperator{operatorType, id, paramsString}, catalog{catalog}, outputPos{outputPos} {
+    }
     virtual ~DDL() override = default;
 
     inline bool isSource() const override { return true; }
 
-    virtual string execute() = 0;
+    void initLocalStateInternal(ResultSet* resultSet, ExecutionContext* context) override;
 
-    bool getNextTuplesInternal() override {
-        throw InternalException("getNextTupleInternal() should not be called on DDL operator.");
-    }
+    bool getNextTuplesInternal() override;
+
+protected:
+    virtual std::string getOutputMsg() = 0;
+    virtual void executeDDLInternal() = 0;
 
 protected:
     Catalog* catalog;
+    DataPos outputPos;
+    ValueVector* outputVector;
+
+    bool hasExecuted = false;
 };
 
 } // namespace processor
diff --git a/src/include/processor/operator/ddl/drop_property.h b/src/include/processor/operator/ddl/drop_property.h
index 330fb7a6959..2d0726728cf 100644
--- a/src/include/processor/operator/ddl/drop_property.h
+++ b/src/include/processor/operator/ddl/drop_property.h
@@ -11,15 +11,18 @@ namespace processor {
 class DropProperty : public DDL {
 public:
     DropProperty(Catalog* catalog, table_id_t tableID, property_id_t propertyID,
-        StorageManager& storageManager, uint32_t id, const string& paramsString)
-        : DDL{PhysicalOperatorType::DROP_PROPERTY, catalog, id, paramsString}, tableID{tableID},
-          propertyID{propertyID}, storageManager{storageManager} {}
+        StorageManager& storageManager, const DataPos& outputPos, uint32_t id,
+        const string& paramsString)
+        : DDL{PhysicalOperatorType::DROP_PROPERTY, catalog, outputPos, id, paramsString},
+          tableID{tableID}, propertyID{propertyID}, storageManager{storageManager} {}
 
-    string execute() override;
+    void executeDDLInternal() override;
+
+    std::string getOutputMsg() override;
 
     unique_ptr<PhysicalOperator> clone() override {
         return make_unique<DropProperty>(
-            catalog, tableID, propertyID, storageManager, id, paramsString);
+            catalog, tableID, propertyID, storageManager, outputPos, id, paramsString);
     }
 
 protected:
diff --git a/src/include/processor/operator/ddl/drop_table.h b/src/include/processor/operator/ddl/drop_table.h
index 039d4af788d..267cd2ab644 100644
--- a/src/include/processor/operator/ddl/drop_table.h
+++ b/src/include/processor/operator/ddl/drop_table.h
@@ -10,20 +10,18 @@ namespace processor {
 
 class DropTable : public DDL {
 public:
-    DropTable(Catalog* catalog, table_id_t tableID, StorageManager& storageManager, uint32_t id,
-        const string& paramsString)
-        : DDL{PhysicalOperatorType::DROP_TABLE, catalog, id, paramsString}, tableID{tableID},
-          storageManager{storageManager} {}
-
-    string execute() override {
-        auto tableSchema = catalog->getReadOnlyVersion()->getTableSchema(tableID);
-        catalog->removeTableSchema(tableSchema);
-        return StringUtils::string_format("%sTable: %s has been dropped.",
-            tableSchema->isNodeTable ? "Node" : "Rel", tableSchema->tableName.c_str());
-    }
+    DropTable(Catalog* catalog, table_id_t tableID, StorageManager& storageManager,
+        const DataPos& outputPos, uint32_t id, const string& paramsString)
+        : DDL{PhysicalOperatorType::DROP_TABLE, catalog, outputPos, id, paramsString},
+          tableID{tableID}, storageManager{storageManager} {}
+
+    void executeDDLInternal() override;
+
+    std::string getOutputMsg() override;
 
     unique_ptr<PhysicalOperator> clone() override {
-        return make_unique<DropTable>(catalog, tableID, storageManager, id, paramsString);
+        return make_unique<DropTable>(
+            catalog, tableID, storageManager, outputPos, id, paramsString);
     }
 
 protected:
diff --git a/src/include/processor/operator/physical_operator.h b/src/include/processor/operator/physical_operator.h
index 0f6d9943cd2..b2955d0ced9 100644
--- a/src/include/processor/operator/physical_operator.h
+++ b/src/include/processor/operator/physical_operator.h
@@ -56,6 +56,7 @@ enum class PhysicalOperatorType : uint8_t {
 class PhysicalOperatorUtils {
 public:
     static std::string operatorTypeToString(PhysicalOperatorType operatorType);
+    static bool isDDLOperator(PhysicalOperatorType operatorType);
 };
 
 struct OperatorMetrics {
diff --git a/src/include/processor/physical_plan.h b/src/include/processor/physical_plan.h
index 782560d15ca..8d5f37b5154 100644
--- a/src/include/processor/physical_plan.h
+++ b/src/include/processor/physical_plan.h
@@ -2,7 +2,6 @@
 
 #include <memory>
 
-#include "common/statement_type.h"
 #include "processor/operator/physical_operator.h"
 
 namespace kuzu {
@@ -10,16 +9,16 @@ namespace processor {
 
 class PhysicalPlan {
 public:
-    explicit PhysicalPlan(
-        unique_ptr<PhysicalOperator> lastOperator, common::StatementType statementType)
-        : lastOperator{std::move(lastOperator)}, statementType{statementType} {}
+    explicit PhysicalPlan(unique_ptr<PhysicalOperator> lastOperator)
+        : lastOperator{std::move(lastOperator)} {}
 
-    inline bool isDDL() const { return StatementTypeUtils::isDDL(statementType); }
-    inline bool isCopyCSV() const { return StatementTypeUtils::isCopyCSV(statementType); }
+    inline bool isCopyCSV() const {
+        return lastOperator->getOperatorType() == PhysicalOperatorType::COPY_NODE_CSV ||
+               lastOperator->getOperatorType() == PhysicalOperatorType::COPY_REL_CSV;
+    }
 
 public:
     unique_ptr<PhysicalOperator> lastOperator;
-    common::StatementType statementType;
 };
 
 class PhysicalPlanUtil {
diff --git a/src/planner/planner.cpp b/src/planner/planner.cpp
index 44f79827a05..588dad12fb5 100644
--- a/src/planner/planner.cpp
+++ b/src/planner/planner.cpp
@@ -54,7 +54,8 @@ unique_ptr<LogicalPlan> Planner::planCreateNodeTable(const BoundStatement& state
     auto& createNodeClause = (BoundCreateNodeClause&)statement;
     auto plan = make_unique<LogicalPlan>();
     auto createNodeTable = make_shared<LogicalCreateNodeTable>(createNodeClause.getTableName(),
-        createNodeClause.getPropertyNameDataTypes(), createNodeClause.getPrimaryKeyIdx());
+        createNodeClause.getPropertyNameDataTypes(), createNodeClause.getPrimaryKeyIdx(),
+        statement.getStatementResult()->getSingleExpressionToCollect());
     createNodeTable->computeSchema();
     plan->setLastOperator(std::move(createNodeTable));
     return plan;
@@ -65,7 +66,8 @@ unique_ptr<LogicalPlan> Planner::planCreateRelTable(const BoundStatement& statem
     auto plan = make_unique<LogicalPlan>();
     auto createRelTable = make_shared<LogicalCreateRelTable>(createRelClause.getTableName(),
         createRelClause.getPropertyNameDataTypes(), createRelClause.getRelMultiplicity(),
-        createRelClause.getSrcDstTableIDs());
+        createRelClause.getSrcDstTableIDs(),
+        statement.getStatementResult()->getSingleExpressionToCollect());
     createRelTable->computeSchema();
     plan->setLastOperator(std::move(createRelTable));
     return plan;
@@ -75,7 +77,8 @@ unique_ptr<LogicalPlan> Planner::planDropTable(const BoundStatement& statement)
     auto& dropTableClause = (BoundDropTable&)statement;
     auto plan = make_unique<LogicalPlan>();
     auto dropTable =
-        make_shared<LogicalDropTable>(dropTableClause.getTableID(), dropTableClause.getTableName());
+        make_shared<LogicalDropTable>(dropTableClause.getTableID(), dropTableClause.getTableName(),
+            statement.getStatementResult()->getSingleExpressionToCollect());
     dropTable->computeSchema();
     plan->setLastOperator(std::move(dropTable));
     return plan;
@@ -85,7 +88,8 @@ unique_ptr<LogicalPlan> Planner::planDropProperty(const BoundStatement& statemen
     auto& dropPropertyClause = (BoundDropProperty&)statement;
     auto plan = make_unique<LogicalPlan>();
     auto dropProperty = make_shared<LogicalDropProperty>(dropPropertyClause.getTableID(),
-        dropPropertyClause.getPropertyID(), dropPropertyClause.getTableName());
+        dropPropertyClause.getPropertyID(), dropPropertyClause.getTableName(),
+        statement.getStatementResult()->getSingleExpressionToCollect());
     dropProperty->computeSchema();
     plan->setLastOperator(std::move(dropProperty));
     return plan;
diff --git a/src/processor/mapper/map_ddl.cpp b/src/processor/mapper/map_ddl.cpp
index 15ebaf3c4df..5adb302d02d 100644
--- a/src/processor/mapper/map_ddl.cpp
+++ b/src/processor/mapper/map_ddl.cpp
@@ -14,12 +14,19 @@
 namespace kuzu {
 namespace processor {
 
+static DataPos getOutputPos(LogicalDDL* logicalDDL) {
+    auto outSchema = logicalDDL->getSchema();
+    auto outputExpression = logicalDDL->getOutputExpression();
+    return DataPos(outSchema->getExpressionPos(*outputExpression));
+}
+
 unique_ptr<PhysicalOperator> PlanMapper::mapLogicalCreateNodeTableToPhysical(
     LogicalOperator* logicalOperator) {
     auto createNodeTable = (LogicalCreateNodeTable*)logicalOperator;
     return make_unique<CreateNodeTable>(catalog, createNodeTable->getTableName(),
         createNodeTable->getPropertyNameDataTypes(), createNodeTable->getPrimaryKeyIdx(),
-        getOperatorID(), createNodeTable->getExpressionsForPrinting(),
+        getOutputPos(createNodeTable), getOperatorID(),
+        createNodeTable->getExpressionsForPrinting(),
         &storageManager.getNodesStore().getNodesStatisticsAndDeletedIDs());
 }
 
@@ -28,7 +35,7 @@ unique_ptr<PhysicalOperator> PlanMapper::mapLogicalCreateRelTableToPhysical(
     auto createRelTable = (LogicalCreateRelTable*)logicalOperator;
     return make_unique<CreateRelTable>(catalog, createRelTable->getTableName(),
         createRelTable->getPropertyNameDataTypes(), createRelTable->getRelMultiplicity(),
-        createRelTable->getSrcDstTableIDs(), getOperatorID(),
+        createRelTable->getSrcDstTableIDs(), getOutputPos(createRelTable), getOperatorID(),
         createRelTable->getExpressionsForPrinting(),
         &storageManager.getRelsStore().getRelsStatistics());
 }
@@ -53,15 +60,15 @@ unique_ptr<PhysicalOperator> PlanMapper::mapLogicalCopyCSVToPhysical(
 unique_ptr<PhysicalOperator> PlanMapper::mapLogicalDropTableToPhysical(
     LogicalOperator* logicalOperator) {
     auto dropTable = (LogicalDropTable*)logicalOperator;
-    return make_unique<DropTable>(catalog, dropTable->getTableID(), storageManager, getOperatorID(),
-        dropTable->getExpressionsForPrinting());
+    return make_unique<DropTable>(catalog, dropTable->getTableID(), storageManager,
+        getOutputPos(dropTable), getOperatorID(), dropTable->getExpressionsForPrinting());
 }
 
 unique_ptr<PhysicalOperator> PlanMapper::mapLogicalDropPropertyToPhysical(
     LogicalOperator* logicalOperator) {
     auto dropProperty = (LogicalDropProperty*)logicalOperator;
     return make_unique<DropProperty>(catalog, dropProperty->getTableID(),
-        dropProperty->getPropertyID(), storageManager, getOperatorID(),
+        dropProperty->getPropertyID(), storageManager, getOutputPos(dropProperty), getOperatorID(),
         dropProperty->getExpressionsForPrinting());
 }
 
diff --git a/src/processor/mapper/plan_mapper.cpp b/src/processor/mapper/plan_mapper.cpp
index 6ab30f899f2..6500e2e1c3d 100644
--- a/src/processor/mapper/plan_mapper.cpp
+++ b/src/processor/mapper/plan_mapper.cpp
@@ -13,11 +13,11 @@ namespace processor {
 unique_ptr<PhysicalPlan> PlanMapper::mapLogicalPlanToPhysical(LogicalPlan* logicalPlan,
     const expression_vector& expressionsToCollect, common::StatementType statementType) {
     auto lastOperator = mapLogicalOperatorToPhysical(logicalPlan->getLastOperator());
-    if (!StatementTypeUtils::isDDLOrCopyCSV(statementType)) {
+    if (!StatementTypeUtils::isCopyCSV(statementType)) {
         lastOperator = appendResultCollector(
             expressionsToCollect, *logicalPlan->getSchema(), std::move(lastOperator));
     }
-    return make_unique<PhysicalPlan>(std::move(lastOperator), statementType);
+    return make_unique<PhysicalPlan>(std::move(lastOperator));
 }
 
 unique_ptr<PhysicalOperator> PlanMapper::mapLogicalOperatorToPhysical(
diff --git a/src/processor/operator/ddl/CMakeLists.txt b/src/processor/operator/ddl/CMakeLists.txt
index 98f2b21e0ef..8c348b8a6a9 100644
--- a/src/processor/operator/ddl/CMakeLists.txt
+++ b/src/processor/operator/ddl/CMakeLists.txt
@@ -2,7 +2,10 @@ add_library(kuzu_processor_operator_ddl
         OBJECT
         create_node_table.cpp
         create_rel_table.cpp
-        drop_property.cpp)
+        ddl.cpp
+        drop_property.cpp
+        drop_table.cpp
+        )
 
 set(ALL_OBJECT_FILES
         ${ALL_OBJECT_FILES} $<TARGET_OBJECTS:kuzu_processor_operator_ddl>
diff --git a/src/processor/operator/ddl/create_node_table.cpp b/src/processor/operator/ddl/create_node_table.cpp
index 32e7a7ba3e8..06e7b95a2ae 100644
--- a/src/processor/operator/ddl/create_node_table.cpp
+++ b/src/processor/operator/ddl/create_node_table.cpp
@@ -3,10 +3,13 @@
 namespace kuzu {
 namespace processor {
 
-string CreateNodeTable::execute() {
+void CreateNodeTable::executeDDLInternal() {
     auto newTableID = catalog->addNodeTableSchema(tableName, primaryKeyIdx, propertyNameDataTypes);
-    nodesStatisticsAndDeletedIDs->addNodeStatisticsAndDeletedIDs(
+    nodesStatistics->addNodeStatisticsAndDeletedIDs(
         catalog->getWriteVersion()->getNodeTableSchema(newTableID));
+}
+
+std::string CreateNodeTable::getOutputMsg() {
     return StringUtils::string_format("NodeTable: %s has been created.", tableName.c_str());
 }
 
diff --git a/src/processor/operator/ddl/create_rel_table.cpp b/src/processor/operator/ddl/create_rel_table.cpp
index 0d3750574fe..874b14c2265 100644
--- a/src/processor/operator/ddl/create_rel_table.cpp
+++ b/src/processor/operator/ddl/create_rel_table.cpp
@@ -3,10 +3,13 @@
 namespace kuzu {
 namespace processor {
 
-string CreateRelTable::execute() {
+void CreateRelTable::executeDDLInternal() {
     auto newRelTableID = catalog->addRelTableSchema(
         tableName, relMultiplicity, propertyNameDataTypes, srcDstTableIDs);
     relsStatistics->addTableStatistic(catalog->getWriteVersion()->getRelTableSchema(newRelTableID));
+}
+
+std::string CreateRelTable::getOutputMsg() {
     return StringUtils::string_format("RelTable: %s has been created.", tableName.c_str());
 }
 
diff --git a/src/processor/operator/ddl/ddl.cpp b/src/processor/operator/ddl/ddl.cpp
index 0e5b66c6351..c2f5defddce 100644
--- a/src/processor/operator/ddl/ddl.cpp
+++ b/src/processor/operator/ddl/ddl.cpp
@@ -4,7 +4,17 @@ namespace kuzu {
 namespace processor {
 
 void DDL::initLocalStateInternal(ResultSet* resultSet, ExecutionContext* context) {
-    auto
+    outputVector = resultSet->getValueVector(outputPos).get();
+}
+
+bool DDL::getNextTuplesInternal() {
+    if (hasExecuted) {
+        return false;
+    }
+    hasExecuted = true;
+    executeDDLInternal();
+    outputVector->setValue<std::string>(0, getOutputMsg());
+    return true;
 }
 
 } // namespace processor
diff --git a/src/processor/operator/ddl/drop_property.cpp b/src/processor/operator/ddl/drop_property.cpp
index cd6fa2f9e66..addb3ff6680 100644
--- a/src/processor/operator/ddl/drop_property.cpp
+++ b/src/processor/operator/ddl/drop_property.cpp
@@ -3,14 +3,14 @@
 namespace kuzu {
 namespace processor {
 
-string DropProperty::execute() {
+void DropProperty::executeDDLInternal() {
     catalog->initCatalogContentForWriteTrxIfNecessary();
     auto tableSchema = catalog->getWriteVersion()->getTableSchema(tableID);
-    auto tableName = tableSchema->tableName;
-    auto propertyName = tableSchema->getPropertyName(propertyID);
     tableSchema->removeProperty(propertyID);
-    return StringUtils::string_format(
-        "%s column of table %s has been dropped.", propertyName.c_str(), tableName.c_str());
+}
+
+std::string DropProperty::getOutputMsg() {
+    return {"Drop Succeed."};
 }
 
 } // namespace processor
diff --git a/src/processor/operator/ddl/drop_table.cpp b/src/processor/operator/ddl/drop_table.cpp
new file mode 100644
index 00000000000..caaee8f6a16
--- /dev/null
+++ b/src/processor/operator/ddl/drop_table.cpp
@@ -0,0 +1,18 @@
+#include "processor/operator/ddl/drop_table.h"
+
+namespace kuzu {
+namespace processor {
+
+void DropTable::executeDDLInternal() {
+    auto tableSchema = catalog->getReadOnlyVersion()->getTableSchema(tableID);
+    catalog->removeTableSchema(tableSchema);
+}
+
+std::string DropTable::getOutputMsg() {
+    auto tableSchema = catalog->getReadOnlyVersion()->getTableSchema(tableID);
+    return StringUtils::string_format("%sTable: %s has been dropped.",
+        tableSchema->isNodeTable ? "Node" : "Rel", tableSchema->tableName.c_str());
+}
+
+} // namespace processor
+} // namespace kuzu
\ No newline at end of file
diff --git a/src/processor/processor.cpp b/src/processor/processor.cpp
index 07fa74b07dd..998618c626b 100644
--- a/src/processor/processor.cpp
+++ b/src/processor/processor.cpp
@@ -22,10 +22,6 @@ shared_ptr<FactorizedTable> QueryProcessor::execute(
         auto copyCSV = (CopyCSV*)physicalPlan->lastOperator.get();
         auto outputMsg = copyCSV->execute(taskScheduler.get(), context);
         return getFactorizedTableForOutputMsg(outputMsg, context->memoryManager);
-    } else if (physicalPlan->isDDL()) {
-        auto ddl = (DDL*)physicalPlan->lastOperator.get();
-        auto outputMsg = ddl->execute();
-        return getFactorizedTableForOutputMsg(outputMsg, context->memoryManager);
     } else {
         auto lastOperator = physicalPlan->lastOperator.get();
         // Init global state before decompose into pipelines. Otherwise, each pipeline will try to
@@ -69,6 +65,13 @@ void QueryProcessor::decomposePlanIntoTasks(
         // in single-thread mode.
     case PhysicalOperatorType::INDEX_SCAN: {
         parentTask->setSingleThreadedTask();
+    } break;
+        // DDL should be executed exactly once.
+    case PhysicalOperatorType::CREATE_NODE_TABLE:
+    case PhysicalOperatorType::CREATE_REL_TABLE:
+    case PhysicalOperatorType::DROP_TABLE:
+    case PhysicalOperatorType::DROP_PROPERTY: {
+        parentTask->setSingleThreadedTask();
     } break;
     default:
         break;