column.h

Go to the documentation of this file.

#pragma once
 
#include "catalog.h"
#include "null_mask.h"
#include "types.h"
#include "compression.h"
#include "db_file_id.h"
#include "column_chunk_data.h"
 
namespace kuzu {
namespace evaluator {
class ExpressionEvaluator;
} // namespace evaluator
namespace storage {
 
struct CompressionMetadata;
 
using read_values_to_vector_func_t =
    std::function<void(uint8_t* frame, PageCursor& pageCursor, common::ValueVector* resultVector,
        uint32_t posInVector, uint32_t numValuesToRead, const CompressionMetadata& metadata)>;
using write_values_from_vector_func_t = std::function<void(uint8_t* frame, uint16_t posInFrame,
    common::ValueVector* vector, uint32_t posInVector, const CompressionMetadata& metadata)>;
using write_values_func_t = std::function<void(uint8_t* frame, uint16_t posInFrame,
    const uint8_t* data, common::offset_t dataOffset, common::offset_t numValues,
    const CompressionMetadata& metadata, const common::NullMask*)>;
 
using read_values_to_page_func_t =
    std::function<void(uint8_t* frame, PageCursor& pageCursor, uint8_t* result,
        uint32_t posInResult, uint64_t numValues, const CompressionMetadata& metadata)>;
// This is a special usage for the `batchLookup` interface.
using batch_lookup_func_t = read_values_to_page_func_t;
 
class NullColumn;
class StructColumn;
class RelTableData;
struct ColumnCheckpointState;
class ShadowFile;
class BufferManager;
class Column {
    friend class StringColumn;
    friend class StructColumn;
    friend class ListColumn;
    friend class RelTableData;
 
public:
    // TODO(Guodong): Remove transaction from interface of Column. There is no need to be aware of
    // transaction when reading/writing from/to disk pages.
    Column(std::string name, common::LogicalType dataType, BMFileHandle* dataFH,
        BufferManager* bufferManager, ShadowFile* shadowFile, bool enableCompression,
        bool requireNullColumn = true);
    virtual ~Column();
 
    static std::unique_ptr<ColumnChunkData> flushChunkData(const ColumnChunkData& chunkData,
        BMFileHandle& dataFH);
    static std::unique_ptr<ColumnChunkData> flushNonNestedChunkData(
        const ColumnChunkData& chunkData, BMFileHandle& dataFH);
    static ColumnChunkMetadata flushData(const ColumnChunkData& chunkData, BMFileHandle& dataFH);
 
    virtual void scan(transaction::Transaction* transaction, const ChunkState& state,
        common::offset_t startOffsetInChunk, common::row_idx_t numValuesToScan,
        common::ValueVector* nodeIDVector, common::ValueVector* resultVector);
    virtual void lookupValue(transaction::Transaction* transaction, const ChunkState& state,
        common::offset_t nodeOffset, common::ValueVector* resultVector, uint32_t posInVector);
 
    // Scan from [startOffsetInGroup, endOffsetInGroup).
    virtual void scan(transaction::Transaction* transaction, const ChunkState& state,
        common::offset_t startOffsetInGroup, common::offset_t endOffsetInGroup,
        common::ValueVector* resultVector, uint64_t offsetInVector);
    // Scan from [startOffsetInGroup, endOffsetInGroup).
    virtual void scan(transaction::Transaction* transaction, const ChunkState& state,
        ColumnChunkData* columnChunk, common::offset_t startOffset = 0,
        common::offset_t endOffset = common::INVALID_OFFSET);
 
    common::LogicalType& getDataType() { return dataType; }
    const common::LogicalType& getDataType() const { return dataType; }
 
    Column* getNullColumn() const;
 
    std::string getName() const { return name; }
 
    virtual void scan(transaction::Transaction* transaction, const ChunkState& state,
        common::offset_t startOffsetInGroup, common::offset_t endOffsetInGroup, uint8_t* result);
 
    // Batch write to a set of sequential pages.
    virtual void write(ColumnChunkData& persistentChunk, ChunkState& state,
        common::offset_t dstOffset, ColumnChunkData* data, common::offset_t srcOffset,
        common::length_t numValues);
 
    // Append values to the end of the node group, resizing it if necessary
    common::offset_t appendValues(ColumnChunkData& persistentChunk, ChunkState& state,
        const uint8_t* data, const common::NullMask* nullChunkData, common::offset_t numValues);
 
    virtual void checkpointColumnChunk(ColumnCheckpointState& checkpointState);
 
    template<class TARGET>
    TARGET& cast() {
        return common::ku_dynamic_cast<Column&, TARGET&>(*this);
    }
    template<class TARGET>
    const TARGET& cast() const {
        return common::ku_dynamic_cast<Column&, TARGET&>(*this);
    }
 
protected:
    virtual void scanInternal(transaction::Transaction* transaction, const ChunkState& state,
        common::offset_t startOffsetInChunk, common::row_idx_t numValuesToScan,
        common::ValueVector* nodeIDVector, common::ValueVector* resultVector);
    void scanUnfiltered(transaction::Transaction* transaction, PageCursor& pageCursor,
        uint64_t numValuesToScan, common::ValueVector* resultVector,
        const ColumnChunkMetadata& chunkMeta, uint64_t startPosInVector = 0) const;
    void scanFiltered(transaction::Transaction* transaction, PageCursor& pageCursor,
        uint64_t numValuesToScan, const common::SelectionVector& selVector,
        common::ValueVector* resultVector, const ColumnChunkMetadata& chunkMeta) const;
 
