Change copy morsel size back to DEFAULT_VECTOR_CAPACITY

kuzudb · Sep 6, 2023 · 37191a7 · 37191a7
1 parent bf7414f
commit 37191a7
Show file tree

Hide file tree

Showing 22 changed files with 155 additions and 123 deletions.
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -1,6 +1,6 @@
 cmake_minimum_required(VERSION 3.11)
 
-project(Kuzu VERSION 0.0.8.3 LANGUAGES CXX)
+project(Kuzu VERSION 0.0.8.4 LANGUAGES CXX)
 
 find_package(Threads REQUIRED)
 

diff --git a/src/binder/bind/bind_copy.cpp b/src/binder/bind/bind_copy.cpp
@@ -82,15 +82,15 @@ expression_vector Binder::bindColumnExpressions(TableSchema* tableSchema) {
     return columnExpressions;
 }
 
-bool Binder::bindPreservingOrder(TableSchema* tableSchema, CopyDescription::FileType fileType) {
-    bool preservingOrder = fileType == CopyDescription::FileType::CSV;
+bool Binder::bindContainsSerial(TableSchema* tableSchema, CopyDescription::FileType fileType) {
+    bool containsSerial = false;
     for (auto& property : tableSchema->properties) {
         if (property->getDataType()->getLogicalTypeID() == LogicalTypeID::SERIAL) {
-            preservingOrder = true;
+            containsSerial = true;
             break;
         }
     }
-    return preservingOrder;
+    return containsSerial;
 }
 
 std::unique_ptr<BoundStatement> Binder::bindCopyFromClause(const Statement& statement) {
@@ -114,7 +114,7 @@ std::unique_ptr<BoundStatement> Binder::bindCopyFromClause(const Statement& stat
     }
     // Bind execution mode.
     // For CSV file, and table with SERIAL columns, we need to read in serial from files.
-    auto preservingOrder = bindPreservingOrder(tableSchema, actualFileType);
+    auto containsSerial = bindContainsSerial(tableSchema, actualFileType);
     // Bind expressions.
     auto columnExpressions = bindColumnExpressions(tableSchema);
     auto copyDescription = std::make_unique<CopyDescription>(
@@ -131,7 +131,7 @@ std::unique_ptr<BoundStatement> Binder::bindCopyFromClause(const Statement& stat
                                    nullptr;
     auto boundCopyFromInfo = std::make_unique<BoundCopyFromInfo>(std::move(copyDescription),
         tableSchema, std::move(columnExpressions), std::move(nodeOffsetExpression),
-        std::move(boundOffsetExpression), std::move(nbrOffsetExpression), preservingOrder);
+        std::move(boundOffsetExpression), std::move(nbrOffsetExpression), containsSerial);
     return std::make_unique<BoundCopyFrom>(std::move(boundCopyFromInfo));
 }
 

diff --git a/src/binder/copy/bound_copy_from.cpp b/src/binder/copy/bound_copy_from.cpp
@@ -13,7 +13,7 @@ std::unique_ptr<BoundCopyFromInfo> BoundCopyFromInfo::copy() {
         std::move(copiedColumnExpressions), offsetExpression->copy(),
         tableSchema->tableType == common::TableType::REL ? boundOffsetExpression->copy() : nullptr,
         tableSchema->tableType == common::TableType::REL ? nbrOffsetExpression->copy() : nullptr,
-        preservingOrder);
+        containsSerial);
 }
 
 } // namespace binder

diff --git a/src/common/vector/value_vector.cpp b/src/common/vector/value_vector.cpp
@@ -299,6 +299,12 @@ void ArrowColumnVector::setArrowColumn(
     arrowColumnBuffer->column = std::move(column);
 }
 
+void ArrowColumnVector::slice(ValueVector* vector, offset_t offset) {
+    auto arrowColumnBuffer =
+        reinterpret_cast<ArrowColumnAuxiliaryBuffer*>(vector->auxiliaryBuffer.get());
+    setArrowColumn(vector, arrowColumnBuffer->column->Slice((int64_t)offset));
+}
+
 template void ValueVector::setValue<nodeID_t>(uint32_t pos, nodeID_t val);
 template void ValueVector::setValue<bool>(uint32_t pos, bool val);
 template void ValueVector::setValue<int64_t>(uint32_t pos, int64_t val);

