ClickHouse/src/Storages/MergeTree/MergeTreeBlockReadUtils.cpp

329 lines
12 KiB
C++
Raw Normal View History

#include <Storages/MergeTree/MergeTreeBlockReadUtils.h>
#include <Storages/MergeTree/MergeTreeData.h>
2020-10-23 17:57:17 +00:00
#include <Core/NamesAndTypes.h>
2020-09-16 13:24:07 +00:00
#include <Common/checkStackSize.h>
#include <Common/typeid_cast.h>
#include <Columns/ColumnConst.h>
#include <unordered_set>
2018-01-10 00:04:08 +00:00
namespace DB
{
2020-02-25 18:10:48 +00:00
namespace ErrorCodes
{
extern const int LOGICAL_ERROR;
extern const int NO_SUCH_COLUMN_IN_TABLE;
2020-02-25 18:10:48 +00:00
}
2020-09-15 11:17:58 +00:00
namespace
{
/// Columns absent in part may depend on other absent columns so we are
/// searching all required physical columns recursively. Return true if found at
/// least one existing (physical) column in part.
2020-09-15 11:17:58 +00:00
bool injectRequiredColumnsRecursively(
const String & column_name,
const ColumnsDescription & storage_columns,
const MergeTreeData::AlterConversions & alter_conversions,
const MergeTreeData::DataPartPtr & part,
Names & columns,
NameSet & required_columns,
NameSet & injected_columns)
{
2020-09-16 13:24:07 +00:00
/// This is needed to prevent stack overflow in case of cyclic defaults or
/// huge AST which for some reason was not validated on parsing/interpreter
/// stages.
checkStackSize();
2020-09-15 11:17:58 +00:00
2020-10-23 17:57:17 +00:00
if (storage_columns.hasPhysicalOrSubcolumn(column_name))
2020-09-15 11:17:58 +00:00
{
2020-10-23 17:57:17 +00:00
auto column_in_storage = storage_columns.getPhysicalOrSubcolumn(column_name);
2020-12-22 15:03:48 +00:00
auto column_name_in_part = column_in_storage.getNameInStorage();
2020-10-23 17:57:17 +00:00
if (alter_conversions.isColumnRenamed(column_name_in_part))
column_name_in_part = alter_conversions.getColumnOldName(column_name_in_part);
auto column_in_part = NameAndTypePair(
column_name_in_part, column_in_storage.getSubcolumnName(),
2020-12-22 15:03:48 +00:00
column_in_storage.getTypeInStorage(), column_in_storage.type);
2020-10-23 17:57:17 +00:00
/// column has files and hence does not require evaluation
if (part->hasColumnFiles(column_in_part))
2020-09-15 11:17:58 +00:00
{
2020-10-23 17:57:17 +00:00
/// ensure each column is added only once
if (required_columns.count(column_name) == 0)
{
columns.emplace_back(column_name);
required_columns.emplace(column_name);
injected_columns.emplace(column_name);
}
return true;
2020-09-15 11:17:58 +00:00
}
}
/// Column doesn't have default value and don't exist in part
/// don't need to add to required set.
const auto column_default = storage_columns.getDefault(column_name);
if (!column_default)
return false;
/// collect identifiers required for evaluation
IdentifierNameSet identifiers;
column_default->expression->collectIdentifierNames(identifiers);
bool result = false;
for (const auto & identifier : identifiers)
result |= injectRequiredColumnsRecursively(identifier, storage_columns, alter_conversions, part, columns, required_columns, injected_columns);
return result;
}
}
2019-11-21 16:10:22 +00:00
NameSet injectRequiredColumns(const MergeTreeData & storage, const StorageMetadataPtr & metadata_snapshot, const MergeTreeData::DataPartPtr & part, Names & columns)
{
NameSet required_columns{std::begin(columns), std::end(columns)};
NameSet injected_columns;
2020-09-15 11:17:58 +00:00
bool have_at_least_one_physical_column = false;
const auto & storage_columns = metadata_snapshot->getColumns();
MergeTreeData::AlterConversions alter_conversions;
if (!part->isProjectionPart())
alter_conversions = storage.getAlterConversionsForPart(part);
for (size_t i = 0; i < columns.size(); ++i)
{
/// We are going to fetch only physical columns
if (!storage_columns.hasPhysicalOrSubcolumn(columns[i]))
throw Exception("There is no physical column or subcolumn " + columns[i] + " in table.", ErrorCodes::NO_SUCH_COLUMN_IN_TABLE);
2020-09-15 11:17:58 +00:00
have_at_least_one_physical_column |= injectRequiredColumnsRecursively(
columns[i], storage_columns, alter_conversions,
part, columns, required_columns, injected_columns);
}
/** Add a column of the minimum size.
* Used in case when no column is needed or files are missing, but at least you need to know number of rows.
* Adds to the columns.
*/
2020-09-15 11:17:58 +00:00
if (!have_at_least_one_physical_column)
{
2020-10-27 11:04:03 +00:00
const auto minimum_size_column_name = part->getColumnNameWithMinimumCompressedSize(metadata_snapshot);
columns.push_back(minimum_size_column_name);
/// correctly report added column
injected_columns.insert(columns.back());
}
return injected_columns;
}
MergeTreeReadTask::MergeTreeReadTask(
2019-08-03 11:02:40 +00:00
const MergeTreeData::DataPartPtr & data_part_, const MarkRanges & mark_ranges_, const size_t part_index_in_query_,
const Names & ordered_names_, const NameSet & column_name_set_, const NamesAndTypesList & columns_,
const NamesAndTypesList & pre_columns_, const bool remove_prewhere_column_, const bool should_reorder_,
MergeTreeBlockSizePredictorPtr && size_predictor_)
2019-08-03 11:02:40 +00:00
: data_part{data_part_}, mark_ranges{mark_ranges_}, part_index_in_query{part_index_in_query_},
ordered_names{ordered_names_}, column_name_set{column_name_set_}, columns{columns_}, pre_columns{pre_columns_},
remove_prewhere_column{remove_prewhere_column_}, should_reorder{should_reorder_}, size_predictor{std::move(size_predictor_)}
{
}
2017-04-05 20:34:19 +00:00
MergeTreeReadTask::~MergeTreeReadTask() = default;
MergeTreeBlockSizePredictor::MergeTreeBlockSizePredictor(
const MergeTreeData::DataPartPtr & data_part_, const Names & columns, const Block & sample_block)
: data_part(data_part_)
{
number_of_rows_in_part = data_part->rows_count;
2019-01-22 19:56:53 +00:00
/// Initialize with sample block until update won't called.
2019-09-26 17:29:41 +00:00
initialize(sample_block, {}, columns);
}
2019-09-26 17:29:41 +00:00
void MergeTreeBlockSizePredictor::initialize(const Block & sample_block, const Columns & columns, const Names & names, bool from_update)
{
fixed_columns_bytes_per_row = 0;
dynamic_columns_infos.clear();
std::unordered_set<String> names_set;
if (!from_update)
2019-09-26 17:29:41 +00:00
names_set.insert(names.begin(), names.end());
2019-09-26 17:29:41 +00:00
size_t num_columns = sample_block.columns();
for (size_t pos = 0; pos < num_columns; ++pos)
{
2019-09-26 17:29:41 +00:00
const auto & column_with_type_and_name = sample_block.getByPosition(pos);
const String & column_name = column_with_type_and_name.name;
2019-09-26 17:29:41 +00:00
const ColumnPtr & column_data = from_update ? columns[pos]
: column_with_type_and_name.column;
if (!from_update && !names_set.count(column_name))
continue;
/// At least PREWHERE filter column might be const.
if (typeid_cast<const ColumnConst *>(column_data.get()))
continue;
if (column_data->valuesHaveFixedSize())
{
size_t size_of_value = column_data->sizeOfValueIfFixed();
fixed_columns_bytes_per_row += column_data->sizeOfValueIfFixed();
max_size_per_row_fixed = std::max<size_t>(max_size_per_row_fixed, size_of_value);
}
else
{
ColumnInfo info;
info.name = column_name;
/// If column isn't fixed and doesn't have checksum, than take first
ColumnSize column_size = data_part->getColumnSize(
column_name, *column_with_type_and_name.type);
info.bytes_per_row_global = column_size.data_uncompressed
? column_size.data_uncompressed / number_of_rows_in_part
: column_data->byteSize() / std::max<size_t>(1, column_data->size());
dynamic_columns_infos.emplace_back(info);
}
}
bytes_per_row_global = fixed_columns_bytes_per_row;
for (auto & info : dynamic_columns_infos)
{
info.bytes_per_row = info.bytes_per_row_global;
bytes_per_row_global += info.bytes_per_row_global;
max_size_per_row_dynamic = std::max<double>(max_size_per_row_dynamic, info.bytes_per_row);
}
bytes_per_row_current = bytes_per_row_global;
}
void MergeTreeBlockSizePredictor::startBlock()
{
block_size_bytes = 0;
block_size_rows = 0;
for (auto & info : dynamic_columns_infos)
info.size_bytes = 0;
}
/// TODO: add last_read_row_in_part parameter to take into account gaps between adjacent ranges
2019-09-26 17:29:41 +00:00
void MergeTreeBlockSizePredictor::update(const Block & sample_block, const Columns & columns, size_t num_rows, double decay)
{
2019-09-26 17:29:41 +00:00
if (columns.size() != sample_block.columns())
throw Exception("Inconsistent number of columns passed to MergeTreeBlockSizePredictor. "
"Have " + toString(sample_block.columns()) + " in sample block "
"and " + toString(columns.size()) + " columns in list", ErrorCodes::LOGICAL_ERROR);
if (!is_initialized_in_update)
{
/// Reinitialize with read block to update estimation for DEFAULT and MATERIALIZED columns without data.
2019-09-26 17:29:41 +00:00
initialize(sample_block, columns, {}, true);
is_initialized_in_update = true;
}
2019-09-26 17:29:41 +00:00
if (num_rows < block_size_rows)
{
2019-09-26 17:29:41 +00:00
throw Exception("Updated block has less rows (" + toString(num_rows) + ") than previous one (" + toString(block_size_rows) + ")",
ErrorCodes::LOGICAL_ERROR);
}
2019-09-26 17:29:41 +00:00
size_t diff_rows = num_rows - block_size_rows;
block_size_bytes = num_rows * fixed_columns_bytes_per_row;
bytes_per_row_current = fixed_columns_bytes_per_row;
2019-09-26 17:29:41 +00:00
block_size_rows = num_rows;
/// Make recursive updates for each read row: v_{i+1} = (1 - decay) v_{i} + decay v_{target}
2019-01-22 19:56:53 +00:00
/// Use sum of geometric sequence formula to update multiple rows: v{n} = (1 - decay)^n v_{0} + (1 - (1 - decay)^n) v_{target}
/// NOTE: DEFAULT and MATERIALIZED columns without data has inaccurate estimation of v_{target}
2017-04-15 03:32:33 +00:00
double alpha = std::pow(1. - decay, diff_rows);
max_size_per_row_dynamic = 0;
for (auto & info : dynamic_columns_infos)
{
2019-09-26 17:29:41 +00:00
size_t new_size = columns[sample_block.getPositionByName(info.name)]->byteSize();
2017-04-15 03:32:33 +00:00
size_t diff_size = new_size - info.size_bytes;
2017-04-15 03:32:33 +00:00
double local_bytes_per_row = static_cast<double>(diff_size) / diff_rows;
info.bytes_per_row = alpha * info.bytes_per_row + (1. - alpha) * local_bytes_per_row;
info.size_bytes = new_size;
block_size_bytes += new_size;
bytes_per_row_current += info.bytes_per_row;
max_size_per_row_dynamic = std::max<double>(max_size_per_row_dynamic, info.bytes_per_row);
}
}
2019-07-19 14:56:00 +00:00
MergeTreeReadTaskColumns getReadTaskColumns(
const MergeTreeData & storage,
const StorageMetadataPtr & metadata_snapshot,
const MergeTreeData::DataPartPtr & data_part,
const Names & required_columns,
const PrewhereInfoPtr & prewhere_info,
bool check_columns)
2019-07-19 14:56:00 +00:00
{
Names column_names = required_columns;
Names pre_column_names;
/// inject columns required for defaults evaluation
bool should_reorder = !injectRequiredColumns(storage, metadata_snapshot, data_part, column_names).empty();
2019-07-19 14:56:00 +00:00
if (prewhere_info)
2019-07-19 14:56:00 +00:00
{
if (prewhere_info->alias_actions)
2021-02-15 19:48:06 +00:00
pre_column_names = prewhere_info->alias_actions->getRequiredColumns();
2021-02-20 11:00:16 +00:00
else
{
2021-02-15 19:48:06 +00:00
pre_column_names = prewhere_info->prewhere_actions->getRequiredColumns();
2019-07-19 14:56:00 +00:00
2021-02-20 11:00:16 +00:00
if (prewhere_info->row_level_filter)
{
NameSet names(pre_column_names.begin(), pre_column_names.end());
for (auto & name : prewhere_info->row_level_filter->getRequiredColumns())
{
if (names.count(name) == 0)
pre_column_names.push_back(name);
}
}
}
2019-07-19 14:56:00 +00:00
if (pre_column_names.empty())
pre_column_names.push_back(column_names[0]);
const auto injected_pre_columns = injectRequiredColumns(storage, metadata_snapshot, data_part, pre_column_names);
2019-07-19 14:56:00 +00:00
if (!injected_pre_columns.empty())
should_reorder = true;
const NameSet pre_name_set(pre_column_names.begin(), pre_column_names.end());
Names post_column_names;
for (const auto & name : column_names)
if (!pre_name_set.count(name))
post_column_names.push_back(name);
column_names = post_column_names;
}
MergeTreeReadTaskColumns result;
if (check_columns)
{
const NamesAndTypesList & physical_columns = metadata_snapshot->getColumns().getAllWithSubcolumns();
result.pre_columns = physical_columns.addTypes(pre_column_names);
result.columns = physical_columns.addTypes(column_names);
2019-07-19 14:56:00 +00:00
}
else
{
2020-01-16 16:15:01 +00:00
result.pre_columns = data_part->getColumns().addTypes(pre_column_names);
result.columns = data_part->getColumns().addTypes(column_names);
2019-07-19 14:56:00 +00:00
}
result.should_reorder = should_reorder;
return result;
}
}