    virtual void lookupInternal(transaction::Transaction* transaction, const ChunkState& state,
        common::offset_t nodeOffset, common::ValueVector* resultVector, uint32_t posInVector);
 
    void readFromPage(transaction::Transaction* transaction, common::page_idx_t pageIdx,
        const std::function<void(uint8_t*)>& func) const;
 
    virtual void writeValues(ColumnChunkData& persistentChunk, ChunkState& state,
        common::offset_t dstOffset, const uint8_t* data, const common::NullMask* nullChunkData,
        common::offset_t srcOffset = 0, common::offset_t numValues = 1);
 
    // Produces a page cursor for the offset relative to the given node group
    static PageCursor getPageCursorForOffsetInGroup(common::offset_t offsetInChunk,
        const ChunkState& state);
    void updatePageWithCursor(PageCursor cursor,
        const std::function<void(uint8_t*, common::offset_t)>& writeOp) const;
 
    void updateStatistics(ColumnChunkMetadata& metadata, common::offset_t maxIndex,
        const std::optional<StorageValue>& min, const std::optional<StorageValue>& max) const;
 
protected:
    bool isMaxOffsetOutOfPagesCapacity(const ColumnChunkMetadata& metadata,
        common::offset_t maxOffset) const;
 
    virtual bool canCheckpointInPlace(const ChunkState& state,
        const ColumnCheckpointState& checkpointState);
 
    virtual void checkpointColumnChunkInPlace(ChunkState& state,
        const ColumnCheckpointState& checkpointState);
    void checkpointNullData(const ColumnCheckpointState& checkpointState) const;
 
    virtual void checkpointColumnChunkOutOfPlace(ChunkState& state,
        const ColumnCheckpointState& checkpointState);
 
    // check if val is in range [start, end)
    static bool isInRange(uint64_t val, uint64_t start, uint64_t end) {
        return val >= start && val < end;
    }
 
protected:
    std::string name;
    DBFileID dbFileID;
    common::LogicalType dataType;
    BMFileHandle* dataFH;
    BufferManager* bufferManager;
    ShadowFile* shadowFile;
    std::unique_ptr<NullColumn> nullColumn;
    read_values_to_vector_func_t readToVectorFunc;
    write_values_from_vector_func_t writeFromVectorFunc;
    write_values_func_t writeFunc;
    read_values_to_page_func_t readToPageFunc;
    batch_lookup_func_t batchLookupFunc;
    bool enableCompression;
};
 
class InternalIDColumn final : public Column {
public:
    InternalIDColumn(std::string name, BMFileHandle* dataFH, BufferManager* bufferManager,
        ShadowFile* shadowFile, bool enableCompression);
 
    void scan(transaction::Transaction* transaction, const ChunkState& state,
        common::offset_t startOffsetInChunk, common::row_idx_t numValuesToScan,
        common::ValueVector* nodeIDVector, common::ValueVector* resultVector) override {
        Column::scan(transaction, state, startOffsetInChunk, numValuesToScan, nodeIDVector,
            resultVector);
        populateCommonTableID(resultVector);
    }
 
    void scan(transaction::Transaction* transaction, const ChunkState& state,
        common::offset_t startOffsetInGroup, common::offset_t endOffsetInGroup,
        common::ValueVector* resultVector, uint64_t offsetInVector) override {
        Column::scan(transaction, state, startOffsetInGroup, endOffsetInGroup, resultVector,
            offsetInVector);
        populateCommonTableID(resultVector);
    }
 
    void lookupInternal(transaction::Transaction* transaction, const ChunkState& state,
        common::offset_t nodeOffset, common::ValueVector* resultVector,
        uint32_t posInVector) override {
        Column::lookupInternal(transaction, state, nodeOffset, resultVector, posInVector);
        populateCommonTableID(resultVector);
    }
 
    common::table_id_t getCommonTableID() const { return commonTableID; }
    // TODO(Guodong): This function should be removed through rewritting INTERNAL_ID as STRUCT.
    void setCommonTableID(common::table_id_t tableID) { commonTableID = tableID; }
 
private:
    void populateCommonTableID(const common::ValueVector* resultVector) const;
 
private:
    common::table_id_t commonTableID;
};
 
struct ColumnFactory {
    static std::unique_ptr<Column> createColumn(std::string name, common::LogicalType dataType,
        BMFileHandle* dataFH, BufferManager* bufferManager, ShadowFile* shadowFile,
        bool enableCompression);
};
 
} // namespace storage
} // namespace kuzu