node_group.h

Go to the documentation of this file.

#pragma once
 
#include "uniq_lock.h"
#include "residency_state.h"
#include "chunked_node_group.h"
#include "group_collection.h"
 
namespace kuzu {
namespace transaction {
class Transaction;
} // namespace transaction
 
namespace storage {
 
struct TableAddColumnState;
class NodeGroup;
struct NodeGroupScanState {
 // Index of committed but not yet checkpointed chunked group to scan.
 common::idx_t chunkedGroupIdx = 0;
 common::row_idx_t nextRowToScan = 0;
 // State of each chunk in the checkpointed chunked group.
 std::vector<ChunkState> chunkStates;
 
 explicit NodeGroupScanState(common::idx_t numChunks) { chunkStates.resize(numChunks); }
 virtual ~NodeGroupScanState() = default;
 DELETE_COPY_DEFAULT_MOVE(NodeGroupScanState);
 
 virtual void resetState() {
 chunkedGroupIdx = 0;
 nextRowToScan = 0;
 for (auto& chunkState : chunkStates) {
 chunkState.resetState();
 }
 }
 
 template<class TARGET>
 TARGET& cast() {
 return common::ku_dynamic_cast<NodeGroupScanState&, TARGET&>(*this);
 }
 template<class TARGETT>
 const TARGETT& constCast() {
 return common::ku_dynamic_cast<const NodeGroupScanState&, const TARGETT&>(*this);
 }
};
 
class MemoryManager;
struct NodeGroupCheckpointState {
 std::vector<common::column_id_t> columnIDs;
 std::vector<std::unique_ptr<Column>> columns;
 BMFileHandle& dataFH;
 MemoryManager* mm;
 
 NodeGroupCheckpointState(std::vector<common::column_id_t> columnIDs,
 std::vector<std::unique_ptr<Column>> columns, BMFileHandle& dataFH, MemoryManager* mm)
 : columnIDs{std::move(columnIDs)}, columns{std::move(columns)}, dataFH{dataFH}, mm{mm} {}
 virtual ~NodeGroupCheckpointState() = default;
 
 template<typename T>
 const T& cast() const {
 return common::ku_dynamic_cast<const NodeGroupCheckpointState&, const T&>(*this);
 }
 template<typename T>
 T& cast() {
 return common::ku_dynamic_cast<NodeGroupCheckpointState&, T&>(*this);
 }
};
 
struct NodeGroupScanResult {
 
 common::row_idx_t startRow = common::INVALID_ROW_IDX;
 common::row_idx_t numRows = 0;
 
 constexpr NodeGroupScanResult() noexcept = default;
 constexpr NodeGroupScanResult(common::row_idx_t startRow, common::row_idx_t numRows) noexcept
 : startRow{startRow}, numRows{numRows} {}
 
 bool operator==(const NodeGroupScanResult& other) const {
 return startRow == other.startRow && numRows == other.numRows;
 }
};
 
static auto NODE_GROUP_SCAN_EMMPTY_RESULT = NodeGroupScanResult{};
 
struct TableScanState;
class NodeGroup {
public:
 NodeGroup(const common::node_group_idx_t nodeGroupIdx, const bool enableCompression,
 std::vector<common::LogicalType> dataTypes,
 common::row_idx_t capacity = common::StorageConstants::NODE_GROUP_SIZE,
 NodeGroupDataFormat format = NodeGroupDataFormat::REGULAR)
 : nodeGroupIdx{nodeGroupIdx}, format{format}, enableCompression{enableCompression},
 numRows{0}, nextRowToAppend{0}, capacity{capacity}, dataTypes{std::move(dataTypes)} {}
 NodeGroup(const common::node_group_idx_t nodeGroupIdx, const bool enableCompression,
 std::unique_ptr<ChunkedNodeGroup> chunkedNodeGroup,
 common::row_idx_t capacity = common::StorageConstants::NODE_GROUP_SIZE,
 NodeGroupDataFormat format = NodeGroupDataFormat::REGULAR)
 : nodeGroupIdx{nodeGroupIdx}, format{format}, enableCompression{enableCompression},
 numRows{chunkedNodeGroup->getNumRows()}, nextRowToAppend{numRows}, capacity{capacity} {
 for (auto i = 0u; i < chunkedNodeGroup->getNumColumns(); i++) {
 dataTypes.push_back(chunkedNodeGroup->getColumnChunk(i).getDataType().copy());
 }
 const auto lock = chunkedGroups.lock();
 chunkedGroups.appendGroup(lock, std::move(chunkedNodeGroup));
 }
 NodeGroup(const common::node_group_idx_t nodeGroupIdx, const bool enableCompression,
 common::row_idx_t capacity, NodeGroupDataFormat format)
 : nodeGroupIdx{nodeGroupIdx}, format{format}, enableCompression{enableCompression},
 numRows{0}, nextRowToAppend{0}, capacity{capacity} {}
 virtual ~NodeGroup() = default;
 
