chunked_node_group.h

Go to the documentation of this file.

#pragma once
 
#include <atomic>
 
#include "rel_multiplicity.h"
#include "internal_id_t.h"
#include "residency_state.h"
#include "column_chunk.h"
#include "column_chunk_data.h"
#include "version_info.h"
 
namespace kuzu {
namespace common {
class SelectionVector;
} // namespace common
 
namespace transaction {
class Transaction;
} // namespace transaction
 
namespace storage {
 
class Column;
struct TableScanState;
struct TableAddColumnState;
struct NodeGroupScanState;
 
enum class NodeGroupDataFormat : uint8_t { REGULAR = 0, CSR = 1 };
 
class ChunkedNodeGroup {
public:
    static constexpr uint64_t CHUNK_CAPACITY = 2048;
 
    ChunkedNodeGroup(std::vector<std::unique_ptr<ColumnChunk>> chunks,
        common::row_idx_t startRowIdx, NodeGroupDataFormat format = NodeGroupDataFormat::REGULAR);
    ChunkedNodeGroup(ChunkedNodeGroup& base,
        const std::vector<common::column_id_t>& selectedColumns);
    ChunkedNodeGroup(const std::vector<common::LogicalType>& columnTypes, bool enableCompression,
        uint64_t capacity, common::row_idx_t startRowIdx, ResidencyState residencyState,
        NodeGroupDataFormat format = NodeGroupDataFormat::REGULAR);
    virtual ~ChunkedNodeGroup() = default;
 
    common::idx_t getNumColumns() const { return chunks.size(); }
    common::row_idx_t getStartRowIdx() const { return startRowIdx; }
    common::row_idx_t getNumRows() const { return numRows; }
    common::row_idx_t getCapacity() const { return capacity; }
    const ColumnChunk& getColumnChunk(const common::column_id_t columnID) const {
        KU_ASSERT(columnID < chunks.size());
        return *chunks[columnID];
    }
    ColumnChunk& getColumnChunk(const common::column_id_t columnID) {
        KU_ASSERT(columnID < chunks.size());
        return *chunks[columnID];
    }
    std::unique_ptr<ColumnChunk> moveColumnChunk(const common::column_id_t columnID) {
        KU_ASSERT(columnID < chunks.size());
        return std::move(chunks[columnID]);
    }
    bool isFullOrOnDisk() const {
        return numRows == capacity || residencyState == ResidencyState::ON_DISK;
    }
    ResidencyState getResidencyState() const { return residencyState; }
    NodeGroupDataFormat getFormat() const { return format; }
 
    void resetToEmpty();
    void resetToAllNull() const;
    void resetNumRowsFromChunks();
    void setNumRows(common::offset_t numRows_);
    void resizeChunks(uint64_t newSize);
    void setVersionInfo(std::unique_ptr<VersionInfo> versionInfo) {
        this->versionInfo = std::move(versionInfo);
    }
    void resetVersionAndUpdateInfo();
 
    uint64_t append(const transaction::Transaction* transaction,
        const std::vector<common::ValueVector*>& columnVectors, common::row_idx_t startRowInVectors,
        uint64_t numValuesToAppend);
    // Appends up to numValuesToAppend from the other chunked node group, returning the actual
    // number of values appended.
    common::offset_t append(const transaction::Transaction* transaction,
        const ChunkedNodeGroup& other, common::offset_t offsetInOtherNodeGroup,
        common::offset_t numRowsToAppend);
    common::offset_t append(const transaction::Transaction* transaction,
        const std::vector<ColumnChunk*>& other, common::offset_t offsetInOtherNodeGroup,
        common::offset_t numRowsToAppend);
    void write(const ChunkedNodeGroup& data, common::column_id_t offsetColumnID);
 
    void scan(const transaction::Transaction* transaction, const TableScanState& scanState,
        const NodeGroupScanState& nodeGroupScanState, common::offset_t rowIdxInGroup,
        common::length_t numRowsToScan) const;
 
    template<ResidencyState SCAN_RESIDENCY_STATE>
    void scanCommitted(transaction::Transaction* transaction, TableScanState& scanState,
        NodeGroupScanState& nodeGroupScanState, ChunkedNodeGroup& output) const;
 
    bool hasUpdates() const;
    common::row_idx_t getNumDeletedRows(const transaction::Transaction* transaction) const;
    common::row_idx_t getNumUpdatedRows(const transaction::Transaction* transaction,
        common::column_id_t columnID);
 
    std::pair<std::unique_ptr<ColumnChunk>, std::unique_ptr<ColumnChunk>> scanUpdates(
        transaction::Transaction* transaction, common::column_id_t columnID);
 
    bool lookup(transaction::Transaction* transaction, const TableScanState& state,
        NodeGroupScanState& nodeGroupScanState, common::offset_t rowIdxInChunk,
        common::sel_t posInOutput) const;
 
    void update(transaction::Transaction* transaction, common::row_idx_t rowIdxInChunk,
        common::column_id_t columnID, const common::ValueVector& propertyVector);
 
    bool delete_(const transaction::Transaction* transaction, common::row_idx_t rowIdxInChunk);
 
    void addColumn(transaction::Transaction* transaction, const TableAddColumnState& addColumnState,
        bool enableCompression, BMFileHandle* dataFH);
 
    bool isDeleted(const transaction::Transaction* transaction, common::row_idx_t rowInChunk) const;
    bool isInserted(const transaction::Transaction* transaction,
        common::row_idx_t rowInChunk) const;
    bool hasAnyUpdates(const transaction::Transaction* transaction, common::column_id_t columnID,
        common::row_idx_t startRow, common::length_t numRows) const;
    common::row_idx_t getNumDeletions(const transaction::Transaction* transaction,
        common::row_idx_t startRow, common::length_t numRows) const;
    bool hasVersionInfo() const { return versionInfo != nullptr; }
 
    void finalize() const;
 
    virtual void writeToColumnChunk(common::idx_t chunkIdx, common::idx_t vectorIdx,
        const std::vector<std::unique_ptr<ColumnChunk>>& data, ColumnChunk& offsetChunk) {
        KU_ASSERT(residencyState != ResidencyState::ON_DISK);
        chunks[chunkIdx]->getData().write(&data[vectorIdx]->getData(), &offsetChunk.getData(),
            common::RelMultiplicity::ONE);
    }
 
    virtual std::unique_ptr<ChunkedNodeGroup> flushAsNewChunkedNodeGroup(
        transaction::Transaction* transaction, BMFileHandle& dataFH) const;
    virtual void flush(BMFileHandle& dataFH);
 
    uint64_t getEstimatedMemoryUsage() const;
 
    virtual void serialize(common::Serializer& serializer) const;
    static std::unique_ptr<ChunkedNodeGroup> deserialize(common::Deserializer& deSer);
 
    template<class TARGET>
    TARGET& cast() {
        return common::ku_dynamic_cast<ChunkedNodeGroup&, TARGET&>(*this);
    }
    template<class TARGETT>
    const TARGETT& cast() const {
        return common::ku_dynamic_cast<const ChunkedNodeGroup&, const TARGETT&>(*this);
    }
 
protected:
    NodeGroupDataFormat format;
    ResidencyState residencyState;
    common::row_idx_t startRowIdx;
    uint64_t capacity;
    std::atomic<common::row_idx_t> numRows;
    std::vector<std::unique_ptr<ColumnChunk>> chunks;
    std::unique_ptr<VersionInfo> versionInfo;
};
 
} // namespace storage
} // namespace kuzu