Packages

An Elixir DuckDB library

Current section

Files

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

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

#include "duckdb/storage/table/column_segment.hpp"
#include "duckdb/common/limits.hpp"
#include "duckdb/storage/table/update_segment.hpp"
#include "duckdb/common/types/null_value.hpp"
#include "duckdb/common/types/vector.hpp"
#include "duckdb/common/vector_operations/vector_operations.hpp"
#include "duckdb/storage/table/column_segment.hpp"
#include "duckdb/storage/table/append_state.hpp"
#include "duckdb/storage/storage_manager.hpp"
#include "duckdb/planner/filter/conjunction_filter.hpp"
#include "duckdb/planner/filter/constant_filter.hpp"
#include "duckdb/planner/filter/null_filter.hpp"
#include "duckdb/main/config.hpp"
#include <cstring>
namespace duckdb {
unique_ptr<ColumnSegment> ColumnSegment::CreatePersistentSegment(DatabaseInstance &db, block_id_t block_id,
idx_t offset, const LogicalType &type, idx_t start,
idx_t count, CompressionType compression_type,
unique_ptr<BaseStatistics> statistics) {
auto &config = DBConfig::GetConfig(db);
CompressionFunction *function;
if (block_id == INVALID_BLOCK) {
function = config.GetCompressionFunction(CompressionType::COMPRESSION_CONSTANT, type.InternalType());
} else {
function = config.GetCompressionFunction(compression_type, type.InternalType());
}
return make_unique<ColumnSegment>(db, type, ColumnSegmentType::PERSISTENT, start, count, function, move(statistics),
block_id, offset);
}
unique_ptr<ColumnSegment> ColumnSegment::CreateTransientSegment(DatabaseInstance &db, const LogicalType &type,
idx_t start) {
auto &config = DBConfig::GetConfig(db);
auto function = config.GetCompressionFunction(CompressionType::COMPRESSION_UNCOMPRESSED, type.InternalType());
return make_unique<ColumnSegment>(db, type, ColumnSegmentType::TRANSIENT, start, 0, function, nullptr,
INVALID_BLOCK, 0);
}
ColumnSegment::ColumnSegment(DatabaseInstance &db, LogicalType type_p, ColumnSegmentType segment_type, idx_t start,
idx_t count, CompressionFunction *function_p, unique_ptr<BaseStatistics> statistics,
block_id_t block_id_p, idx_t offset_p)
: SegmentBase(start, count), db(db), type(move(type_p)), type_size(GetTypeIdSize(type.InternalType())),
segment_type(segment_type), function(function_p), stats(type, move(statistics)), block_id(block_id_p),
offset(offset_p) {
D_ASSERT(function);
auto &buffer_manager = BufferManager::GetBufferManager(db);
if (block_id == INVALID_BLOCK) {
// no block id specified
// there are two cases here:
// transient: allocate a buffer for the uncompressed segment
// persistent: constant segment, no need to allocate anything
if (segment_type == ColumnSegmentType::TRANSIENT) {
this->block = buffer_manager.RegisterMemory(Storage::BLOCK_SIZE, false);
}
} else {
D_ASSERT(segment_type == ColumnSegmentType::PERSISTENT);
this->block = buffer_manager.RegisterBlock(block_id);
}
if (function->init_segment) {
segment_state = function->init_segment(*this, block_id);
}
}
ColumnSegment::~ColumnSegment() {
}
//===--------------------------------------------------------------------===//
// Scan
//===--------------------------------------------------------------------===//
void ColumnSegment::InitializeScan(ColumnScanState &state) {
state.scan_state = function->init_scan(*this);
}
void ColumnSegment::Scan(ColumnScanState &state, idx_t scan_count, Vector &result, idx_t result_offset,
bool entire_vector) {
if (entire_vector) {
D_ASSERT(result_offset == 0);
Scan(state, scan_count, result);
} else {
D_ASSERT(result.GetVectorType() == VectorType::FLAT_VECTOR);
ScanPartial(state, scan_count, result, result_offset);
D_ASSERT(result.GetVectorType() == VectorType::FLAT_VECTOR);
}
}
void ColumnSegment::Skip(ColumnScanState &state) {
function->skip(*this, state, state.row_index - state.internal_index);
state.internal_index = state.row_index;
}
void ColumnSegment::Scan(ColumnScanState &state, idx_t scan_count, Vector &result) {
function->scan_vector(*this, state, scan_count, result);
}
void ColumnSegment::ScanPartial(ColumnScanState &state, idx_t scan_count, Vector &result, idx_t result_offset) {
function->scan_partial(*this, state, scan_count, result, result_offset);
}
//===--------------------------------------------------------------------===//
// Fetch
//===--------------------------------------------------------------------===//
void ColumnSegment::FetchRow(ColumnFetchState &state, row_t row_id, Vector &result, idx_t result_idx) {
function->fetch_row(*this, state, row_id - this->start, result, result_idx);
}
void ColumnSegment::InitializeAppend(ColumnAppendState &state) {
D_ASSERT(segment_type == ColumnSegmentType::TRANSIENT);
}
//===--------------------------------------------------------------------===//
// Append
//===--------------------------------------------------------------------===//
idx_t ColumnSegment::Append(ColumnAppendState &state, VectorData &append_data, idx_t offset, idx_t count) {
D_ASSERT(segment_type == ColumnSegmentType::TRANSIENT);
if (!function->append) {
throw InternalException("Attempting to append to a segment without append method");
}
return function->append(*this, stats, append_data, offset, count);
}
idx_t ColumnSegment::FinalizeAppend() {
D_ASSERT(segment_type == ColumnSegmentType::TRANSIENT);
if (!function->finalize_append) {
throw InternalException("Attempting to call FinalizeAppend on a segment without a finalize_append method");
}
return function->finalize_append(*this, stats);
}
void ColumnSegment::RevertAppend(idx_t start_row) {
D_ASSERT(segment_type == ColumnSegmentType::TRANSIENT);
if (function->revert_append) {
function->revert_append(*this, start_row);
}
this->count = start_row - this->start;
}
//===--------------------------------------------------------------------===//
// Convert To Persistent
//===--------------------------------------------------------------------===//
void ColumnSegment::ConvertToPersistent(block_id_t block_id_p) {
D_ASSERT(segment_type == ColumnSegmentType::TRANSIENT);
segment_type = ColumnSegmentType::PERSISTENT;
block_id = block_id_p;
offset = 0;
if (block_id == INVALID_BLOCK) {
// constant block: reset the block buffer
block.reset();
} else {
// non-constant block: write the block to disk
auto &buffer_manager = BufferManager::GetBufferManager(db);
auto &block_manager = BlockManager::GetBlockManager(db);
// the data for the block already exists in-memory of our block
// instead of copying the data we alter some metadata so the buffer points to an on-disk block
block = buffer_manager.ConvertToPersistent(block_manager, block_id, move(block));
}
segment_state.reset();
if (function->init_segment) {
segment_state = function->init_segment(*this, block_id);
}
}
void ColumnSegment::ConvertToPersistent(shared_ptr<BlockHandle> block_p, block_id_t block_id_p, uint32_t offset_p) {
D_ASSERT(segment_type == ColumnSegmentType::TRANSIENT);
segment_type = ColumnSegmentType::PERSISTENT;
block_id = block_id_p;
offset = offset_p;
block = move(block_p);
segment_state.reset();
if (function->init_segment) {
segment_state = function->init_segment(*this, block_id);
}
}
//===--------------------------------------------------------------------===//
// Filter Selection
//===--------------------------------------------------------------------===//
template <class T, class OP, bool HAS_NULL>
static idx_t TemplatedFilterSelection(T *vec, T *predicate, SelectionVector &sel, idx_t approved_tuple_count,
ValidityMask &mask, SelectionVector &result_sel) {
idx_t result_count = 0;
for (idx_t i = 0; i < approved_tuple_count; i++) {
auto idx = sel.get_index(i);
if ((!HAS_NULL || mask.RowIsValid(idx)) && OP::Operation(vec[idx], *predicate)) {
result_sel.set_index(result_count++, idx);
}
}
return result_count;
}
template <class T>
static void FilterSelectionSwitch(T *vec, T *predicate, SelectionVector &sel, idx_t &approved_tuple_count,
ExpressionType comparison_type, ValidityMask &mask) {
SelectionVector new_sel(approved_tuple_count);
// the inplace loops take the result as the last parameter
switch (comparison_type) {
case ExpressionType::COMPARE_EQUAL: {
if (mask.AllValid()) {
approved_tuple_count =
TemplatedFilterSelection<T, Equals, false>(vec, predicate, sel, approved_tuple_count, mask, new_sel);
} else {
approved_tuple_count =
TemplatedFilterSelection<T, Equals, true>(vec, predicate, sel, approved_tuple_count, mask, new_sel);
}
break;
}
case ExpressionType::COMPARE_LESSTHAN: {
if (mask.AllValid()) {
approved_tuple_count =
TemplatedFilterSelection<T, LessThan, false>(vec, predicate, sel, approved_tuple_count, mask, new_sel);
} else {
approved_tuple_count =
TemplatedFilterSelection<T, LessThan, true>(vec, predicate, sel, approved_tuple_count, mask, new_sel);
}
break;
}
case ExpressionType::COMPARE_GREATERTHAN: {
if (mask.AllValid()) {
approved_tuple_count = TemplatedFilterSelection<T, GreaterThan, false>(vec, predicate, sel,
approved_tuple_count, mask, new_sel);
} else {
approved_tuple_count = TemplatedFilterSelection<T, GreaterThan, true>(vec, predicate, sel,
approved_tuple_count, mask, new_sel);
}
break;
}
case ExpressionType::COMPARE_LESSTHANOREQUALTO: {
if (mask.AllValid()) {
approved_tuple_count = TemplatedFilterSelection<T, LessThanEquals, false>(
vec, predicate, sel, approved_tuple_count, mask, new_sel);
} else {
approved_tuple_count = TemplatedFilterSelection<T, LessThanEquals, true>(
vec, predicate, sel, approved_tuple_count, mask, new_sel);
}
break;
}
case ExpressionType::COMPARE_GREATERTHANOREQUALTO: {
if (mask.AllValid()) {
approved_tuple_count = TemplatedFilterSelection<T, GreaterThanEquals, false>(
vec, predicate, sel, approved_tuple_count, mask, new_sel);
} else {
approved_tuple_count = TemplatedFilterSelection<T, GreaterThanEquals, true>(
vec, predicate, sel, approved_tuple_count, mask, new_sel);
}
break;
}
default:
throw NotImplementedException("Unknown comparison type for filter pushed down to table!");
}
sel.Initialize(new_sel);
}
template <bool IS_NULL>
static idx_t TemplatedNullSelection(SelectionVector &sel, idx_t approved_tuple_count, ValidityMask &mask) {
if (mask.AllValid()) {
// no NULL values
if (IS_NULL) {
return 0;
} else {
return approved_tuple_count;
}
} else {
SelectionVector result_sel(approved_tuple_count);
idx_t result_count = 0;
for (idx_t i = 0; i < approved_tuple_count; i++) {
auto idx = sel.get_index(i);
if (mask.RowIsValid(idx) != IS_NULL) {
result_sel.set_index(result_count++, idx);
}
}
sel.Initialize(result_sel);
return result_count;
}
}
void ColumnSegment::FilterSelection(SelectionVector &sel, Vector &result, const TableFilter &filter,
idx_t &approved_tuple_count, ValidityMask &mask) {
switch (filter.filter_type) {
case TableFilterType::CONJUNCTION_AND: {
auto &conjunction_and = (ConjunctionAndFilter &)filter;
for (auto &child_filter : conjunction_and.child_filters) {
FilterSelection(sel, result, *child_filter, approved_tuple_count, mask);
}
break;
}
case TableFilterType::CONSTANT_COMPARISON: {
auto &constant_filter = (ConstantFilter &)filter;
// the inplace loops take the result as the last parameter
switch (result.GetType().InternalType()) {
case PhysicalType::UINT8: {
auto result_flat = FlatVector::GetData<uint8_t>(result);
Vector predicate_vector(constant_filter.constant);
auto predicate = FlatVector::GetData<uint8_t>(predicate_vector);
FilterSelectionSwitch<uint8_t>(result_flat, predicate, sel, approved_tuple_count,
constant_filter.comparison_type, mask);
break;
}
case PhysicalType::UINT16: {
auto result_flat = FlatVector::GetData<uint16_t>(result);
Vector predicate_vector(constant_filter.constant);
auto predicate = FlatVector::GetData<uint16_t>(predicate_vector);
FilterSelectionSwitch<uint16_t>(result_flat, predicate, sel, approved_tuple_count,
constant_filter.comparison_type, mask);
break;
}
case PhysicalType::UINT32: {
auto result_flat = FlatVector::GetData<uint32_t>(result);
Vector predicate_vector(constant_filter.constant);
auto predicate = FlatVector::GetData<uint32_t>(predicate_vector);
FilterSelectionSwitch<uint32_t>(result_flat, predicate, sel, approved_tuple_count,
constant_filter.comparison_type, mask);
break;
}
case PhysicalType::UINT64: {
auto result_flat = FlatVector::GetData<uint64_t>(result);
Vector predicate_vector(constant_filter.constant);
auto predicate = FlatVector::GetData<uint64_t>(predicate_vector);
FilterSelectionSwitch<uint64_t>(result_flat, predicate, sel, approved_tuple_count,
constant_filter.comparison_type, mask);
break;
}
case PhysicalType::INT8: {
auto result_flat = FlatVector::GetData<int8_t>(result);
Vector predicate_vector(constant_filter.constant);
auto predicate = FlatVector::GetData<int8_t>(predicate_vector);
FilterSelectionSwitch<int8_t>(result_flat, predicate, sel, approved_tuple_count,
constant_filter.comparison_type, mask);
break;
}
case PhysicalType::INT16: {
auto result_flat = FlatVector::GetData<int16_t>(result);
Vector predicate_vector(constant_filter.constant);
auto predicate = FlatVector::GetData<int16_t>(predicate_vector);
FilterSelectionSwitch<int16_t>(result_flat, predicate, sel, approved_tuple_count,
constant_filter.comparison_type, mask);
break;
}
case PhysicalType::INT32: {
auto result_flat = FlatVector::GetData<int32_t>(result);
Vector predicate_vector(constant_filter.constant);
auto predicate = FlatVector::GetData<int32_t>(predicate_vector);
FilterSelectionSwitch<int32_t>(result_flat, predicate, sel, approved_tuple_count,
constant_filter.comparison_type, mask);
break;
}
case PhysicalType::INT64: {
auto result_flat = FlatVector::GetData<int64_t>(result);
Vector predicate_vector(constant_filter.constant);
auto predicate = FlatVector::GetData<int64_t>(predicate_vector);
FilterSelectionSwitch<int64_t>(result_flat, predicate, sel, approved_tuple_count,
constant_filter.comparison_type, mask);
break;
}
case PhysicalType::INT128: {
auto result_flat = FlatVector::GetData<hugeint_t>(result);
Vector predicate_vector(constant_filter.constant);
auto predicate = FlatVector::GetData<hugeint_t>(predicate_vector);
FilterSelectionSwitch<hugeint_t>(result_flat, predicate, sel, approved_tuple_count,
constant_filter.comparison_type, mask);
break;
}
case PhysicalType::FLOAT: {
auto result_flat = FlatVector::GetData<float>(result);
Vector predicate_vector(constant_filter.constant);
auto predicate = FlatVector::GetData<float>(predicate_vector);
FilterSelectionSwitch<float>(result_flat, predicate, sel, approved_tuple_count,
constant_filter.comparison_type, mask);
break;
}
case PhysicalType::DOUBLE: {
auto result_flat = FlatVector::GetData<double>(result);
Vector predicate_vector(constant_filter.constant);
auto predicate = FlatVector::GetData<double>(predicate_vector);
FilterSelectionSwitch<double>(result_flat, predicate, sel, approved_tuple_count,
constant_filter.comparison_type, mask);
break;
}
case PhysicalType::VARCHAR: {
auto result_flat = FlatVector::GetData<string_t>(result);
Vector predicate_vector(constant_filter.constant.str_value);
auto predicate = FlatVector::GetData<string_t>(predicate_vector);
FilterSelectionSwitch<string_t>(result_flat, predicate, sel, approved_tuple_count,
constant_filter.comparison_type, mask);
break;
}
case PhysicalType::BOOL: {
auto result_flat = FlatVector::GetData<bool>(result);
Vector predicate_vector(constant_filter.constant);
auto predicate = FlatVector::GetData<bool>(predicate_vector);
FilterSelectionSwitch<bool>(result_flat, predicate, sel, approved_tuple_count,
constant_filter.comparison_type, mask);
break;
}
default:
throw InvalidTypeException(result.GetType(), "Invalid type for filter pushed down to table comparison");
}
break;
}
case TableFilterType::IS_NULL:
TemplatedNullSelection<true>(sel, approved_tuple_count, mask);
break;
case TableFilterType::IS_NOT_NULL:
TemplatedNullSelection<false>(sel, approved_tuple_count, mask);
break;
default:
throw InternalException("FIXME: unsupported type for filter selection");
}
}
} // namespace duckdb