 virtual bool isEmpty() const { return numRows.load() == 0; }
 virtual common::row_idx_t getNumRows() const { return numRows.load(); }
 void moveNextRowToAppend(common::row_idx_t numRowsToAppend) {
 nextRowToAppend += numRowsToAppend;
 }
 common::row_idx_t getNumRowsLeftToAppend() const { return capacity - nextRowToAppend; }
 bool isFull() const { return numRows.load() == capacity; }
 const std::vector<common::LogicalType>& getDataTypes() const { return dataTypes; }
 NodeGroupDataFormat getFormat() const { return format; }
 common::row_idx_t append(const transaction::Transaction* transaction,
 ChunkedNodeGroup& chunkedGroup, common::row_idx_t startRowIdx,
 common::row_idx_t numRowsToAppend);
 common::row_idx_t append(const transaction::Transaction* transaction,
 const std::vector<ColumnChunk*>& chunkedGroup, common::row_idx_t startRowIdx,
 common::row_idx_t numRowsToAppend);
 void append(const transaction::Transaction* transaction,
 const std::vector<common::ValueVector*>& vectors, common::row_idx_t startRowIdx,
 common::row_idx_t numRowsToAppend);
 
 void merge(transaction::Transaction* transaction,
 std::unique_ptr<ChunkedNodeGroup> chunkedGroup);
 
 virtual void initializeScanState(transaction::Transaction* transaction, TableScanState& state);
 void initializeScanState(transaction::Transaction* transaction, const common::UniqLock& lock,
 TableScanState& state);
 virtual NodeGroupScanResult scan(transaction::Transaction* transaction, TableScanState& state);
 
 bool lookup(const common::UniqLock& lock, transaction::Transaction* transaction,
 const TableScanState& state);
 bool lookup(transaction::Transaction* transaction, const TableScanState& state);
 
 void update(transaction::Transaction* transaction, common::row_idx_t rowIdxInGroup,
 common::column_id_t columnID, const common::ValueVector& propertyVector);
 bool delete_(const transaction::Transaction* transaction, common::row_idx_t rowIdxInGroup);
 
 common::row_idx_t getNumDeletedRows(const transaction::Transaction* transaction);
 virtual void addColumn(transaction::Transaction* transaction,
 TableAddColumnState& addColumnState, BMFileHandle* dataFH);
 
 void flush(BMFileHandle& dataFH);
 
 virtual void checkpoint(NodeGroupCheckpointState& state);
 
 bool hasChanges();
 uint64_t getEstimatedMemoryUsage();
 
 virtual void serialize(common::Serializer& serializer);
 static std::unique_ptr<NodeGroup> deserialize(common::Deserializer& deSer);
 
 common::node_group_idx_t getNumChunkedGroups() {
 const auto lock = chunkedGroups.lock();
 return chunkedGroups.getNumGroups(lock);
 }
 ChunkedNodeGroup* getChunkedNodeGroup(common::node_group_idx_t groupIdx) {
 const auto lock = chunkedGroups.lock();
 return chunkedGroups.getGroup(lock, groupIdx);
 }
 
 template<class TARGET>
 TARGET& cast() {
 return common::ku_dynamic_cast<NodeGroup&, TARGET&>(*this);
 }
 template<class TARGETT>
 const TARGETT& cast() const {
 return common::ku_dynamic_cast<const NodeGroup&, const TARGETT&>(*this);
 }
 
 bool isVisible(const transaction::Transaction* transaction, common::row_idx_t rowIdxInGroup);
 bool isDeleted(const transaction::Transaction* transaction, common::offset_t offsetInGroup);
 bool isInserted(const transaction::Transaction* transaction, common::offset_t offsetInGroup);
 
private:
 ChunkedNodeGroup* findChunkedGroupFromRowIdx(const common::UniqLock& lock,
 common::row_idx_t rowIdx);
 ChunkedNodeGroup* findChunkedGroupFromRowIdxNoLock(common::row_idx_t rowIdx);
 
 common::row_idx_t getNumDeletedRows(const common::UniqLock& lock);
 
 std::unique_ptr<ChunkedNodeGroup> checkpointInMemOnly(const common::UniqLock& lock,
 NodeGroupCheckpointState& state);
 std::unique_ptr<ChunkedNodeGroup> checkpointInMemAndOnDisk(const common::UniqLock& lock,
 NodeGroupCheckpointState& state);
 std::unique_ptr<VersionInfo> checkpointVersionInfo(const common::UniqLock& lock,
 const transaction::Transaction* transaction);
 
 template<ResidencyState SCAN_RESIDENCY_STATE>
 common::row_idx_t getNumResidentRows(const common::UniqLock& lock);
 template<ResidencyState SCAN_RESIDENCY_STATE>
 std::unique_ptr<ChunkedNodeGroup> scanAllInsertedAndVersions(const common::UniqLock& lock,
 const std::vector<common::column_id_t>& columnIDs, const std::vector<Column*>& columns);
 
 static void populateNodeID(common::ValueVector& nodeIDVector, common::table_id_t tableID,
 common::offset_t startNodeOffset, common::row_idx_t numRows);
 
protected:
 common::node_group_idx_t nodeGroupIdx;
 NodeGroupDataFormat format;
 bool enableCompression;
 std::atomic<common::row_idx_t> numRows;
 // `nextRowToAppend` is a cursor to allow us to pre-reserve a set of rows to append before
 // acutally appending data. This is an optimization to reduce lock-contention when appending in
 // parallel.
 // TODO(Guodong): Remove this field.
 common::row_idx_t nextRowToAppend;
 common::row_idx_t capacity;
 std::vector<common::LogicalType> dataTypes;
 GroupCollection<ChunkedNodeGroup> chunkedGroups;
};
 
} // namespace storage
} // namespace kuzu