diff --git a/src/include/binder/binder.h b/src/include/binder/binder.h
@@ -98,7 +98,7 @@ class Binder {
     std::unique_ptr<common::LogicalType> bindDataType(const std::string& dataType);
 
     /*** bind copy from/to ***/
-    static bool bindPreservingOrder(
+    static bool bindContainsSerial(
         catalog::TableSchema* tableSchema, common::CopyDescription::FileType fileType);
     expression_vector bindColumnExpressions(catalog::TableSchema* tableSchema);
     std::unique_ptr<BoundStatement> bindCopyFromClause(const parser::Statement& statement);

diff --git a/src/include/binder/copy/bound_copy_from.h b/src/include/binder/copy/bound_copy_from.h
@@ -20,18 +20,18 @@ struct BoundCopyFromInfo {
     std::shared_ptr<Expression> boundOffsetExpression;
     std::shared_ptr<Expression> nbrOffsetExpression;
 
-    bool preservingOrder;
+    bool containsSerial;
 
     BoundCopyFromInfo(std::unique_ptr<common::CopyDescription> copyDesc,
         catalog::TableSchema* tableSchema, expression_vector columnExpressions,
         std::shared_ptr<Expression> offsetExpression,
         std::shared_ptr<Expression> boundOffsetExpression,
-        std::shared_ptr<Expression> nbrOffsetExpression, bool preservingOrder)
+        std::shared_ptr<Expression> nbrOffsetExpression, bool containsSerial)
         : copyDesc{std::move(copyDesc)}, tableSchema{tableSchema}, columnExpressions{std::move(
                                                                        columnExpressions)},
           offsetExpression{std::move(offsetExpression)}, boundOffsetExpression{std::move(
                                                              boundOffsetExpression)},
-          nbrOffsetExpression{std::move(nbrOffsetExpression)}, preservingOrder{preservingOrder} {}
+          nbrOffsetExpression{std::move(nbrOffsetExpression)}, containsSerial{containsSerial} {}
 
     std::unique_ptr<BoundCopyFromInfo> copy();
 };

diff --git a/src/include/common/vector/value_vector.h b/src/include/common/vector/value_vector.h
@@ -230,6 +230,8 @@ class ArrowColumnVector {
     }
 
     static void setArrowColumn(ValueVector* vector, std::shared_ptr<arrow::ChunkedArray> column);
+
+    static void slice(ValueVector* vector, offset_t offset);
 };
 
 class MapVector {

diff --git a/src/include/planner/logical_plan/copy/logical_copy_from.h b/src/include/planner/logical_plan/copy/logical_copy_from.h
@@ -14,13 +14,7 @@ class LogicalCopyFrom : public LogicalOperator {
     LogicalCopyFrom(std::unique_ptr<binder::BoundCopyFromInfo> info,
         std::shared_ptr<binder::Expression> outputExpression)
         : LogicalOperator{LogicalOperatorType::COPY_FROM}, info{std::move(info)},
-          outputExpression{std::move(outputExpression)} {
-        assert((this->info->tableSchema->tableType == common::TableType::REL &&
-                   this->info->boundOffsetExpression && this->info->nbrOffsetExpression) ||
-               (this->info->tableSchema->tableType == common::TableType::NODE &&
-                   (this->info->boundOffsetExpression == nullptr) &&
-                   (this->info->nbrOffsetExpression == nullptr)));
-    }
+          outputExpression{std::move(outputExpression)} {}
 
     inline std::string getExpressionsForPrinting() const override {
         return info->tableSchema->tableName;

diff --git a/src/include/processor/operator/persistent/copy_node.h b/src/include/processor/operator/persistent/copy_node.h
@@ -58,7 +58,7 @@ struct CopyNodeInfo {
     storage::RelsStore* relsStore;
     catalog::Catalog* catalog;
     storage::WAL* wal;
-    bool orderPreserving;
+    bool containsSerial;
 };
 
 class CopyNode : public Sink {
@@ -80,7 +80,10 @@ class CopyNode : public Sink {
 
     void executeInternal(ExecutionContext* context) final;
 
-    void finalize(ExecutionContext* context) final;
+    static void sliceDataChunk(const common::DataChunk& dataChunk,
+        const std::vector<DataPos>& dataColumnPoses, common::offset_t offset);
+
+    void finalize(ExecutionContext* context) override;
 
     inline std::unique_ptr<PhysicalOperator> clone() final {
         return std::make_unique<CopyNode>(sharedState, copyNodeInfo, resultSetDescriptor->copy(),

diff --git a/src/include/processor/operator/persistent/reader.h b/src/include/processor/operator/persistent/reader.h
@@ -9,15 +9,15 @@ namespace processor {
 struct ReaderInfo {
     DataPos nodeOffsetPos;
     std::vector<DataPos> dataColumnsPos;
-    bool orderPreserving;
+    bool containsSerial;
 
     ReaderInfo(
-        const DataPos& nodeOffsetPos, std::vector<DataPos> dataColumnsPos, bool orderPreserving)
+        const DataPos& nodeOffsetPos, std::vector<DataPos> dataColumnsPos, bool containsSerial)
         : nodeOffsetPos{nodeOffsetPos}, dataColumnsPos{std::move(dataColumnsPos)},
-          orderPreserving{orderPreserving} {}
+          containsSerial{containsSerial} {}
     ReaderInfo(const ReaderInfo& other)
         : nodeOffsetPos{other.nodeOffsetPos}, dataColumnsPos{other.dataColumnsPos},
-          orderPreserving{other.orderPreserving} {}
+          containsSerial{other.containsSerial} {}
 
     inline uint32_t getNumColumns() const { return dataColumnsPos.size(); }
 
@@ -45,12 +45,15 @@ class Reader : public PhysicalOperator {
         return make_unique<Reader>(info->copy(), sharedState, getOperatorID(), paramsString);
     }
 
+    inline bool getContainsSerial() const { return info->containsSerial; }
+
 protected:
     bool getNextTuplesInternal(ExecutionContext* context) final;
 
 private:
     void getNextNodeGroupInSerial();
     void getNextNodeGroupInParallel();
+    void readNextNodeGroupInParallel();
 
 private:
     std::unique_ptr<ReaderInfo> info;

diff --git a/src/include/processor/operator/persistent/reader_state.h b/src/include/processor/operator/persistent/reader_state.h
@@ -160,6 +160,7 @@ class ReaderSharedState {
 
 private:
     std::unique_ptr<ReaderMorsel> getMorselOfNextBlock();
+    void readNextBlock(common::DataChunk* dataChunk);
 
 public:
     std::mutex mtx;

diff --git a/src/include/storage/storage_info.h b/src/include/storage/storage_info.h
@@ -12,10 +12,11 @@ using storage_version_t = uint64_t;
 
 struct StorageVersionInfo {
     static std::unordered_map<std::string, storage_version_t> getStorageVersionInfo() {
-        return {{"0.0.8.3", 19}, {"0.0.8.2", 19}, {"0.0.8.1", 18}, {"0.0.8", 17}, {"0.0.7.1", 16},
-            {"0.0.7", 15}, {"0.0.6.5", 14}, {"0.0.6.4", 13}, {"0.0.6.3", 12}, {"0.0.6.2", 11},
-            {"0.0.6.1", 10}, {"0.0.6", 9}, {"0.0.5", 8}, {"0.0.4", 7}, {"0.0.3.5", 6},
-            {"0.0.3.4", 5}, {"0.0.3.3", 4}, {"0.0.3.2", 3}, {"0.0.3.1", 2}, {"0.0.3", 1}};
+        return {{"0.0.8.4", 19}, {"0.0.8.3", 19}, {"0.0.8.2", 19}, {"0.0.8.1", 18}, {"0.0.8", 17},
+            {"0.0.7.1", 16}, {"0.0.7", 15}, {"0.0.6.5", 14}, {"0.0.6.4", 13}, {"0.0.6.3", 12},
+            {"0.0.6.2", 11}, {"0.0.6.1", 10}, {"0.0.6", 9}, {"0.0.5", 8}, {"0.0.4", 7},
+            {"0.0.3.5", 6}, {"0.0.3.4", 5}, {"0.0.3.3", 4}, {"0.0.3.2", 3}, {"0.0.3.1", 2},
+            {"0.0.3", 1}};
     }
 
     static storage_version_t getStorageVersion();

diff --git a/src/processor/map/map_copy_from.cpp b/src/processor/map/map_copy_from.cpp
@@ -54,7 +54,7 @@ std::unique_ptr<PhysicalOperator> PlanMapper::mapCopyNodeFrom(
     // Map reader.
     auto reader = createReader(copyFromInfo->copyDesc.get(), copyFromInfo->tableSchema,
         copyFrom->getSchema(), copyFromInfo->columnExpressions, copyFromInfo->offsetExpression,
-        copyFromInfo->preservingOrder);
+        copyFromInfo->containsSerial);
     auto readerOp = reinterpret_cast<Reader*>(reader.get());
     auto readerInfo = readerOp->getReaderInfo();
     auto nodeTable = storageManager.getNodesStore().getNodeTable(tableSchema->tableID);
@@ -64,7 +64,7 @@ std::unique_ptr<PhysicalOperator> PlanMapper::mapCopyNodeFrom(
             tableSchema, nodeTable, *copyFromInfo->copyDesc, memoryManager);
     CopyNodeInfo copyNodeDataInfo{readerInfo->dataColumnsPos, readerInfo->nodeOffsetPos,
         *copyFromInfo->copyDesc, nodeTable, &storageManager.getRelsStore(), catalog,
-        storageManager.getWAL(), copyFromInfo->preservingOrder};
+        storageManager.getWAL(), copyFromInfo->containsSerial};
     auto copyNode = std::make_unique<CopyNode>(copyNodeSharedState, copyNodeDataInfo,
         std::make_unique<ResultSetDescriptor>(copyFrom->getSchema()), std::move(reader),
         getOperatorID(), copyFrom->getExpressionsForPrinting());
@@ -102,7 +102,7 @@ std::unique_ptr<PhysicalOperator> PlanMapper::createCopyRelColumnsOrLists(
     auto tableSchema = reinterpret_cast<RelTableSchema*>(copyFromInfo->tableSchema);
     auto reader = createReader(copyFromInfo->copyDesc.get(), copyFromInfo->tableSchema, outFSchema,
         copyFromInfo->columnExpressions, copyFromInfo->offsetExpression,
-        copyFromInfo->preservingOrder);
+        copyFromInfo->containsSerial);
     auto readerOp = reinterpret_cast<Reader*>(reader.get());
     auto readerInfo = readerOp->getReaderInfo();
     auto offsetDataPos = DataPos{outFSchema->getExpressionPos(*copyFromInfo->offsetExpression)};

diff --git a/src/processor/operator/persistent/copy_node.cpp b/src/processor/operator/persistent/copy_node.cpp
@@ -86,29 +86,37 @@ void CopyNode::executeInternal(ExecutionContext* context) {
         auto numTuplesToAppend = ArrowColumnVector::getArrowColumn(
             resultSet->getValueVector(copyNodeInfo.dataColumnPoses[0]).get())
                                      ->length();
-        assert(numTuplesToAppend <= StorageConstants::NODE_GROUP_SIZE);
-        auto nodeOffset = nodeOffsetVector->getValue<offset_t>(
-            nodeOffsetVector->state->selVector->selectedPositions[0]);
-        auto numAppendedTuplesInNodeGroup =
-            localNodeGroup->append(resultSet, copyNodeInfo.dataColumnPoses, numTuplesToAppend);
-        assert(numAppendedTuplesInNodeGroup == numTuplesToAppend);
-        if (localNodeGroup->isFull()) {
-            node_group_idx_t nodeGroupIdx;
-            if (copyNodeInfo.orderPreserving) {
-                nodeGroupIdx = StorageUtils::getNodeGroupIdx(nodeOffset);
-                sharedState->setNextNodeGroupIdx(nodeGroupIdx + 1);
-            } else {
+        auto numAppendedTuples = 0ul;
+        while (numAppendedTuples < numTuplesToAppend) {
+            auto numAppendedTuplesInNodeGroup = localNodeGroup->append(
+                resultSet, copyNodeInfo.dataColumnPoses, numTuplesToAppend - numAppendedTuples);
+            numAppendedTuples += numAppendedTuplesInNodeGroup;
+            if (localNodeGroup->isFull()) {
+                node_group_idx_t nodeGroupIdx;
                 nodeGroupIdx = sharedState->getNextNodeGroupIdx();
+                writeAndResetNodeGroup(nodeGroupIdx, sharedState->pkIndex.get(),
+                    sharedState->pkColumnID, sharedState->table, localNodeGroup.get());
+            }
+            if (numAppendedTuples < numTuplesToAppend) {
+                sliceDataChunk(
+                    *resultSet->getDataChunk(copyNodeInfo.dataColumnPoses[0].dataChunkPos),
+                    copyNodeInfo.dataColumnPoses, (offset_t)numAppendedTuplesInNodeGroup);
             }
-            writeAndResetNodeGroup(nodeGroupIdx, sharedState->pkIndex.get(),
-                sharedState->pkColumnID, sharedState->table, localNodeGroup.get());
         }
     }
     if (localNodeGroup->getNumNodes() > 0) {
         sharedState->appendLocalNodeGroup(std::move(localNodeGroup));
     }
 }
 
+void CopyNode::sliceDataChunk(
+    const DataChunk& dataChunk, const std::vector<DataPos>& dataColumnPoses, offset_t offset) {
+    for (auto& dataColumnPos : dataColumnPoses) {
+        ArrowColumnVector::slice(
+            dataChunk.valueVectors[dataColumnPos.valueVectorPos].get(), offset);
+    }
+}
+
 void CopyNode::writeAndResetNodeGroup(node_group_idx_t nodeGroupIdx,
     PrimaryKeyIndexBuilder* pkIndex, column_id_t pkColumnID, NodeTable* table,
     NodeGroup* nodeGroup) {

diff --git a/src/processor/operator/persistent/reader.cpp b/src/processor/operator/persistent/reader.cpp
@@ -26,7 +26,9 @@ void Reader::initLocalStateInternal(ResultSet* resultSet, ExecutionContext* cont
 }
 
 bool Reader::getNextTuplesInternal(ExecutionContext* context) {
-    info->orderPreserving ? getNextNodeGroupInSerial() : getNextNodeGroupInParallel();
+    sharedState->copyDescription->fileType == common::CopyDescription::FileType::CSV ?
+        getNextNodeGroupInSerial() :
+        getNextNodeGroupInParallel();
     return dataChunk->state->selVector->selectedSize != 0;
 }
 
@@ -40,10 +42,21 @@ void Reader::getNextNodeGroupInSerial() {
 }
 
 void Reader::getNextNodeGroupInParallel() {
-    while (leftArrowArrays.getLeftNumRows() < StorageConstants::NODE_GROUP_SIZE) {
+    readNextNodeGroupInParallel();
+    if (leftArrowArrays.getLeftNumRows() == 0) {
+        dataChunk->state->selVector->selectedSize = 0;
+    } else {
+        int64_t numRowsToReturn =
+            std::min(leftArrowArrays.getLeftNumRows(), DEFAULT_VECTOR_CAPACITY);
+        leftArrowArrays.appendToDataChunk(dataChunk.get(), numRowsToReturn);
+    }
+}
+
+void Reader::readNextNodeGroupInParallel() {
+    if (leftArrowArrays.getLeftNumRows() == 0) {
         auto morsel = sharedState->getParallelMorsel();
         if (morsel->fileIdx == INVALID_VECTOR_IDX) {
-            break;
+            return;
         }
         if (!readFuncData || morsel->fileIdx != readFuncData->fileIdx) {
             readFuncData = initFunc(sharedState->copyDescription->filePaths, morsel->fileIdx,
@@ -52,13 +65,6 @@ void Reader::getNextNodeGroupInParallel() {
         readFunc(*readFuncData, morsel->blockIdx, dataChunk.get());
         leftArrowArrays.appendFromDataChunk(dataChunk.get());
     }
-    if (leftArrowArrays.getLeftNumRows() == 0) {
-        dataChunk->state->selVector->selectedSize = 0;
-    } else {
-        int64_t numRowsToReturn =
-            std::min(leftArrowArrays.getLeftNumRows(), StorageConstants::NODE_GROUP_SIZE);
-        leftArrowArrays.appendToDataChunk(dataChunk.get(), numRowsToReturn);
-    }
 }
 
 } // namespace processor