Packages

An Elixir DuckDB library

Current section

Files

Jump to
exduckdb c_src duckdb src storage table column_checkpoint_state.cpp
Raw

c_src/duckdb/src/storage/table/column_checkpoint_state.cpp

#include "duckdb/storage/table/column_data.hpp"
#include "duckdb/storage/table/column_segment.hpp"
#include "duckdb/storage/checkpoint/write_overflow_strings_to_disk.hpp"
#include "duckdb/storage/table/validity_column_data.hpp"
#include "duckdb/storage/table/standard_column_data.hpp"
#include "duckdb/storage/table/list_column_data.hpp"
#include "duckdb/transaction/transaction.hpp"
#include "duckdb/storage/table/row_group.hpp"
#include "duckdb/storage/checkpoint/table_data_writer.hpp"
#include "duckdb/main/config.hpp"
namespace duckdb {
ColumnCheckpointState::ColumnCheckpointState(RowGroup &row_group, ColumnData &column_data, TableDataWriter &writer)
: row_group(row_group), column_data(column_data), writer(writer) {
}
ColumnCheckpointState::~ColumnCheckpointState() {
}
void ColumnCheckpointState::FlushSegment(unique_ptr<ColumnSegment> segment, idx_t segment_size) {
D_ASSERT(segment_size <= Storage::BLOCK_SIZE);
auto tuple_count = segment->count.load();
if (tuple_count == 0) { // LCOV_EXCL_START
return;
} // LCOV_EXCL_STOP
// merge the segment stats into the global stats
global_stats->Merge(*segment->stats.statistics);
// get the buffer of the segment and pin it
auto &db = column_data.GetDatabase();
auto &buffer_manager = BufferManager::GetBufferManager(db);
auto &block_manager = BlockManager::GetBlockManager(db);
auto &checkpoint_manager = writer.GetCheckpointManager();
bool block_is_constant = segment->stats.statistics->IsConstant();
block_id_t block_id = INVALID_BLOCK;
uint32_t offset_in_block = 0;
bool need_to_write = true;
PartialBlock *partial_block = nullptr;
unique_ptr<PartialBlock> owned_partial_block;
if (!block_is_constant) {
// non-constant block
// if the block is less than 80% full, we consider it a "partial block"
// which means we will try to fit it with other blocks
if (segment_size <= CheckpointManager::PARTIAL_BLOCK_THRESHOLD) {
// the block is a partial block
// check if there is a partial block available we can write to
if (checkpoint_manager.GetPartialBlock(segment.get(), segment_size, block_id, offset_in_block,
partial_block, owned_partial_block)) {
//! there is! increase the reference count of this block
block_manager.IncreaseBlockReferenceCount(block_id);
} else {
// there isn't: generate a new block for this segment
block_id = block_manager.GetFreeBlockId();
offset_in_block = 0;
need_to_write = false;
// now register this block as a partial block
checkpoint_manager.RegisterPartialBlock(segment.get(), segment_size, block_id);
}
} else {
// full block: get a free block to write to
block_id = block_manager.GetFreeBlockId();
offset_in_block = 0;
}
} else {
// constant block: no need to write anything to disk besides the stats
// set up the compression function to constant
auto &config = DBConfig::GetConfig(db);
segment->function =
config.GetCompressionFunction(CompressionType::COMPRESSION_CONSTANT, segment->type.InternalType());
}
// construct the data pointer
DataPointer data_pointer;
data_pointer.block_pointer.block_id = block_id;
data_pointer.block_pointer.offset = offset_in_block;
data_pointer.row_start = row_group.start;
if (!data_pointers.empty()) {
auto &last_pointer = data_pointers.back();
data_pointer.row_start = last_pointer.row_start + last_pointer.tuple_count;
}
data_pointer.tuple_count = tuple_count;
data_pointer.compression_type = segment->function->type;
data_pointer.statistics = segment->stats.statistics->Copy();
if (need_to_write) {
if (partial_block) {
// pin the current block
auto old_handle = buffer_manager.Pin(segment->block);
// pin the new block
auto new_handle = buffer_manager.Pin(partial_block->block);
// memcpy the contents of the old block to the new block
memcpy(new_handle->Ptr() + offset_in_block, old_handle->Ptr(), segment_size);
} else {
// convert the segment into a persistent segment that points to this block
segment->ConvertToPersistent(block_id);
}
}
if (owned_partial_block) {
// the partial block has become full: write it to disk
owned_partial_block->FlushToDisk(db);
}
// append the segment to the new segment tree
new_tree.AppendSegment(move(segment));
data_pointers.push_back(move(data_pointer));
}
void ColumnCheckpointState::FlushToDisk() {
auto &meta_writer = writer.GetMetaWriter();
meta_writer.Write<idx_t>(data_pointers.size());
// then write the data pointers themselves
for (idx_t k = 0; k < data_pointers.size(); k++) {
auto &data_pointer = data_pointers[k];
meta_writer.Write<idx_t>(data_pointer.row_start);
meta_writer.Write<idx_t>(data_pointer.tuple_count);
meta_writer.Write<block_id_t>(data_pointer.block_pointer.block_id);
meta_writer.Write<uint32_t>(data_pointer.block_pointer.offset);
meta_writer.Write<CompressionType>(data_pointer.compression_type);
data_pointer.statistics->Serialize(meta_writer);
}
}
} // namespace duckdb