X

kuzudb · Jul 20, 2023 · 22395ab · 22395ab
1 parent 3b7001a
commit 22395ab
Show file tree

Hide file tree

Showing 5 changed files with 13 additions and 16 deletions.
diff --git a/src/include/processor/operator/result_collector.h b/src/include/processor/operator/result_collector.h
@@ -8,6 +8,9 @@ namespace processor {
 
 class ResultCollectorSharedState {
 public:
+    explicit ResultCollectorSharedState(std::shared_ptr<FactorizedTable> table)
+        : table{std::move(table)} {}
+
     inline void mergeLocalTable(FactorizedTable& localTable) {
         std::unique_lock lck{mtx};
         table->merge(localTable);
@@ -28,7 +31,7 @@ struct ResultCollectorInfo {
         std::unique_ptr<FactorizedTableSchema> tableSchema, std::vector<DataPos> payloadPositions)
         : tableSchema{std::move(tableSchema)}, payloadPositions{std::move(payloadPositions)} {}
     ResultCollectorInfo(const ResultCollectorInfo& other)
-        : tableSchema{other.tableSchema->copy()} {}
+        : tableSchema{other.tableSchema->copy()}, payloadPositions{other.payloadPositions} {}
 
     inline std::unique_ptr<ResultCollectorInfo> copy() const {
         return std::make_unique<ResultCollectorInfo>(*this);

diff --git a/src/processor/mapper/create_result_collector.cpp b/src/processor/mapper/create_result_collector.cpp
@@ -10,7 +10,6 @@ namespace processor {
 std::unique_ptr<ResultCollector> PlanMapper::createResultCollector(
     const binder::expression_vector& expressions, Schema* schema,
     std::unique_ptr<PhysicalOperator> prevOperator) {
-    bool hasUnFlatColumn = false;
     std::vector<DataPos> payloadsPos;
     auto tableSchema = std::make_unique<FactorizedTableSchema>();
     for (auto& expression : expressions) {
@@ -22,15 +21,12 @@ std::unique_ptr<ResultCollector> PlanMapper::createResultCollector(
         } else {
             columnSchema = std::make_unique<ColumnSchema>(
                 true /* isUnFlat */, dataPos.dataChunkPos, (uint32_t)sizeof(overflow_value_t));
-            hasUnFlatColumn = true;
         }
         tableSchema->appendColumn(std::move(columnSchema));
         payloadsPos.push_back(dataPos);
     }
-    //    auto sharedState = std::make_shared<ResultCollectorSharedState>(
-    //        memoryManager, tableSchema->copy(), hasUnFlatColumn ? 1 :
-    //        common::DEFAULT_VECTOR_CAPACITY);
-    auto sharedState = std::make_shared<ResultCollectorSharedState>();
+    auto table = std::make_shared<FactorizedTable>(memoryManager, tableSchema->copy());
+    auto sharedState = std::make_shared<ResultCollectorSharedState>(std::move(table));
     auto info = std::make_unique<ResultCollectorInfo>(std::move(tableSchema), payloadsPos);
     return make_unique<ResultCollector>(std::make_unique<ResultSetDescriptor>(schema),
         std::move(info), std::move(sharedState), std::move(prevOperator), getOperatorID(),

diff --git a/src/processor/mapper/map_cross_product.cpp b/src/processor/mapper/map_cross_product.cpp
@@ -30,8 +30,7 @@ std::unique_ptr<PhysicalOperator> PlanMapper::mapLogicalCrossProductToPhysical(
         std::make_unique<CrossProductInfo>(std::move(outVecPos), std::move(colIndicesToScan));
     auto table = resultCollector->getResultFactorizedTable();
     auto maxMorselSize = table->hasUnflatCol() ? 1 : common::DEFAULT_VECTOR_CAPACITY;
-    auto localState = std::make_unique<CrossProductLocalState>(
-        table, maxMorselSize);
+    auto localState = std::make_unique<CrossProductLocalState>(table, maxMorselSize);
     return make_unique<CrossProduct>(std::move(info), std::move(localState),
         std::move(probeSidePrevOperator), std::move(resultCollector), getOperatorID(),
         logicalCrossProduct->getExpressionsForPrinting());

diff --git a/src/processor/mapper/map_union.cpp b/src/processor/mapper/map_union.cpp
@@ -32,13 +32,12 @@ std::unique_ptr<PhysicalOperator> PlanMapper::mapLogicalUnionAllToPhysical(
         outputPositions.emplace_back(outSchema->getExpressionPos(*expression));
         columnIndices.push_back(i);
     }
-    auto info = std::make_unique<UnionAllScanInfo>(std::move(outputPositions), std::move(columnIndices));
+    auto info =
+        std::make_unique<UnionAllScanInfo>(std::move(outputPositions), std::move(columnIndices));
     auto maxMorselSize = tables[0]->hasUnflatCol() ? 1 : common::DEFAULT_VECTOR_CAPACITY;
-    auto unionSharedState =
-        make_shared<UnionAllScanSharedState>(std::move(tables), maxMorselSize);
-    return make_unique<UnionAllScan>(std::move(info),
-        unionSharedState, std::move(prevOperators), getOperatorID(),
-        logicalUnionAll.getExpressionsForPrinting());
+    auto unionSharedState = make_shared<UnionAllScanSharedState>(std::move(tables), maxMorselSize);
+    return make_unique<UnionAllScan>(std::move(info), unionSharedState, std::move(prevOperators),
+        getOperatorID(), logicalUnionAll.getExpressionsForPrinting());
 }
 
 } // namespace processor

diff --git a/src/processor/operator/table_scan/factorized_table_scan.cpp b/src/processor/operator/table_scan/factorized_table_scan.cpp
@@ -29,4 +29,4 @@ bool FactorizedTableScan::getNextTuplesInternal(ExecutionContext* context) {
 }
 
 } // namespace processor
-} // namespace kuzu
+} // namespace kuzu