thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-21 23:21:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

allow to use constant adaptive granularity for part

Browse Source
This commit is contained in:
Anton Popov 2024-11-11 16:54:38 +00:00
parent c16e1f021b
commit c7ad0b897a
43 changed files with 844 additions and 376 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/Processors/QueryPlan/Optimizations/optimizeUseAggregateProjection.cpp
View File

@ -647,7 +647,7 @@ std::optional<String> optimizeUseAggregateProjections(QueryPlan::Node & node, Qu
range.begin = exact_ranges[i].end;
ordinary_reading_marks -= exact_ranges[i].end - exact_ranges[i].begin;
exact_count += part_with_ranges.data_part->index_granularity.getRowsCountInRange(exact_ranges[i]);
exact_count += part_with_ranges.data_part->index_granularity->getRowsCountInRange(exact_ranges[i]);
++i;
}

6
src/Processors/QueryPlan/PartsSplitter.cpp
View File

@ -201,7 +201,7 @@ public:
size_t getMarkRows(size_t part_idx, size_t mark) const
{
return parts[part_idx].data_part->index_granularity.getMarkRows(mark);
return parts[part_idx].data_part->index_granularity->getMarkRows(mark);
}
private:
const RangesInDataParts & parts;
@ -444,7 +444,7 @@ SplitPartsRangesResult splitPartsRanges(RangesInDataParts ranges_in_data_parts,
parts_ranges.push_back(
{index_access.getValue(part_index, range.begin), range, part_index, PartsRangesIterator::EventType::RangeStart});
const bool value_is_defined_at_end_mark = range.end < index_granularity.getMarksCount();
const bool value_is_defined_at_end_mark = range.end < index_granularity->getMarksCount();
if (!value_is_defined_at_end_mark)
continue;
@ -667,7 +667,7 @@ std::pair<std::vector<RangesInDataParts>, std::vector<Values>> splitIntersecting
PartRangeIndex parts_range_start_index(parts_range_start);
parts_ranges_queue.push({std::move(parts_range_start), std::move(parts_range_start_index)});
const bool value_is_defined_at_end_mark = range.end < index_granularity.getMarksCount();
const bool value_is_defined_at_end_mark = range.end < index_granularity->getMarksCount();
if (!value_is_defined_at_end_mark)
continue;

4
src/Processors/QueryPlan/ReadFromMergeTree.cpp
View File

@ -663,7 +663,7 @@ Pipe ReadFromMergeTree::readInOrder(
part_with_ranges.ranges.size(),
read_type == ReadType::InReverseOrder ? " reverse " : " ",
part_with_ranges.data_part->name, total_rows,
part_with_ranges.data_part->index_granularity.getMarkStartingRow(part_with_ranges.ranges.front().begin));
part_with_ranges.data_part->index_granularity->getMarkStartingRow(part_with_ranges.ranges.front().begin));
MergeTreeSelectAlgorithmPtr algorithm;
if (read_type == ReadType::InReverseOrder)
@ -1727,7 +1727,7 @@ ReadFromMergeTree::AnalysisResultPtr ReadFromMergeTree::selectRangesToRead(
return std::make_shared<AnalysisResult>(std::move(result));
for (const auto & part : parts)
total_marks_pk += part->index_granularity.getMarksCountWithoutFinal();
total_marks_pk += part->index_granularity->getMarksCountWithoutFinal();
parts_before_pk = parts.size();
auto reader_settings = getMergeTreeReaderSettings(context_, query_info_);

30
src/Storages/MergeTree/IMergeTreeDataPart.cpp
View File

@ -3,6 +3,7 @@
#include <base/types.h>
#include <exception>
#include <memory>
#include <optional>
#include <string_view>
#include <Compression/CompressedReadBuffer.h>
@ -38,6 +39,8 @@
#include <Common/StringUtils.h>
#include <Common/escapeForFileName.h>
#include <Common/logger_useful.h>
#include "Storages/MergeTree/MergeTreeIndexGranularity.h"
#include "Storages/MergeTree/MergeTreeIndexGranularityAdaptive.h"
#include <Disks/IO/CachedOnDiskReadBufferFromFile.h>
@ -652,7 +655,7 @@ void IMergeTreeDataPart::assertOnDisk() const
UInt64 IMergeTreeDataPart::getMarksCount() const
{
return index_granularity.getMarksCount();
return index_granularity->getMarksCount();
}
UInt64 IMergeTreeDataPart::getExistingBytesOnDisk() const
@ -931,13 +934,13 @@ void IMergeTreeDataPart::loadIndex() const
for (size_t i = 0; i < key_size; ++i)
{
loaded_index[i] = primary_key.data_types[i]->createColumn();
loaded_index[i]->reserve(index_granularity.getMarksCount());
loaded_index[i]->reserve(index_granularity->getMarksCount());
}
String index_name = "primary" + getIndexExtensionFromFilesystem(getDataPartStorage());
String index_path = fs::path(getDataPartStorage().getRelativePath()) / index_name;
auto index_file = metadata_manager->read(index_name);
size_t marks_count = index_granularity.getMarksCount();
size_t marks_count = index_granularity->getMarksCount();
Serializations key_serializations(key_size);
for (size_t j = 0; j < key_size; ++j)
@ -1350,7 +1353,7 @@ void IMergeTreeDataPart::loadRowsCount()
assertEOF(*buf);
};
if (index_granularity.empty())
if (index_granularity->empty())
{
rows_count = 0;
}
@ -1385,9 +1388,9 @@ void IMergeTreeDataPart::loadRowsCount()
backQuote(column.name), rows_in_column, name, rows_count);
}
size_t last_possibly_incomplete_mark_rows = index_granularity.getLastNonFinalMarkRows();
size_t last_possibly_incomplete_mark_rows = index_granularity->getLastNonFinalMarkRows();
/// All this rows have to be written in column
size_t index_granularity_without_last_mark = index_granularity.getTotalRows() - last_possibly_incomplete_mark_rows;
size_t index_granularity_without_last_mark = index_granularity->getTotalRows() - last_possibly_incomplete_mark_rows;
/// We have more rows in column than in index granularity without last possibly incomplete mark
if (rows_in_column < index_granularity_without_last_mark)
{
@ -1397,7 +1400,7 @@ void IMergeTreeDataPart::loadRowsCount()
"and size of single value, "
"but index granularity in part {} without last mark has {} rows, which "
"is more than in column",
backQuote(column.name), rows_in_column, name, index_granularity.getTotalRows());
backQuote(column.name), rows_in_column, name, index_granularity->getTotalRows());
}
/// In last mark we actually written less or equal rows than stored in last mark of index granularity
@ -1445,8 +1448,8 @@ void IMergeTreeDataPart::loadRowsCount()
column.name, column_size, sizeof_field);
}
size_t last_mark_index_granularity = index_granularity.getLastNonFinalMarkRows();
size_t rows_approx = index_granularity.getTotalRows();
size_t last_mark_index_granularity = index_granularity->getLastNonFinalMarkRows();
size_t rows_approx = index_granularity->getTotalRows();
if (!(rows_count <= rows_approx && rows_approx < rows_count + last_mark_index_granularity))
throw Exception(ErrorCodes::LOGICAL_ERROR, "Unexpected size of column {}: "
"{} rows, expected {}+-{} rows according to the index",
@ -1509,7 +1512,7 @@ UInt64 IMergeTreeDataPart::readExistingRowsCount()
while (current_row < rows_count)
{
size_t rows_to_read = index_granularity.getMarkRows(current_mark);
size_t rows_to_read = index_granularity->getMarkRows(current_mark);
continue_reading = (current_mark != 0);
Columns result;
@ -1957,6 +1960,9 @@ void IMergeTreeDataPart::initializeIndexGranularityInfo()
index_granularity_info = MergeTreeIndexGranularityInfo(storage, *mrk_type);
else
index_granularity_info = MergeTreeIndexGranularityInfo(storage, part_type);
/// It may be converted to constant index granularity after loading it.
index_granularity = std::make_unique<MergeTreeIndexGranularityAdaptive>();
}
void IMergeTreeDataPart::remove()
@ -2230,9 +2236,9 @@ void IMergeTreeDataPart::checkConsistency(bool require_part_metadata) const
"part_state: [{}]",
columns.toString(),
index_granularity_info.getMarkSizeInBytes(columns.size()),
index_granularity.getMarksCount(),
index_granularity->getMarksCount(),
index_granularity_info.describe(),
index_granularity.describe(),
index_granularity->describe(),
part_state);
e.addMessage(debug_info);

2
src/Storages/MergeTree/IMergeTreeDataPart.h
View File

@ -321,7 +321,7 @@ public:
/// Amount of rows between marks
/// As index always loaded into memory
MergeTreeIndexGranularity index_granularity;
MergeTreeIndexGranularityPtr index_granularity;
/// Index that for each part stores min and max values of a set of columns. This allows quickly excluding
/// parts based on conditions on these columns imposed by a query.

40
src/Storages/MergeTree/IMergeTreeDataPartWriter.cpp
View File

@ -1,5 +1,11 @@
#include <memory>
#include <optional>
#include <Storages/MergeTree/IMergeTreeDataPartWriter.h>
#include <Common/MemoryTrackerBlockerInThread.h>
#include "Storages/MergeTree/MergeTreeIndexGranularity.h"
#include "Storages/MergeTree/MergeTreeIndexGranularityAdaptive.h"
#include "Storages/MergeTree/MergeTreeIndexGranularityConstant.h"
#include "Storages/MergeTree/MergeTreeSettings.h"
#include <Columns/ColumnSparse.h>
namespace DB
@ -11,7 +17,6 @@ namespace ErrorCodes
extern const int NO_SUCH_COLUMN_IN_TABLE;
}
Block getIndexBlockAndPermute(const Block & block, const Names & names, const IColumn::Permutation * permutation)
{
Block result;
@ -57,7 +62,7 @@ IMergeTreeDataPartWriter::IMergeTreeDataPartWriter(
const StorageMetadataPtr & metadata_snapshot_,
const VirtualsDescriptionPtr & virtual_columns_,
const MergeTreeWriterSettings & settings_,
const MergeTreeIndexGranularity & index_granularity_)
MergeTreeIndexGranularityPtr index_granularity_)
: data_part_name(data_part_name_)
, serializations(serializations_)
, index_granularity_info(index_granularity_info_)
@ -68,7 +73,7 @@ IMergeTreeDataPartWriter::IMergeTreeDataPartWriter(
, settings(settings_)
, with_final_mark(settings.can_use_adaptive_granularity)
, data_part_storage(data_part_storage_)
, index_granularity(index_granularity_)
, index_granularity(std::move(index_granularity_))
{
}
@ -145,7 +150,7 @@ MergeTreeDataPartWriterPtr createMergeTreeDataPartCompactWriter(
const String & marks_file_extension_,
const CompressionCodecPtr & default_codec_,
const MergeTreeWriterSettings & writer_settings,
const MergeTreeIndexGranularity & computed_index_granularity);
MergeTreeIndexGranularityPtr computed_index_granularity);
MergeTreeDataPartWriterPtr createMergeTreeDataPartWideWriter(
const String & data_part_name_,
@ -162,8 +167,7 @@ MergeTreeDataPartWriterPtr createMergeTreeDataPartWideWriter(
const String & marks_file_extension_,
const CompressionCodecPtr & default_codec_,
const MergeTreeWriterSettings & writer_settings,
const MergeTreeIndexGranularity & computed_index_granularity);
MergeTreeIndexGranularityPtr computed_index_granularity);
MergeTreeDataPartWriterPtr createMergeTreeDataPartWriter(
MergeTreeDataPartType part_type,
@ -182,12 +186,26 @@ MergeTreeDataPartWriterPtr createMergeTreeDataPartWriter(
const String & marks_file_extension_,
const CompressionCodecPtr & default_codec_,
const MergeTreeWriterSettings & writer_settings,
const MergeTreeIndexGranularity & computed_index_granularity)
MergeTreeIndexGranularityPtr computed_index_granularity)
{
if (part_type == MergeTreeDataPartType::Compact)
return createMergeTreeDataPartCompactWriter(data_part_name_, logger_name_, serializations_, data_part_storage_,
index_granularity_info_, storage_settings_, columns_list, column_positions, metadata_snapshot, virtual_columns, indices_to_recalc, stats_to_recalc_,
marks_file_extension_, default_codec_, writer_settings, computed_index_granularity);
return createMergeTreeDataPartCompactWriter(
data_part_name_,
logger_name_,
serializations_,
data_part_storage_,
index_granularity_info_,
storage_settings_,
columns_list,
column_positions,
metadata_snapshot,
virtual_columns,
indices_to_recalc,
stats_to_recalc_,
marks_file_extension_,
default_codec_,
writer_settings,
std::move(computed_index_granularity));
if (part_type == MergeTreeDataPartType::Wide)
return createMergeTreeDataPartWideWriter(
data_part_name_,
@ -204,7 +222,7 @@ MergeTreeDataPartWriterPtr createMergeTreeDataPartWriter(
marks_file_extension_,
default_codec_,
writer_settings,
computed_index_granularity);
std::move(computed_index_granularity));
throw Exception(ErrorCodes::LOGICAL_ERROR, "Unknown part type: {}", part_type.toString());
}

8
src/Storages/MergeTree/IMergeTreeDataPartWriter.h
View File

@ -36,7 +36,7 @@ public:
const StorageMetadataPtr & metadata_snapshot_,
const VirtualsDescriptionPtr & virtual_columns_,
const MergeTreeWriterSettings & settings_,
const MergeTreeIndexGranularity & index_granularity_ = {});
MergeTreeIndexGranularityPtr index_granularity_);
virtual ~IMergeTreeDataPartWriter();
@ -52,7 +52,7 @@ public:
PlainMarksByName releaseCachedMarks();
const MergeTreeIndexGranularity & getIndexGranularity() const { return index_granularity; }
MergeTreeIndexGranularityPtr getIndexGranularity() const { return index_granularity; }
protected:
SerializationPtr getSerialization(const String & column_name) const;
@ -74,7 +74,7 @@ protected:
MutableDataPartStoragePtr data_part_storage;
MutableColumns index_columns;
MergeTreeIndexGranularity index_granularity;
MergeTreeIndexGranularityPtr index_granularity;
/// Marks that will be saved to cache on finish.
PlainMarksByName cached_marks;
};
@ -99,6 +99,6 @@ MergeTreeDataPartWriterPtr createMergeTreeDataPartWriter(
const String & marks_file_extension,
const CompressionCodecPtr & default_codec_,
const MergeTreeWriterSettings & writer_settings,
const MergeTreeIndexGranularity & computed_index_granularity);
MergeTreeIndexGranularityPtr computed_index_granularity);
}

2
src/Storages/MergeTree/IMergedBlockOutputStream.h
View File

@ -29,7 +29,7 @@ public:
virtual void write(const Block & block) = 0;
const MergeTreeIndexGranularity & getIndexGranularity() const
MergeTreeIndexGranularityPtr getIndexGranularity() const
{
return writer->getIndexGranularity();
}

2
src/Storages/MergeTree/LoadedMergeTreeDataPartInfoForReader.h
View File

@ -51,7 +51,7 @@ public:
const MergeTreeIndexGranularityInfo & getIndexGranularityInfo() const override { return data_part->index_granularity_info; }
const MergeTreeIndexGranularity & getIndexGranularity() const override { return data_part->index_granularity; }
const MergeTreeIndexGranularity & getIndexGranularity() const override { return *data_part->index_granularity; }
const SerializationInfoByName & getSerializationInfos() const override { return data_part->getSerializationInfos(); }

2
src/Storages/MergeTree/MergeList.cpp
View File

@ -52,7 +52,7 @@ MergeListElement::MergeListElement(const StorageID & table_id_, FutureMergedMuta
total_size_bytes_compressed += source_part->getBytesOnDisk();
total_size_bytes_uncompressed += source_part->getTotalColumnsSize().data_uncompressed;
total_size_marks += source_part->getMarksCount();
total_rows_count += source_part->index_granularity.getTotalRows();
total_rows_count += source_part->index_granularity->getTotalRows();
}
if (!future_part->parts.empty())

33
src/Storages/MergeTree/MergeTask.cpp
View File

@ -8,6 +8,7 @@
#include <Common/logger_useful.h>
#include <Core/Settings.h>
#include <Common/ProfileEvents.h>
#include <Storages/MergeTree/MergeTreeIndexGranularity.h>
#include <Compression/CompressedWriteBuffer.h>
#include <DataTypes/ObjectUtils.h>
#include <DataTypes/Serializations/SerializationInfo.h>
@ -67,6 +68,7 @@ namespace ProfileEvents
namespace DB
{
namespace Setting
{
extern const SettingsBool compile_sort_description;
@ -94,6 +96,7 @@ namespace MergeTreeSetting
extern const MergeTreeSettingsUInt64 vertical_merge_algorithm_min_rows_to_activate;
extern const MergeTreeSettingsBool vertical_merge_remote_filesystem_prefetch;
extern const MergeTreeSettingsBool prewarm_mark_cache;
extern const MergeTreeSettingsBool use_const_adaptive_granularity;
}
namespace ErrorCodes
@ -409,10 +412,11 @@ bool MergeTask::ExecuteAndFinalizeHorizontalPart::prepare() const
};
auto mutations_snapshot = global_ctx->data->getMutationsSnapshot(params);
const auto & storage_settings = *global_ctx->data->getSettings();
SerializationInfo::Settings info_settings =
{
.ratio_of_defaults_for_sparse = (*global_ctx->data->getSettings())[MergeTreeSetting::ratio_of_defaults_for_sparse_serialization],
.ratio_of_defaults_for_sparse = storage_settings[MergeTreeSetting::ratio_of_defaults_for_sparse_serialization],
.choose_kind = true,
};
@ -461,6 +465,7 @@ bool MergeTask::ExecuteAndFinalizeHorizontalPart::prepare() const
ctx->sum_input_rows_upper_bound = global_ctx->merge_list_element_ptr->total_rows_count;
ctx->sum_compressed_bytes_upper_bound = global_ctx->merge_list_element_ptr->total_size_bytes_compressed;
ctx->sum_uncompressed_bytes_upper_bound = global_ctx->merge_list_element_ptr->total_size_bytes_uncompressed;
global_ctx->chosen_merge_algorithm = chooseMergeAlgorithm();
global_ctx->merge_list_element_ptr->merge_algorithm.store(global_ctx->chosen_merge_algorithm, std::memory_order_relaxed);
@ -504,8 +509,14 @@ bool MergeTask::ExecuteAndFinalizeHorizontalPart::prepare() const
throw Exception(ErrorCodes::LOGICAL_ERROR, "Merge algorithm must be chosen");
}
/// If merge is vertical we cannot calculate it
ctx->blocks_are_granules_size = (global_ctx->chosen_merge_algorithm == MergeAlgorithm::Vertical);
bool use_adaptive_granularity = global_ctx->new_data_part->index_granularity_info.mark_type.adaptive;
bool use_const_adaptive_granularity = storage_settings[MergeTreeSetting::use_const_adaptive_granularity];
/// If merge is vertical we cannot calculate it.
/// If granularity is constant we don't need to calculate it.
ctx->blocks_are_granules_size = use_adaptive_granularity
&& !use_const_adaptive_granularity
&& global_ctx->chosen_merge_algorithm == MergeAlgorithm::Vertical;
/// Merged stream will be created and available as merged_stream variable
createMergedStream();
@ -547,7 +558,14 @@ bool MergeTask::ExecuteAndFinalizeHorizontalPart::prepare() const
}
}
bool save_marks_in_cache = (*global_ctx->data->getSettings())[MergeTreeSetting::prewarm_mark_cache] && global_ctx->context->getMarkCache();
auto index_granularity_ptr = createMergeTreeIndexGranularity(
ctx->sum_input_rows_upper_bound,
ctx->sum_uncompressed_bytes_upper_bound,
storage_settings,
global_ctx->new_data_part->index_granularity_info,
ctx->blocks_are_granules_size);
bool save_marks_in_cache = storage_settings[MergeTreeSetting::prewarm_mark_cache] && global_ctx->context->getMarkCache();
global_ctx->to = std::make_shared<MergedBlockOutputStream>(
global_ctx->new_data_part,
@ -556,6 +574,7 @@ bool MergeTask::ExecuteAndFinalizeHorizontalPart::prepare() const
MergeTreeIndexFactory::instance().getMany(global_ctx->merging_skip_indexes),
getStatisticsForColumns(global_ctx->merging_columns, global_ctx->metadata_snapshot),
ctx->compression_codec,
std::move(index_granularity_ptr),
global_ctx->txn ? global_ctx->txn->tid : Tx::PrehistoricTID,
/*reset_columns=*/ true,
save_marks_in_cache,
@ -1095,12 +1114,12 @@ void MergeTask::VerticalMergeStage::prepareVerticalMergeForOneColumn() const
global_ctx->new_data_part,
global_ctx->metadata_snapshot,
columns_list,
ctx->compression_codec,
column_pipepline.indexes_to_recalc,
getStatisticsForColumns(columns_list, global_ctx->metadata_snapshot),
ctx->compression_codec,
global_ctx->to->getIndexGranularity(),
&global_ctx->written_offset_columns,
save_marks_in_cache,
global_ctx->to->getIndexGranularity());
save_marks_in_cache);
ctx->column_elems_written = 0;
}

3
src/Storages/MergeTree/MergeTask.h
View File

@ -243,7 +243,6 @@ private:
bool need_remove_expired_values{false};
bool force_ttl{false};
CompressionCodecPtr compression_codec{nullptr};
size_t sum_input_rows_upper_bound{0};
std::shared_ptr<RowsSourcesTemporaryFile> rows_sources_temporary_file;
std::optional<ColumnSizeEstimator> column_sizes{};
@ -261,7 +260,9 @@ private:
std::function<bool()> is_cancelled{};
/// Local variables for this stage
size_t sum_input_rows_upper_bound{0};
size_t sum_compressed_bytes_upper_bound{0};
size_t sum_uncompressed_bytes_upper_bound{0};
bool blocks_are_granules_size{false};
LoggerPtr log{getLogger("MergeTask::PrepareStage")};

13
src/Storages/MergeTree/MergeTreeData.cpp
View File

@ -21,6 +21,7 @@
#include <Common/quoteString.h>
#include <Common/scope_guard_safe.h>
#include <Common/typeid_cast.h>
#include "Storages/MergeTree/MergeTreeIndexGranularityAdaptive.h"
#include <Core/Settings.h>
#include <Core/ServerSettings.h>
#include <Storages/MergeTree/RangesInDataPart.h>
@ -95,6 +96,7 @@
#include <atomic>
#include <chrono>
#include <limits>
#include <memory>
#include <optional>
#include <ranges>
#include <set>
@ -7237,7 +7239,7 @@ Block MergeTreeData::getMinMaxCountProjectionBlock(
/// It's extremely rare that some parts have final marks while others don't. To make it
/// straightforward, disable minmax_count projection when `max(pk)' encounters any part with
/// no final mark.
if (need_primary_key_max_column && !part->index_granularity.hasFinalMark())
if (need_primary_key_max_column && !part->index_granularity->hasFinalMark())
return {};
real_parts.push_back(part);
@ -8960,10 +8962,15 @@ std::pair<MergeTreeData::MutableDataPartPtr, scope_guard> MergeTreeData::createE
auto compression_codec = getContext()->chooseCompressionCodec(0, 0);
const auto & index_factory = MergeTreeIndexFactory::instance();
MergedBlockOutputStream out(new_data_part, metadata_snapshot, columns,
MergedBlockOutputStream out(
new_data_part,
metadata_snapshot,
columns,
index_factory.getMany(metadata_snapshot->getSecondaryIndices()),
ColumnsStatistics{},
compression_codec, txn ? txn->tid : Tx::PrehistoricTID);
compression_codec,
std::make_shared<MergeTreeIndexGranularityAdaptive>(),
txn ? txn->tid : Tx::PrehistoricTID);
bool sync_on_insert = (*settings)[MergeTreeSetting::fsync_after_insert];

31
src/Storages/MergeTree/MergeTreeDataPartCompact.cpp
View File

@ -3,6 +3,7 @@
#include <Storages/MergeTree/MergeTreeReaderCompactSingleBuffer.h>
#include <Storages/MergeTree/MergeTreeDataPartWriterCompact.h>
#include <Storages/MergeTree/LoadedMergeTreeDataPartInfoForReader.h>
#include <Storages/MergeTree/MergeTreeIndexGranularityAdaptive.h>
namespace DB
@ -15,6 +16,11 @@ namespace ErrorCodes
extern const int BAD_SIZE_OF_FILE_IN_DATA_PART;
}
namespace MergeTreeSetting
{
extern MergeTreeSettingsBool enable_index_granularity_compression;
}
MergeTreeDataPartCompact::MergeTreeDataPartCompact(
const MergeTreeData & storage_,
const String & name_,
@ -62,7 +68,7 @@ MergeTreeDataPartWriterPtr createMergeTreeDataPartCompactWriter(
const String & marks_file_extension_,
const CompressionCodecPtr & default_codec_,
const MergeTreeWriterSettings & writer_settings,
const MergeTreeIndexGranularity & computed_index_granularity)
MergeTreeIndexGranularityPtr computed_index_granularity)
{
NamesAndTypesList ordered_columns_list;
std::copy_if(columns_list.begin(), columns_list.end(), std::back_inserter(ordered_columns_list),
@ -76,7 +82,7 @@ MergeTreeDataPartWriterPtr createMergeTreeDataPartCompactWriter(
data_part_name_, logger_name_, serializations_, data_part_storage_,
index_granularity_info_, storage_settings_, ordered_columns_list, metadata_snapshot, virtual_columns,
indices_to_recalc, stats_to_recalc_, marks_file_extension_,
default_codec_, writer_settings, computed_index_granularity);
default_codec_, writer_settings, std::move(computed_index_granularity));
}
@ -95,8 +101,11 @@ void MergeTreeDataPartCompact::calculateEachColumnSizes(ColumnSizeByName & /*eac
}
void MergeTreeDataPartCompact::loadIndexGranularityImpl(
MergeTreeIndexGranularity & index_granularity_, const MergeTreeIndexGranularityInfo & index_granularity_info_,
size_t columns_count, const IDataPartStorage & data_part_storage_)
MergeTreeIndexGranularityPtr & index_granularity_ptr,
const MergeTreeIndexGranularityInfo & index_granularity_info_,
size_t columns_count,
const IDataPartStorage & data_part_storage_,
const MergeTreeSettings & storage_settings)
{
if (!index_granularity_info_.mark_type.adaptive)
throw Exception(ErrorCodes::NOT_IMPLEMENTED, "MergeTreeDataPartCompact cannot be created with non-adaptive granularity.");
@ -122,10 +131,14 @@ void MergeTreeDataPartCompact::loadIndexGranularityImpl(
marks_reader->ignore(columns_count * sizeof(MarkInCompressedFile));
size_t granularity;
readBinaryLittleEndian(granularity, *marks_reader);
index_granularity_.appendMark(granularity);
index_granularity_ptr->appendMark(granularity);
}
index_granularity_.setInitialized();
if (storage_settings[MergeTreeSetting::enable_index_granularity_compression])
{
if (auto new_granularity_ptr = index_granularity_ptr->optimize())
index_granularity_ptr = std::move(new_granularity_ptr);
}
}
void MergeTreeDataPartCompact::loadIndexGranularity()
@ -133,7 +146,7 @@ void MergeTreeDataPartCompact::loadIndexGranularity()
if (columns.empty())
throw Exception(ErrorCodes::NO_FILE_IN_DATA_PART, "No columns in part {}", name);
loadIndexGranularityImpl(index_granularity, index_granularity_info, columns.size(), getDataPartStorage());
loadIndexGranularityImpl(index_granularity, index_granularity_info, columns.size(), getDataPartStorage(), *storage.getSettings());
}
void MergeTreeDataPartCompact::loadMarksToCache(const Names & column_names, MarkCache * mark_cache) const
@ -152,7 +165,7 @@ void MergeTreeDataPartCompact::loadMarksToCache(const Names & column_names, Mark
info_for_read,
mark_cache,
index_granularity_info.getMarksFilePath(DATA_FILE_NAME),
index_granularity.getMarksCount(),
index_granularity->getMarksCount(),
index_granularity_info,
/*save_marks_in_cache=*/ true,
read_settings,
@ -227,7 +240,7 @@ void MergeTreeDataPartCompact::doCheckConsistency(bool require_part_metadata) co
getDataPartStorage().getRelativePath(),
std::string(fs::path(getDataPartStorage().getFullPath()) / mrk_file_name));
UInt64 expected_file_size = index_granularity_info.getMarkSizeInBytes(columns.size()) * index_granularity.getMarksCount();
UInt64 expected_file_size = index_granularity_info.getMarkSizeInBytes(columns.size()) * index_granularity->getMarksCount();
if (expected_file_size != file_size)
throw Exception(
ErrorCodes::BAD_SIZE_OF_FILE_IN_DATA_PART,

7
src/Storages/MergeTree/MergeTreeDataPartCompact.h
View File

@ -60,8 +60,11 @@ public:
protected:
static void loadIndexGranularityImpl(
MergeTreeIndexGranularity & index_granularity_, const MergeTreeIndexGranularityInfo & index_granularity_info_,
size_t columns_count, const IDataPartStorage & data_part_storage_);
MergeTreeIndexGranularityPtr & index_granularity_,
const MergeTreeIndexGranularityInfo & index_granularity_info_,
size_t columns_count,
const IDataPartStorage & data_part_storage_,
const MergeTreeSettings & storage_settings);
void doCheckConsistency(bool require_part_metadata) const override;

38
src/Storages/MergeTree/MergeTreeDataPartWide.cpp
View File

@ -5,6 +5,7 @@
#include <Storages/MergeTree/LoadedMergeTreeDataPartInfoForReader.h>
#include <DataTypes/NestedUtils.h>
#include <Core/NamesAndTypes.h>
#include <Storages/MergeTree/MergeTreeIndexGranularityConstant.h>
namespace DB
@ -17,6 +18,11 @@ namespace ErrorCodes
extern const int LOGICAL_ERROR;
}
namespace MergeTreeSetting
{
extern MergeTreeSettingsBool enable_index_granularity_compression;
}
MergeTreeDataPartWide::MergeTreeDataPartWide(
const MergeTreeData & storage_,
const String & name_,
@ -68,14 +74,14 @@ MergeTreeDataPartWriterPtr createMergeTreeDataPartWideWriter(
const String & marks_file_extension_,
const CompressionCodecPtr & default_codec_,
const MergeTreeWriterSettings & writer_settings,
const MergeTreeIndexGranularity & computed_index_granularity)
MergeTreeIndexGranularityPtr computed_index_granularity)
{
return std::make_unique<MergeTreeDataPartWriterWide>(
data_part_name_, logger_name_, serializations_, data_part_storage_,
index_granularity_info_, storage_settings_, columns_list,
metadata_snapshot, virtual_columns, indices_to_recalc, stats_to_recalc_,
marks_file_extension_,
default_codec_, writer_settings, computed_index_granularity);
default_codec_, writer_settings, std::move(computed_index_granularity));
}
@ -114,8 +120,11 @@ ColumnSize MergeTreeDataPartWide::getColumnSizeImpl(
}
void MergeTreeDataPartWide::loadIndexGranularityImpl(
MergeTreeIndexGranularity & index_granularity_, MergeTreeIndexGranularityInfo & index_granularity_info_,
const IDataPartStorage & data_part_storage_, const std::string & any_column_file_name)
MergeTreeIndexGranularityPtr & index_granularity_ptr,
MergeTreeIndexGranularityInfo & index_granularity_info_,
const IDataPartStorage & data_part_storage_,
const std::string & any_column_file_name,
const MergeTreeSettings & storage_settings)
{
index_granularity_info_.changeGranularityIfRequired(data_part_storage_);
@ -127,12 +136,13 @@ void MergeTreeDataPartWide::loadIndexGranularityImpl(
std::string(fs::path(data_part_storage_.getFullPath()) / marks_file_path));
size_t marks_file_size = data_part_storage_.getFileSize(marks_file_path);
size_t fixed_granularity = index_granularity_info_.fixed_index_granularity;
if (!index_granularity_info_.mark_type.adaptive && !index_granularity_info_.mark_type.compressed)
{
/// The most easy way - no need to read the file, everything is known from its size.
size_t marks_count = marks_file_size / index_granularity_info_.getMarkSizeInBytes();
index_granularity_.resizeWithFixedGranularity(marks_count, index_granularity_info_.fixed_index_granularity); /// all the same
index_granularity_ptr = std::make_shared<MergeTreeIndexGranularityConstant>(fixed_granularity, fixed_granularity, marks_count, false);
}
else
{
@ -145,6 +155,7 @@ void MergeTreeDataPartWide::loadIndexGranularityImpl(
marks_reader = std::make_unique<CompressedReadBufferFromFile>(std::move(marks_file));
size_t marks_count = 0;
while (!marks_reader->eof())
{
MarkInCompressedFile mark;
@ -157,15 +168,20 @@ void MergeTreeDataPartWide::loadIndexGranularityImpl(
if (index_granularity_info_.mark_type.adaptive)
{
readBinaryLittleEndian(granularity, *marks_reader);
index_granularity_.appendMark(granularity);
index_granularity_ptr->appendMark(granularity);
}
}
if (!index_granularity_info_.mark_type.adaptive)
index_granularity_.resizeWithFixedGranularity(marks_count, index_granularity_info_.fixed_index_granularity); /// all the same
{
index_granularity_ptr = std::make_shared<MergeTreeIndexGranularityConstant>(fixed_granularity, fixed_granularity, marks_count, false);
}
else if (storage_settings[MergeTreeSetting::enable_index_granularity_compression])
{
if (auto new_granularity_ptr = index_granularity_ptr->optimize())
index_granularity_ptr = std::move(new_granularity_ptr);
}
}
index_granularity_.setInitialized();
}
void MergeTreeDataPartWide::loadIndexGranularity()
@ -179,7 +195,7 @@ void MergeTreeDataPartWide::loadIndexGranularity()
"There are no files for column {} in part {}",
columns.front().name, getDataPartStorage().getFullPath());
loadIndexGranularityImpl(index_granularity, index_granularity_info, getDataPartStorage(), *any_column_filename);
loadIndexGranularityImpl(index_granularity, index_granularity_info, getDataPartStorage(), *any_column_filename, *storage.getSettings());
}
void MergeTreeDataPartWide::loadMarksToCache(const Names & column_names, MarkCache * mark_cache) const
@ -209,7 +225,7 @@ void MergeTreeDataPartWide::loadMarksToCache(const Names & column_names, MarkCac
info_for_read,
mark_cache,
index_granularity_info.getMarksFilePath(*stream_name),
index_granularity.getMarksCount(),
index_granularity->getMarksCount(),
index_granularity_info,
/*save_marks_in_cache=*/ true,
read_settings,

7
src/Storages/MergeTree/MergeTreeDataPartWide.h
View File

@ -55,8 +55,11 @@ public:
protected:
static void loadIndexGranularityImpl(
MergeTreeIndexGranularity & index_granularity_, MergeTreeIndexGranularityInfo & index_granularity_info_,
const IDataPartStorage & data_part_storage_, const std::string & any_column_file_name);
MergeTreeIndexGranularityPtr & index_granularity_ptr,
MergeTreeIndexGranularityInfo & index_granularity_info_,
const IDataPartStorage & data_part_storage_,
const std::string & any_column_file_name,
const MergeTreeSettings & storage_settings);
void doCheckConsistency(bool require_part_metadata) const override;

19
src/Storages/MergeTree/MergeTreeDataPartWriterCompact.cpp
View File

@ -25,13 +25,13 @@ MergeTreeDataPartWriterCompact::MergeTreeDataPartWriterCompact(
const String & marks_file_extension_,
const CompressionCodecPtr & default_codec_,
const MergeTreeWriterSettings & settings_,
const MergeTreeIndexGranularity & index_granularity_)
MergeTreeIndexGranularityPtr index_granularity_)
: MergeTreeDataPartWriterOnDisk(
data_part_name_, logger_name_, serializations_,
data_part_storage_, index_granularity_info_, storage_settings_,
columns_list_, metadata_snapshot_, virtual_columns_,
indices_to_recalc_, stats_to_recalc, marks_file_extension_,
default_codec_, settings_, index_granularity_)
default_codec_, settings_, std::move(index_granularity_))
, plain_file(getDataPartStorage().writeFile(
MergeTreeDataPartCompact::DATA_FILE_NAME_WITH_EXTENSION,
settings.max_compress_block_size,
@ -200,13 +200,13 @@ void MergeTreeDataPartWriterCompact::write(const Block & block, const IColumn::P
header = result_block.cloneEmpty();
columns_buffer.add(result_block.mutateColumns());
size_t current_mark_rows = index_granularity.getMarkRows(getCurrentMark());
size_t current_mark_rows = index_granularity->getMarkRows(getCurrentMark());
size_t rows_in_buffer = columns_buffer.size();
if (rows_in_buffer >= current_mark_rows)
{
Block flushed_block = header.cloneWithColumns(columns_buffer.releaseColumns());
auto granules_to_write = getGranulesToWrite(index_granularity, flushed_block.rows(), getCurrentMark(), /* last_block = */ false);
auto granules_to_write = getGranulesToWrite(*index_granularity, flushed_block.rows(), getCurrentMark(), /* last_block = */ false);
writeDataBlockPrimaryIndexAndSkipIndices(flushed_block, granules_to_write);
setCurrentMark(getCurrentMark() + granules_to_write.size());
calculateAndSerializeStatistics(flushed_block);
@ -285,12 +285,11 @@ void MergeTreeDataPartWriterCompact::fillDataChecksums(MergeTreeDataPartChecksum
if (columns_buffer.size() != 0)
{
auto block = header.cloneWithColumns(columns_buffer.releaseColumns());
auto granules_to_write = getGranulesToWrite(index_granularity, block.rows(), getCurrentMark(), /* last_block = */ true);
auto granules_to_write = getGranulesToWrite(*index_granularity, block.rows(), getCurrentMark(), /*last_block=*/ true);
if (!granules_to_write.back().is_complete)
{
/// Correct last mark as it should contain exact amount of rows.
index_granularity.popMark();
index_granularity.appendMark(granules_to_write.back().rows_to_write);
index_granularity->adjustLastMark(granules_to_write.back().rows_to_write);
}
writeDataBlockPrimaryIndexAndSkipIndices(block, granules_to_write);
}
@ -386,11 +385,11 @@ static void fillIndexGranularityImpl(
void MergeTreeDataPartWriterCompact::fillIndexGranularity(size_t index_granularity_for_block, size_t rows_in_block)
{
size_t index_offset = 0;
if (index_granularity.getMarksCount() > getCurrentMark())
index_offset = index_granularity.getMarkRows(getCurrentMark()) - columns_buffer.size();
if (index_granularity->getMarksCount() > getCurrentMark())
index_offset = index_granularity->getMarkRows(getCurrentMark()) - columns_buffer.size();
fillIndexGranularityImpl(
index_granularity,
*index_granularity,
index_offset,
index_granularity_for_block,
rows_in_block);

2
src/Storages/MergeTree/MergeTreeDataPartWriterCompact.h
View File

@ -25,7 +25,7 @@ public:
const String & marks_file_extension,
const CompressionCodecPtr & default_codec,
const MergeTreeWriterSettings & settings,
const MergeTreeIndexGranularity & index_granularity);
MergeTreeIndexGranularityPtr index_granularity_);
void write(const Block & block, const IColumn::Permutation * permutation) override;

72
src/Storages/MergeTree/MergeTreeDataPartWriterOnDisk.cpp
View File

@ -162,20 +162,20 @@ MergeTreeDataPartWriterOnDisk::MergeTreeDataPartWriterOnDisk(
const String & marks_file_extension_,
const CompressionCodecPtr & default_codec_,
const MergeTreeWriterSettings & settings_,
const MergeTreeIndexGranularity & index_granularity_)
MergeTreeIndexGranularityPtr index_granularity_)
: IMergeTreeDataPartWriter(
data_part_name_, serializations_, data_part_storage_, index_granularity_info_,
storage_settings_, columns_list_, metadata_snapshot_, virtual_columns_, settings_, index_granularity_)
storage_settings_, columns_list_, metadata_snapshot_, virtual_columns_, settings_, std::move(index_granularity_))
, skip_indices(indices_to_recalc_)
, stats(stats_to_recalc_)
, marks_file_extension(marks_file_extension_)
, default_codec(default_codec_)
, compute_granularity(index_granularity.empty())
, compute_granularity(index_granularity->empty())
, compress_primary_key(settings.compress_primary_key)
, execution_stats(skip_indices.size(), stats.size())
, log(getLogger(logger_name_ + " (DataPartWriter)"))
{
if (settings.blocks_are_granules_size && !index_granularity.empty())
if (settings.blocks_are_granules_size && !index_granularity->empty())
throw Exception(ErrorCodes::LOGICAL_ERROR,
"Can't take information about index granularity from blocks, when non empty index_granularity array specified");
@ -189,63 +189,15 @@ MergeTreeDataPartWriterOnDisk::MergeTreeDataPartWriterOnDisk(
initStatistics();
}
// Implementation is split into static functions for ability
/// of making unit tests without creation instance of IMergeTreeDataPartWriter,
/// which requires a lot of dependencies and access to filesystem.
static size_t computeIndexGranularityImpl(
const Block & block,
size_t index_granularity_bytes,
size_t fixed_index_granularity_rows,
bool blocks_are_granules,
bool can_use_adaptive_index_granularity)
{
size_t rows_in_block = block.rows();
size_t index_granularity_for_block;
if (!can_use_adaptive_index_granularity)
{
index_granularity_for_block = fixed_index_granularity_rows;
}
else
{
size_t block_size_in_memory = block.bytes();
if (blocks_are_granules)
{
index_granularity_for_block = rows_in_block;
}
else if (block_size_in_memory >= index_granularity_bytes)
{
size_t granules_in_block = block_size_in_memory / index_granularity_bytes;
index_granularity_for_block = rows_in_block / granules_in_block;
}
else
{
size_t size_of_row_in_bytes = std::max(block_size_in_memory / rows_in_block, 1UL);
index_granularity_for_block = index_granularity_bytes / size_of_row_in_bytes;
}
}
/// We should be less or equal than fixed index granularity.
/// But if block size is a granule size then do not adjust it.
/// Granularity greater than fixed granularity might come from compact part.
if (!blocks_are_granules)
index_granularity_for_block = std::min(fixed_index_granularity_rows, index_granularity_for_block);
/// Very rare case when index granularity bytes less than single row.
if (index_granularity_for_block == 0)
index_granularity_for_block = 1;
return index_granularity_for_block;
}
size_t MergeTreeDataPartWriterOnDisk::computeIndexGranularity(const Block & block) const
{
return computeIndexGranularityImpl(
block,
(*storage_settings)[MergeTreeSetting::index_granularity_bytes],
(*storage_settings)[MergeTreeSetting::index_granularity],
settings.blocks_are_granules_size,
settings.can_use_adaptive_granularity);
return computeIndexGranularityForBlock(
block.rows(),
block.bytes(),
(*storage_settings)[MergeTreeSetting::index_granularity_bytes],
(*storage_settings)[MergeTreeSetting::index_granularity],
settings.blocks_are_granules_size,
settings.can_use_adaptive_granularity);
}
void MergeTreeDataPartWriterOnDisk::initPrimaryIndex()
@ -433,7 +385,7 @@ void MergeTreeDataPartWriterOnDisk::fillPrimaryIndexChecksums(MergeTreeData::Dat
{
bool write_final_mark = (with_final_mark && data_written);
if (write_final_mark && compute_granularity)
index_granularity.appendMark(0);
index_granularity->appendMark(0);
if (index_file_hashing_stream)
{

2
src/Storages/MergeTree/MergeTreeDataPartWriterOnDisk.h
View File

@ -116,7 +116,7 @@ public:
const String & marks_file_extension,
const CompressionCodecPtr & default_codec,
const MergeTreeWriterSettings & settings,
const MergeTreeIndexGranularity & index_granularity);
MergeTreeIndexGranularityPtr index_granularity_);
void setWrittenOffsetColumns(WrittenOffsetColumns * written_offset_columns_)
{

59
src/Storages/MergeTree/MergeTreeDataPartWriterWide.cpp
View File

@ -99,13 +99,13 @@ MergeTreeDataPartWriterWide::MergeTreeDataPartWriterWide(
const String & marks_file_extension_,
const CompressionCodecPtr & default_codec_,
const MergeTreeWriterSettings & settings_,
const MergeTreeIndexGranularity & index_granularity_)
MergeTreeIndexGranularityPtr index_granularity_)
: MergeTreeDataPartWriterOnDisk(
data_part_name_, logger_name_, serializations_,
data_part_storage_, index_granularity_info_, storage_settings_,
columns_list_, metadata_snapshot_, virtual_columns_,
indices_to_recalc_, stats_to_recalc_, marks_file_extension_,
default_codec_, settings_, index_granularity_)
default_codec_, settings_, std::move(index_granularity_))
{
if (settings.save_marks_in_cache)
{
@ -255,8 +255,8 @@ void MergeTreeDataPartWriterWide::shiftCurrentMark(const Granules & granules_wri
if (settings.can_use_adaptive_granularity && settings.blocks_are_granules_size)
throw Exception(ErrorCodes::LOGICAL_ERROR, "Incomplete granules are not allowed while blocks are granules size. "
"Mark number {} (rows {}), rows written in last mark {}, rows to write in last mark from block {} (from row {}), "
"total marks currently {}", last_granule.mark_number, index_granularity.getMarkRows(last_granule.mark_number),
rows_written_in_last_mark, last_granule.rows_to_write, last_granule.start_row, index_granularity.getMarksCount());
"total marks currently {}", last_granule.mark_number, index_granularity->getMarkRows(last_granule.mark_number),
rows_written_in_last_mark, last_granule.rows_to_write, last_granule.start_row, index_granularity->getMarksCount());
/// Shift forward except last granule
setCurrentMark(getCurrentMark() + granules_written.size() - 1);
@ -288,7 +288,7 @@ void MergeTreeDataPartWriterWide::write(const Block & block, const IColumn::Perm
size_t index_granularity_for_block = computeIndexGranularity(block);
if (rows_written_in_last_mark > 0)
{
size_t rows_left_in_last_mark = index_granularity.getMarkRows(getCurrentMark()) - rows_written_in_last_mark;
size_t rows_left_in_last_mark = index_granularity->getMarkRows(getCurrentMark()) - rows_written_in_last_mark;
/// Previous granularity was much bigger than our new block's
/// granularity let's adjust it, because we want add new
/// heavy-weight blocks into small old granule.
@ -308,7 +308,7 @@ void MergeTreeDataPartWriterWide::write(const Block & block, const IColumn::Perm
Block block_to_write = block;
auto granules_to_write = getGranulesToWrite(index_granularity, block_to_write.rows(), getCurrentMark(), rows_written_in_last_mark);
auto granules_to_write = getGranulesToWrite(*index_granularity, block_to_write.rows(), getCurrentMark(), rows_written_in_last_mark);
auto offset_columns = written_offset_columns ? *written_offset_columns : WrittenOffsetColumns{};
Block primary_key_block;
@ -496,7 +496,7 @@ void MergeTreeDataPartWriterWide::writeColumn(
throw Exception(ErrorCodes::LOGICAL_ERROR,
"We have to add new mark for column, but already have non written mark. "
"Current mark {}, total marks {}, offset {}",
getCurrentMark(), index_granularity.getMarksCount(), rows_written_in_last_mark);
getCurrentMark(), index_granularity->getMarksCount(), rows_written_in_last_mark);
last_non_written_marks[name] = getCurrentMarksForColumn(name_and_type, column.getPtr(), offset_columns);
}
@ -516,7 +516,7 @@ void MergeTreeDataPartWriterWide::writeColumn(
throw Exception(ErrorCodes::LOGICAL_ERROR, "No mark was saved for incomplete granule for column {}", backQuoteIfNeed(name));
for (const auto & mark : marks_it->second)
flushMarkToFile(mark, index_granularity.getMarkRows(granule.mark_number));
flushMarkToFile(mark, index_granularity->getMarkRows(granule.mark_number));
last_non_written_marks.erase(marks_it);
}
}
@ -563,10 +563,10 @@ void MergeTreeDataPartWriterWide::validateColumnOfFixedSize(const NameAndTypePai
for (mark_num = 0; !mrk_in->eof(); ++mark_num)
{
if (mark_num > index_granularity.getMarksCount())
if (mark_num > index_granularity->getMarksCount())
throw Exception(ErrorCodes::LOGICAL_ERROR,
"Incorrect number of marks in memory {}, on disk (at least) {}",
index_granularity.getMarksCount(), mark_num + 1);
index_granularity->getMarksCount(), mark_num + 1);
readBinaryLittleEndian(offset_in_compressed_file, *mrk_in);
readBinaryLittleEndian(offset_in_decompressed_block, *mrk_in);
@ -597,10 +597,10 @@ void MergeTreeDataPartWriterWide::validateColumnOfFixedSize(const NameAndTypePai
throw Exception(ErrorCodes::LOGICAL_ERROR,
"Still have {} rows in bin stream, last mark #{}"
" index granularity size {}, last rows {}",
column->size(), mark_num, index_granularity.getMarksCount(), index_granularity_rows);
column->size(), mark_num, index_granularity->getMarksCount(), index_granularity_rows);
}
if (index_granularity_rows != index_granularity.getMarkRows(mark_num))
if (index_granularity_rows != index_granularity->getMarkRows(mark_num))
{
throw Exception(
ErrorCodes::LOGICAL_ERROR,
@ -608,8 +608,8 @@ void MergeTreeDataPartWriterWide::validateColumnOfFixedSize(const NameAndTypePai
" (compressed offset {}, decompressed offset {}), in-memory {}, on disk {}, total marks {}",
getDataPartStorage().getFullPath(),
mark_num, offset_in_compressed_file, offset_in_decompressed_block,
index_granularity.getMarkRows(mark_num), index_granularity_rows,
index_granularity.getMarksCount());
index_granularity->getMarkRows(mark_num), index_granularity_rows,
index_granularity->getMarksCount());
}
auto column = type->createColumn();
@ -644,7 +644,7 @@ void MergeTreeDataPartWriterWide::validateColumnOfFixedSize(const NameAndTypePai
ErrorCodes::LOGICAL_ERROR, "Incorrect mark rows for mark #{} (compressed offset {}, decompressed offset {}), "
"actually in bin file {}, in mrk file {}, total marks {}",
mark_num, offset_in_compressed_file, offset_in_decompressed_block, column->size(),
index_granularity.getMarkRows(mark_num), index_granularity.getMarksCount());
index_granularity->getMarkRows(mark_num), index_granularity->getMarksCount());
}
}
@ -652,7 +652,7 @@ void MergeTreeDataPartWriterWide::validateColumnOfFixedSize(const NameAndTypePai
throw Exception(ErrorCodes::LOGICAL_ERROR,
"Still have something in marks stream, last mark #{}"
" index granularity size {}, last rows {}",
mark_num, index_granularity.getMarksCount(), index_granularity_rows);
mark_num, index_granularity->getMarksCount(), index_granularity_rows);
if (!bin_in.eof())
{
auto column = type->createColumn();
@ -662,7 +662,7 @@ void MergeTreeDataPartWriterWide::validateColumnOfFixedSize(const NameAndTypePai
throw Exception(ErrorCodes::LOGICAL_ERROR,
"Still have {} rows in bin stream, last mark #{}"
" index granularity size {}, last rows {}",
column->size(), mark_num, index_granularity.getMarksCount(), index_granularity_rows);
column->size(), mark_num, index_granularity->getMarksCount(), index_granularity_rows);
}
}
@ -679,8 +679,8 @@ void MergeTreeDataPartWriterWide::fillDataChecksums(MergeTreeDataPartChecksums &
throw Exception(ErrorCodes::LOGICAL_ERROR,
"Incomplete granule is not allowed while blocks are granules size even for last granule. "
"Mark number {} (rows {}), rows written for last mark {}, total marks {}",
getCurrentMark(), index_granularity.getMarkRows(getCurrentMark()),
rows_written_in_last_mark, index_granularity.getMarksCount());
getCurrentMark(), index_granularity->getMarkRows(getCurrentMark()),
rows_written_in_last_mark, index_granularity->getMarksCount());
adjustLastMarkIfNeedAndFlushToDisk(rows_written_in_last_mark);
}
@ -799,16 +799,16 @@ static void fillIndexGranularityImpl(
void MergeTreeDataPartWriterWide::fillIndexGranularity(size_t index_granularity_for_block, size_t rows_in_block)
{
if (getCurrentMark() < index_granularity.getMarksCount() && getCurrentMark() != index_granularity.getMarksCount() - 1)
if (getCurrentMark() < index_granularity->getMarksCount() && getCurrentMark() != index_granularity->getMarksCount() - 1)
throw Exception(ErrorCodes::LOGICAL_ERROR, "Trying to add marks, while current mark {}, but total marks {}",
getCurrentMark(), index_granularity.getMarksCount());
getCurrentMark(), index_granularity->getMarksCount());
size_t index_offset = 0;
if (rows_written_in_last_mark != 0)
index_offset = index_granularity.getLastMarkRows() - rows_written_in_last_mark;
index_offset = index_granularity->getLastMarkRows() - rows_written_in_last_mark;
fillIndexGranularityImpl(
index_granularity,
*index_granularity,
index_offset,
index_granularity_for_block,
rows_in_block);
@ -827,27 +827,26 @@ void MergeTreeDataPartWriterWide::adjustLastMarkIfNeedAndFlushToDisk(size_t new_
/// other columns
if (compute_granularity && settings.can_use_adaptive_granularity)
{
if (getCurrentMark() != index_granularity.getMarksCount() - 1)
if (getCurrentMark() != index_granularity->getMarksCount() - 1)
throw Exception(ErrorCodes::LOGICAL_ERROR,
"Non last mark {} (with {} rows) having rows offset {}, total marks {}",
getCurrentMark(), index_granularity.getMarkRows(getCurrentMark()),
rows_written_in_last_mark, index_granularity.getMarksCount());
getCurrentMark(), index_granularity->getMarkRows(getCurrentMark()),
rows_written_in_last_mark, index_granularity->getMarksCount());
index_granularity.popMark();
index_granularity.appendMark(new_rows_in_last_mark);
index_granularity->adjustLastMark(new_rows_in_last_mark);
}
/// Last mark should be filled, otherwise it's a bug
if (last_non_written_marks.empty())
throw Exception(ErrorCodes::LOGICAL_ERROR, "No saved marks for last mark {} having rows offset {}, total marks {}",
getCurrentMark(), rows_written_in_last_mark, index_granularity.getMarksCount());
getCurrentMark(), rows_written_in_last_mark, index_granularity->getMarksCount());
if (rows_written_in_last_mark == new_rows_in_last_mark)
{
for (const auto & [name, marks] : last_non_written_marks)
{
for (const auto & mark : marks)
flushMarkToFile(mark, index_granularity.getMarkRows(getCurrentMark()));
flushMarkToFile(mark, index_granularity->getMarkRows(getCurrentMark()));
}
last_non_written_marks.clear();

2
src/Storages/MergeTree/MergeTreeDataPartWriterWide.h
View File

@ -35,7 +35,7 @@ public:
const String & marks_file_extension,
const CompressionCodecPtr & default_codec,
const MergeTreeWriterSettings & settings,
const MergeTreeIndexGranularity & index_granularity);
MergeTreeIndexGranularityPtr index_granularity_);
void write(const Block & block, const IColumn::Permutation * permutation) override;

37
src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp
View File

@ -129,7 +129,7 @@ size_t MergeTreeDataSelectExecutor::getApproximateTotalRowsToRead(
{
MarkRanges part_ranges = markRangesFromPKRange(part, metadata_snapshot, key_condition, {}, &exact_ranges, settings, log);
for (const auto & range : part_ranges)
rows_count += part->index_granularity.getRowsCountInRange(range);
rows_count += part->index_granularity->getRowsCountInRange(range);
}
UNUSED(exact_ranges);
@ -688,7 +688,7 @@ RangesInDataParts MergeTreeDataSelectExecutor::filterPartsByPrimaryKeyAndSkipInd
auto & part = parts[part_index];
RangesInDataPart ranges(part, part_index);
size_t total_marks_count = part->index_granularity.getMarksCountWithoutFinal();
size_t total_marks_count = part->index_granularity->getMarksCountWithoutFinal();
if (metadata_snapshot->hasPrimaryKey() || part_offset_condition)
{
@ -1044,12 +1044,12 @@ MarkRanges MergeTreeDataSelectExecutor::markRangesFromPKRange(
{
MarkRanges res;
size_t marks_count = part->index_granularity.getMarksCount();
size_t marks_count = part->index_granularity->getMarksCount();
const auto & index = part->getIndex();
if (marks_count == 0)
return res;
bool has_final_mark = part->index_granularity.hasFinalMark();
bool has_final_mark = part->index_granularity->hasFinalMark();
bool key_condition_useful = !key_condition.alwaysUnknownOrTrue();
bool part_offset_condition_useful = part_offset_condition && !part_offset_condition->alwaysUnknownOrTrue();
@ -1156,16 +1156,16 @@ MarkRanges MergeTreeDataSelectExecutor::markRangesFromPKRange(
auto check_part_offset_condition = [&]()
{
auto begin = part->index_granularity.getMarkStartingRow(range.begin);
auto end = part->index_granularity.getMarkStartingRow(range.end) - 1;
auto begin = part->index_granularity->getMarkStartingRow(range.begin);
auto end = part->index_granularity->getMarkStartingRow(range.end) - 1;
if (begin > end)
{
/// Empty mark (final mark)
return BoolMask(false, true);
}
part_offset_left[0] = part->index_granularity.getMarkStartingRow(range.begin);
part_offset_right[0] = part->index_granularity.getMarkStartingRow(range.end) - 1;
part_offset_left[0] = part->index_granularity->getMarkStartingRow(range.begin);
part_offset_right[0] = part->index_granularity->getMarkStartingRow(range.end) - 1;
part_offset_left[1] = part->name;
part_offset_right[1] = part->name;
@ -1377,9 +1377,8 @@ MarkRanges MergeTreeDataSelectExecutor::filterMarksUsingIndex(
part->index_granularity_info.fixed_index_granularity,
part->index_granularity_info.index_granularity_bytes);
size_t marks_count = part->getMarksCount();
size_t final_mark = part->index_granularity.hasFinalMark();
size_t index_marks_count = (marks_count - final_mark + index_granularity - 1) / index_granularity;
size_t marks_count = part->index_granularity->getMarksCountWithoutFinal();
size_t index_marks_count = (marks_count + index_granularity - 1) / index_granularity;
MarkRanges index_ranges;
for (const auto & range : ranges)
@ -1427,8 +1426,7 @@ MarkRanges MergeTreeDataSelectExecutor::filterMarksUsingIndex(
for (auto row : rows)
{
const MergeTreeIndexGranularity & merge_tree_index_granularity = part->index_granularity;
size_t num_marks = merge_tree_index_granularity.countMarksForRows(index_mark * index_granularity, row);
size_t num_marks = part->index_granularity->countMarksForRows(index_mark * index_granularity, row);
MarkRange data_range(
std::max(ranges[i].begin, (index_mark * index_granularity) + num_marks),
@ -1501,9 +1499,8 @@ MarkRanges MergeTreeDataSelectExecutor::filterMarksUsingMergedIndex(
part->index_granularity_info.fixed_index_granularity,
part->index_granularity_info.index_granularity_bytes);
size_t marks_count = part->getMarksCount();
size_t final_mark = part->index_granularity.hasFinalMark();
size_t index_marks_count = (marks_count - final_mark + index_granularity - 1) / index_granularity;
size_t marks_count = part->index_granularity->getMarksCountWithoutFinal();
size_t index_marks_count = (marks_count + index_granularity - 1) / index_granularity;
std::vector<std::unique_ptr<MergeTreeIndexReader>> readers;
for (const auto & index_helper : indices)
@ -1603,9 +1600,7 @@ void MergeTreeDataSelectExecutor::selectPartsToRead(
continue;
}
size_t num_granules = part->getMarksCount();
if (num_granules && part->index_granularity.hasFinalMark())
--num_granules;
size_t num_granules = part->index_granularity->getMarksCountWithoutFinal();
counters.num_initial_selected_parts += 1;
counters.num_initial_selected_granules += num_granules;
@ -1672,9 +1667,7 @@ void MergeTreeDataSelectExecutor::selectPartsToReadWithUUIDFilter(
if (part->uuid != UUIDHelpers::Nil && ignored_part_uuids->has(part->uuid))
continue;
size_t num_granules = part->getMarksCount();
if (num_granules && part->index_granularity.hasFinalMark())
--num_granules;
size_t num_granules = part->index_granularity->getMarksCountWithoutFinal();
counters.num_initial_selected_parts += 1;
counters.num_initial_selected_granules += num_granules;

16
src/Storages/MergeTree/MergeTreeDataWriter.cpp
View File

@ -687,6 +687,13 @@ MergeTreeDataWriter::TemporaryPart MergeTreeDataWriter::writeTempPartImpl(
auto compression_codec = data.getContext()->chooseCompressionCodec(0, 0);
bool save_marks_in_cache = (*data_settings)[MergeTreeSetting::prewarm_mark_cache] && data.getContext()->getMarkCache();
auto index_granularity_ptr = createMergeTreeIndexGranularity(
block.rows(),
block.bytes(),
*data.getSettings(),
new_data_part->index_granularity_info,
/*blocks_are_granules=*/ false);
auto out = std::make_unique<MergedBlockOutputStream>(
new_data_part,
metadata_snapshot,
@ -694,6 +701,7 @@ MergeTreeDataWriter::TemporaryPart MergeTreeDataWriter::writeTempPartImpl(
indices,
statistics,
compression_codec,
std::move(index_granularity_ptr),
context->getCurrentTransaction() ? context->getCurrentTransaction()->tid : Tx::PrehistoricTID,
/*reset_columns=*/ false,
save_marks_in_cache,
@ -834,6 +842,13 @@ MergeTreeDataWriter::TemporaryPart MergeTreeDataWriter::writeProjectionPartImpl(
auto compression_codec = data.getContext()->chooseCompressionCodec(0, 0);
bool save_marks_in_cache = (*data.getSettings())[MergeTreeSetting::prewarm_mark_cache] && data.getContext()->getMarkCache();
auto index_granularity_ptr = createMergeTreeIndexGranularity(
block.rows(),
block.bytes(),
*data.getSettings(),
new_data_part->index_granularity_info,
/*blocks_are_granules=*/ false);
auto out = std::make_unique<MergedBlockOutputStream>(
new_data_part,
metadata_snapshot,
@ -842,6 +857,7 @@ MergeTreeDataWriter::TemporaryPart MergeTreeDataWriter::writeProjectionPartImpl(
/// TODO(hanfei): It should be helpful to write statistics for projection result.
ColumnsStatistics{},
compression_codec,
std::move(index_granularity_ptr),
Tx::PrehistoricTID,
/*reset_columns=*/ false,
save_marks_in_cache,

188
src/Storages/MergeTree/MergeTreeIndexGranularity.cpp
View File

@ -1,77 +1,25 @@
#include <memory>
#include <Storages/MergeTree/MergeTreeIndexGranularity.h>
#include <Common/Exception.h>
#include <Storages/MergeTree/MergeTreeIndexGranularityAdaptive.h>
#include <Storages/MergeTree/MergeTreeIndexGranularityConstant.h>
#include <Storages/MergeTree/MergeTreeIndexGranularityInfo.h>
#include <IO/WriteHelpers.h>
#include "Common/Exception.h"
#include "Storages/MergeTree/MergeTreeDataPartType.h"
namespace DB
{
namespace ErrorCodes
{
extern const int LOGICAL_ERROR;
}
MergeTreeIndexGranularity::MergeTreeIndexGranularity(const std::vector<size_t> & marks_rows_partial_sums_)
: marks_rows_partial_sums(marks_rows_partial_sums_)
namespace MergeTreeSetting
{
}
/// Rows after mark to next mark
size_t MergeTreeIndexGranularity::getMarkRows(size_t mark_index) const
{
if (mark_index >= getMarksCount())
throw Exception(ErrorCodes::LOGICAL_ERROR, "Trying to get non existing mark {}, while size is {}", mark_index, getMarksCount());
if (mark_index == 0)
return marks_rows_partial_sums[0];
return marks_rows_partial_sums[mark_index] - marks_rows_partial_sums[mark_index - 1];
}
size_t MergeTreeIndexGranularity::getMarkStartingRow(size_t mark_index) const
{
if (mark_index == 0)
return 0;
return marks_rows_partial_sums[mark_index - 1];
}
size_t MergeTreeIndexGranularity::getMarksCount() const
{
return marks_rows_partial_sums.size();
}
size_t MergeTreeIndexGranularity::getTotalRows() const
{
if (marks_rows_partial_sums.empty())
return 0;
return marks_rows_partial_sums.back();
}
void MergeTreeIndexGranularity::appendMark(size_t rows_count)
{
if (marks_rows_partial_sums.empty())
marks_rows_partial_sums.push_back(rows_count);
else
marks_rows_partial_sums.push_back(marks_rows_partial_sums.back() + rows_count);
}
void MergeTreeIndexGranularity::addRowsToLastMark(size_t rows_count)
{
if (marks_rows_partial_sums.empty())
marks_rows_partial_sums.push_back(rows_count);
else
marks_rows_partial_sums.back() += rows_count;
}
void MergeTreeIndexGranularity::popMark()
{
if (!marks_rows_partial_sums.empty())
marks_rows_partial_sums.pop_back();
}
size_t MergeTreeIndexGranularity::getRowsCountInRange(size_t begin, size_t end) const
{
size_t subtrahend = 0;
if (begin != 0)
subtrahend = marks_rows_partial_sums[begin - 1];
return marks_rows_partial_sums[end - 1] - subtrahend;
extern const MergeTreeSettingsUInt64 index_granularity;
extern const MergeTreeSettingsUInt64 index_granularity_bytes;
extern const MergeTreeSettingsBool use_const_adaptive_granularity;
}
size_t MergeTreeIndexGranularity::getRowsCountInRange(const MarkRange & range) const
@ -87,39 +35,111 @@ size_t MergeTreeIndexGranularity::getRowsCountInRanges(const MarkRanges & ranges
return total;
}
size_t MergeTreeIndexGranularity::countMarksForRows(size_t from_mark, size_t number_of_rows) const
size_t MergeTreeIndexGranularity::getMarksCountWithoutFinal() const
{
size_t rows_before_mark = getMarkStartingRow(from_mark);
size_t last_row_pos = rows_before_mark + number_of_rows;
auto it = std::upper_bound(marks_rows_partial_sums.begin(), marks_rows_partial_sums.end(), last_row_pos);
size_t to_mark = it - marks_rows_partial_sums.begin();
return to_mark - from_mark;
size_t total = getMarksCount();
if (total == 0)
return total;
return total - hasFinalMark();
}
size_t MergeTreeIndexGranularity::countRowsForRows(size_t from_mark, size_t number_of_rows, size_t offset_in_rows) const
size_t MergeTreeIndexGranularity::getLastMarkRows() const
{
size_t rows_before_mark = getMarkStartingRow(from_mark);
size_t last_row_pos = rows_before_mark + offset_in_rows + number_of_rows;
auto it = std::upper_bound(marks_rows_partial_sums.begin(), marks_rows_partial_sums.end(), last_row_pos);
size_t to_mark = it - marks_rows_partial_sums.begin();
return getRowsCountInRange(from_mark, std::max(1UL, to_mark)) - offset_in_rows;
return getMarkRows(getMarksCount() - 1);
}
void MergeTreeIndexGranularity::resizeWithFixedGranularity(size_t size, size_t fixed_granularity)
size_t MergeTreeIndexGranularity::getLastNonFinalMarkRows() const
{
marks_rows_partial_sums.resize(size);
size_t last_mark_rows = getLastMarkRows();
if (last_mark_rows != 0)
return last_mark_rows;
return getMarkRows(getMarksCount() - 2);
}
size_t prev = 0;
for (size_t i = 0; i < size; ++i)
void MergeTreeIndexGranularity::addRowsToLastMark(size_t rows_count)
{
if (hasFinalMark())
{
marks_rows_partial_sums[i] = fixed_granularity + prev;
prev = marks_rows_partial_sums[i];
throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot add rows to final mark");
}
else if (empty())
{
appendMark(rows_count);
}
else
{
adjustLastMark(getLastMarkRows() + rows_count);
}
}
std::string MergeTreeIndexGranularity::describe() const
size_t computeIndexGranularityForBlock(
size_t rows_in_block,
size_t bytes_in_block,
size_t index_granularity_bytes,
size_t fixed_index_granularity_rows,
bool blocks_are_granules,
bool can_use_adaptive_index_granularity)
{
return fmt::format("initialized: {}, marks_rows_partial_sums: [{}]", initialized, fmt::join(marks_rows_partial_sums, ", "));
size_t index_granularity_for_block;
if (!can_use_adaptive_index_granularity)
{
index_granularity_for_block = fixed_index_granularity_rows;
}
else
{
if (blocks_are_granules)
{
index_granularity_for_block = rows_in_block;
}
else if (bytes_in_block >= index_granularity_bytes)
{
size_t granules_in_block = bytes_in_block / index_granularity_bytes;
index_granularity_for_block = rows_in_block / granules_in_block;
}
else
{
size_t size_of_row_in_bytes = std::max(bytes_in_block / rows_in_block, 1UL);
index_granularity_for_block = index_granularity_bytes / size_of_row_in_bytes;
}
}
/// We should be less or equal than fixed index granularity.
/// But if block size is a granule size then do not adjust it.
/// Granularity greater than fixed granularity might come from compact part.
if (!blocks_are_granules)
index_granularity_for_block = std::min(fixed_index_granularity_rows, index_granularity_for_block);
/// Very rare case when index granularity bytes less than single row.
if (index_granularity_for_block == 0)
index_granularity_for_block = 1;
return index_granularity_for_block;
}
MergeTreeIndexGranularityPtr createMergeTreeIndexGranularity(
size_t rows_in_block,
size_t bytes_in_block,
const MergeTreeSettings & settings,
const MergeTreeIndexGranularityInfo & info,
bool blocks_are_granules)
{
bool use_adaptive_granularity = info.mark_type.adaptive;
bool use_const_adaptive_granularity = settings[MergeTreeSetting::use_const_adaptive_granularity];
bool is_compact_part = info.mark_type.part_type == MergeTreeDataPartType::Compact;
if (blocks_are_granules || is_compact_part || (use_adaptive_granularity && !use_const_adaptive_granularity))
return std::make_shared<MergeTreeIndexGranularityAdaptive>();
size_t computed_granularity = computeIndexGranularityForBlock(
rows_in_block,
bytes_in_block,
settings[MergeTreeSetting::index_granularity_bytes],
settings[MergeTreeSetting::index_granularity],
blocks_are_granules,
use_adaptive_granularity);
return std::make_shared<MergeTreeIndexGranularityConstant>(computed_granularity);
}
}

92
src/Storages/MergeTree/MergeTreeIndexGranularity.h
View File

@ -1,6 +1,7 @@
#pragma once
#include <vector>
#include <optional>
#include <Storages/MergeTree/MarkRange.h>
#include "Storages/MergeTree/MergeTreeSettings.h"
namespace DB
{
@ -13,23 +14,19 @@ namespace DB
/// all values in inner vector would have constant stride (default 8192).
class MergeTreeIndexGranularity
{
private:
std::vector<size_t> marks_rows_partial_sums;
bool initialized = false;
public:
MergeTreeIndexGranularity() = default;
explicit MergeTreeIndexGranularity(const std::vector<size_t> & marks_rows_partial_sums_);
virtual ~MergeTreeIndexGranularity() = default;
/// Return count of rows between marks
size_t getRowsCountInRange(const MarkRange & range) const;
virtual size_t getRowsCountInRange(size_t begin, size_t end) const = 0;
/// Return count of rows between marks
size_t getRowsCountInRange(size_t begin, size_t end) const;
size_t getRowsCountInRange(const MarkRange & range) const;
/// Return sum of rows between all ranges
size_t getRowsCountInRanges(const MarkRanges & ranges) const;
/// Return number of marks, starting from `from_marks` that contain `number_of_rows`
size_t countMarksForRows(size_t from_mark, size_t number_of_rows) const;
virtual size_t countMarksForRows(size_t from_mark, size_t number_of_rows) const = 0;
/// Return number of rows, starting from `from_mark`, that contains amount of `number_of_rows`
/// and possible some offset_in_rows from `from_mark`
@ -37,69 +34,64 @@ public:
/// |-----|---------------------------|----|----|
/// ^------------------------^-----------^
//// from_mark offset_in_rows number_of_rows
size_t countRowsForRows(size_t from_mark, size_t number_of_rows, size_t offset_in_rows) const;
virtual size_t countRowsForRows(size_t from_mark, size_t number_of_rows, size_t offset_in_rows) const = 0;
/// Total marks
size_t getMarksCount() const;
virtual size_t getMarksCount() const = 0;
/// Total rows
size_t getTotalRows() const;
virtual size_t getTotalRows() const = 0;
/// Total number marks without final mark if it exists
size_t getMarksCountWithoutFinal() const { return getMarksCount() - hasFinalMark(); }
size_t getMarksCountWithoutFinal() const;
/// Rows after mark to next mark
size_t getMarkRows(size_t mark_index) const;
virtual size_t getMarkRows(size_t mark_index) const = 0;
/// Return amount of rows before mark
size_t getMarkStartingRow(size_t mark_index) const;
virtual size_t getMarkStartingRow(size_t mark_index) const = 0;
/// Amount of rows after last mark
size_t getLastMarkRows() const
{
size_t last = marks_rows_partial_sums.size() - 1;
return getMarkRows(last);
}
size_t getLastMarkRows() const;
size_t getLastNonFinalMarkRows() const
{
size_t last_mark_rows = getLastMarkRows();
if (last_mark_rows != 0)
return last_mark_rows;
return getMarkRows(marks_rows_partial_sums.size() - 2);
}
/// Amount of rows after last non-final mark
size_t getLastNonFinalMarkRows() const;
bool hasFinalMark() const
{
return getLastMarkRows() == 0;
}
virtual bool hasFinalMark() const = 0;
bool empty() const
{
return marks_rows_partial_sums.empty();
}
bool empty() const { return getMarksCount() == 0; }
bool isInitialized() const
{
return initialized;
}
void setInitialized()
{
initialized = true;
}
/// Add new mark with rows_count
void appendMark(size_t rows_count);
virtual void appendMark(size_t rows_count) = 0;
/// Extends last mark by rows_count.
virtual void adjustLastMark(size_t rows_count) = 0;
void addRowsToLastMark(size_t rows_count);
/// Drops last mark if any exists.
void popMark();
virtual void shrinkToFitInMemory() = 0;
/// Add `size` of marks with `fixed_granularity` rows
void resizeWithFixedGranularity(size_t size, size_t fixed_granularity);
virtual std::shared_ptr<MergeTreeIndexGranularity> optimize() const = 0;
std::string describe() const;
virtual std::string describe() const = 0;
};
using MergeTreeIndexGranularityPtr = std::shared_ptr<MergeTreeIndexGranularity>;
size_t computeIndexGranularityForBlock(
size_t rows_in_block,
size_t bytes_in_block,
size_t index_granularity_bytes,
size_t fixed_index_granularity_rows,
bool blocks_are_granules,
bool can_use_adaptive_index_granularity);
struct MergeTreeSettings;
struct MergeTreeIndexGranularityInfo;
MergeTreeIndexGranularityPtr createMergeTreeIndexGranularity(
size_t rows_in_block,
size_t bytes_in_block,
const MergeTreeSettings & settings,
const MergeTreeIndexGranularityInfo & info,
bool blocks_are_granules);
}

151
src/Storages/MergeTree/MergeTreeIndexGranularityAdaptive.cpp Normal file
View File

@ -0,0 +1,151 @@
#include <memory>
#include <Storages/MergeTree/MergeTreeIndexGranularityAdaptive.h>
#include <IO/WriteHelpers.h>
#include "Common/Logger.h"
#include "Common/logger_useful.h"
#include "Storages/MergeTree/MergeTreeIndexGranularityConstant.h"
namespace DB
{
namespace ErrorCodes
{
extern const int LOGICAL_ERROR;
}
MergeTreeIndexGranularityAdaptive::MergeTreeIndexGranularityAdaptive(const std::vector<size_t> & marks_rows_partial_sums_)
: marks_rows_partial_sums(marks_rows_partial_sums_)
{
}
/// Rows after mark to next mark
size_t MergeTreeIndexGranularityAdaptive::getMarkRows(size_t mark_index) const
{
if (mark_index >= getMarksCount())
throw Exception(ErrorCodes::LOGICAL_ERROR, "Trying to get non existing mark {}, while size is {}", mark_index, getMarksCount());
if (mark_index == 0)
return marks_rows_partial_sums[0];
return marks_rows_partial_sums[mark_index] - marks_rows_partial_sums[mark_index - 1];
}
size_t MergeTreeIndexGranularityAdaptive::getMarkStartingRow(size_t mark_index) const
{
if (mark_index > getMarksCount())
throw Exception(ErrorCodes::LOGICAL_ERROR, "Trying to get non existing mark {}, while size is {}", mark_index, getMarksCount());
if (mark_index == 0)
return 0;
return marks_rows_partial_sums[mark_index - 1];
}
bool MergeTreeIndexGranularityAdaptive::hasFinalMark() const
{
if (marks_rows_partial_sums.empty())
return false;
return marks_rows_partial_sums.back() == 0;
}
size_t MergeTreeIndexGranularityAdaptive::getMarksCount() const
{
return marks_rows_partial_sums.size();
}
size_t MergeTreeIndexGranularityAdaptive::getTotalRows() const
{
if (marks_rows_partial_sums.empty())
return 0;
return marks_rows_partial_sums.back();
}
void MergeTreeIndexGranularityAdaptive::appendMark(size_t rows_count)
{
if (hasFinalMark())
{
throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot append mark after final");
}
else if (marks_rows_partial_sums.empty())
{
marks_rows_partial_sums.push_back(rows_count);
}
else
{
marks_rows_partial_sums.push_back(marks_rows_partial_sums.back() + rows_count);
}
}
void MergeTreeIndexGranularityAdaptive::adjustLastMark(size_t rows_count)
{
if (hasFinalMark())
{
throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot adjust final mark");
}
else if (marks_rows_partial_sums.empty())
{
marks_rows_partial_sums.push_back(rows_count);
}
else
{
marks_rows_partial_sums.pop_back();
appendMark(rows_count);
}
}
size_t MergeTreeIndexGranularityAdaptive::getRowsCountInRange(size_t begin, size_t end) const
{
size_t subtrahend = 0;
if (begin != 0)
subtrahend = marks_rows_partial_sums[begin - 1];
return marks_rows_partial_sums[end - 1] - subtrahend;
}
size_t MergeTreeIndexGranularityAdaptive::countMarksForRows(size_t from_mark, size_t number_of_rows) const
{
size_t rows_before_mark = getMarkStartingRow(from_mark);
size_t last_row_pos = rows_before_mark + number_of_rows;
auto it = std::upper_bound(marks_rows_partial_sums.begin(), marks_rows_partial_sums.end(), last_row_pos);
size_t to_mark = it - marks_rows_partial_sums.begin();
return to_mark - from_mark;
}
size_t MergeTreeIndexGranularityAdaptive::countRowsForRows(size_t from_mark, size_t number_of_rows, size_t offset_in_rows) const
{
size_t rows_before_mark = getMarkStartingRow(from_mark);
size_t last_row_pos = rows_before_mark + offset_in_rows + number_of_rows;
auto it = std::upper_bound(marks_rows_partial_sums.begin(), marks_rows_partial_sums.end(), last_row_pos);
size_t to_mark = it - marks_rows_partial_sums.begin();
return getRowsCountInRange(from_mark, std::max(1UL, to_mark)) - offset_in_rows;
}
void MergeTreeIndexGranularityAdaptive::shrinkToFitInMemory()
{
marks_rows_partial_sums.shrink_to_fit();
}
std::shared_ptr<MergeTreeIndexGranularity> MergeTreeIndexGranularityAdaptive::optimize() const
{
size_t marks_count = getMarksCountWithoutFinal();
if (marks_count == 0)
return nullptr;
size_t first_mark = getMarkRows(0);
for (size_t i = 1; i < marks_count - 1; ++i)
{
if (getMarkRows(i) != first_mark)
return nullptr;
}
size_t last_mark = getMarkRows(marks_count - 1);
return std::make_shared<MergeTreeIndexGranularityConstant>(first_mark, last_mark, marks_count, hasFinalMark());
}
std::string MergeTreeIndexGranularityAdaptive::describe() const
{
return fmt::format("Adaptive(marks_rows_partial_sums: [{}])", fmt::join(marks_rows_partial_sums, ", "));
}
}

36
src/Storages/MergeTree/MergeTreeIndexGranularityAdaptive.h Normal file
View File

@ -0,0 +1,36 @@
#pragma once
#include <Storages/MergeTree/MergeTreeIndexGranularity.h>
namespace DB
{
class MergeTreeIndexGranularityAdaptive : public MergeTreeIndexGranularity
{
public:
MergeTreeIndexGranularityAdaptive() = default;
explicit MergeTreeIndexGranularityAdaptive(const std::vector<size_t> & marks_rows_partial_sums_);
size_t getRowsCountInRange(size_t begin, size_t end) const override;
size_t countMarksForRows(size_t from_mark, size_t number_of_rows) const override;
size_t countRowsForRows(size_t from_mark, size_t number_of_rows, size_t offset_in_rows) const override;
size_t getMarksCount() const override;
size_t getTotalRows() const override;
size_t getMarkRows(size_t mark_index) const override;
size_t getMarkStartingRow(size_t mark_index) const override;
bool hasFinalMark() const override;
void appendMark(size_t rows_count) override;
void adjustLastMark(size_t rows_count) override;
void shrinkToFitInMemory() override;
std::shared_ptr<MergeTreeIndexGranularity> optimize() const override;
std::string describe() const override;
private:
std::vector<size_t> marks_rows_partial_sums;
};
}

150
src/Storages/MergeTree/MergeTreeIndexGranularityConstant.cpp Normal file
View File

@ -0,0 +1,150 @@
#include <Storages/MergeTree/MergeTreeIndexGranularityConstant.h>
#include <Common/Exception.h>
#include <IO/WriteHelpers.h>
namespace DB
{
namespace ErrorCodes
{
extern const int LOGICAL_ERROR;
}
MergeTreeIndexGranularityConstant::MergeTreeIndexGranularityConstant(size_t constant_granularity_)
: constant_granularity(constant_granularity_)
, last_mark_granularity(constant_granularity_)
{
}
MergeTreeIndexGranularityConstant::MergeTreeIndexGranularityConstant(size_t constant_granularity_, size_t last_mark_granularity_, size_t num_marks_without_final_, bool has_final_mark_)
: constant_granularity(constant_granularity_)
, last_mark_granularity(last_mark_granularity_)
, num_marks_without_final(num_marks_without_final_)
, has_final_mark(has_final_mark_)
{
}
/// Rows after mark to next mark
size_t MergeTreeIndexGranularityConstant::getMarkRows(size_t mark_index) const
{
if (mark_index >= getMarksCount())
throw Exception(ErrorCodes::LOGICAL_ERROR, "Trying to get non existing mark {}, while size is {}", mark_index, getMarksCount());
if (mark_index + 1 < num_marks_without_final)
return constant_granularity;
if (mark_index + 1 == num_marks_without_final)
return last_mark_granularity;
return 0; // Final mark.
}
size_t MergeTreeIndexGranularityConstant::getMarkStartingRow(size_t mark_index) const
{
if (mark_index > getMarksCount())
throw Exception(ErrorCodes::LOGICAL_ERROR, "Trying to get non existing mark {}, while size is {}", mark_index, getMarksCount());
size_t total_rows = 0;
if (mark_index >= num_marks_without_final && mark_index != 0)
{
total_rows += last_mark_granularity;
mark_index = num_marks_without_final - 1;
}
total_rows += constant_granularity * mark_index;
return total_rows;
}
size_t MergeTreeIndexGranularityConstant::getMarksCount() const
{
return num_marks_without_final + has_final_mark;
}
size_t MergeTreeIndexGranularityConstant::getTotalRows() const
{
if (num_marks_without_final == 0)
return 0;
return constant_granularity * (num_marks_without_final - 1) + last_mark_granularity;
}
void MergeTreeIndexGranularityConstant::appendMark(size_t rows_count)
{
if (has_final_mark)
{
throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot append mark after final");
}
else if (rows_count == 0)
{
has_final_mark = true;
}
else if (rows_count != constant_granularity)
{
throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot append mark with {} rows. Granularity is constant ({})", rows_count, constant_granularity);
}
else
{
++num_marks_without_final;
}
}
void MergeTreeIndexGranularityConstant::adjustLastMark(size_t rows_count)
{
if (has_final_mark)
{
throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot adjust final mark");
}
else
{
if (num_marks_without_final == 0)
++num_marks_without_final;
last_mark_granularity = rows_count;
}
}
size_t MergeTreeIndexGranularityConstant::getRowsCountInRange(size_t begin, size_t end) const
{
size_t total_rows = 0;
if (end >= num_marks_without_final && end != 0)
{
total_rows += last_mark_granularity;
end = num_marks_without_final - 1;
}
total_rows += constant_granularity * (end - begin);
return total_rows;
}
size_t MergeTreeIndexGranularityConstant::countMarksForRows(size_t from_mark, size_t number_of_rows) const
{
size_t rows_before_mark = getMarkStartingRow(from_mark);
size_t last_row_pos = rows_before_mark + number_of_rows;
if (last_row_pos >= (num_marks_without_final - 1) * constant_granularity)
return num_marks_without_final - from_mark;
return (last_row_pos + constant_granularity - 1) / constant_granularity - from_mark;
}
size_t MergeTreeIndexGranularityConstant::countRowsForRows(size_t from_mark, size_t number_of_rows, size_t offset_in_rows) const
{
UNUSED(from_mark, number_of_rows, offset_in_rows);
return 0;
// size_t rows_before_mark = getMarkStartingRow(from_mark);
// size_t last_row_pos = rows_before_mark + offset_in_rows + number_of_rows;
// auto it = std::upper_bound(marks_rows_partial_sums.begin(), marks_rows_partial_sums.end(), last_row_pos);
// size_t to_mark = it - marks_rows_partial_sums.begin();
// return getRowsCountInRange(from_mark, std::max(1UL, to_mark)) - offset_in_rows;
}
std::string MergeTreeIndexGranularityConstant::describe() const
{
return fmt::format(
"Constant(constant_granularity: {}, last_mark_granularity: {}, num_marks_without_final: {}, has_final_mark: {}",
constant_granularity, last_mark_granularity, num_marks_without_final, has_final_mark);
}
}

41
src/Storages/MergeTree/MergeTreeIndexGranularityConstant.h Normal file
View File

@ -0,0 +1,41 @@
#pragma once
#include <Storages/MergeTree/MergeTreeIndexGranularity.h>
namespace DB
{
class MergeTreeIndexGranularityConstant : public MergeTreeIndexGranularity
{
private:
size_t constant_granularity;
size_t last_mark_granularity;
size_t num_marks_without_final = 0;
bool has_final_mark = false;
public:
MergeTreeIndexGranularityConstant() = default;
explicit MergeTreeIndexGranularityConstant(size_t constant_granularity_);
MergeTreeIndexGranularityConstant(size_t constant_granularity_, size_t last_mark_granularity_, size_t num_marks_without_final_, bool has_final_mark_);
size_t getRowsCountInRange(size_t begin, size_t end) const override;
size_t countMarksForRows(size_t from_mark, size_t number_of_rows) const override;
size_t countRowsForRows(size_t from_mark, size_t number_of_rows, size_t offset_in_rows) const override;
size_t getMarksCount() const override;
size_t getTotalRows() const override;
size_t getMarkRows(size_t mark_index) const override;
size_t getMarkStartingRow(size_t mark_index) const override;
bool hasFinalMark() const override { return has_final_mark; }
void appendMark(size_t rows_count) override;
void adjustLastMark(size_t rows_count) override;
void shrinkToFitInMemory() override {}
std::shared_ptr<MergeTreeIndexGranularity> optimize() const override { return nullptr; }
std::string describe() const override;
};
}

2
src/Storages/MergeTree/MergeTreeIndexGranularityInfo.h
View File

@ -4,12 +4,12 @@
#include <base/types.h>
#include <Storages/MergeTree/MergeTreeDataPartType.h>
#include <Disks/IDisk.h>
#include <Storages/MergeTree/IDataPartStorage.h>
namespace DB
{
class MergeTreeData;
class IDataPartStorage;
/** Various types of mark files are stored in files with various extensions:

1
src/Storages/MergeTree/MergeTreeRangeReader.cpp
View File

@ -4,6 +4,7 @@
#include <Columns/FilterDescription.h>
#include <Columns/ColumnConst.h>
#include <Columns/ColumnsCommon.h>
#include "Common/Logger.h"
#include <Common/TargetSpecific.h>
#include <Common/logger_useful.h>
#include <Core/UUID.h>

2
src/Storages/MergeTree/MergeTreeReadTask.cpp
View File

@ -151,7 +151,7 @@ UInt64 MergeTreeReadTask::estimateNumRows() const
return rows_to_read;
const auto & index_granularity = info->data_part->index_granularity;
return index_granularity.countRowsForRows(range_readers.main.currentMark(), rows_to_read, range_readers.main.numReadRowsInCurrentGranule());
return index_granularity->countRowsForRows(range_readers.main.currentMark(), rows_to_read, range_readers.main.numReadRowsInCurrentGranule());
}
MergeTreeReadTask::BlockAndProgress MergeTreeReadTask::read()

2
src/Storages/MergeTree/MergeTreeSequentialSource.cpp
View File

@ -229,7 +229,7 @@ try
if (!isCancelled() && current_row < data_part->rows_count)
{
size_t rows_to_read = data_part->index_granularity.getMarkRows(current_mark);
size_t rows_to_read = data_part->index_granularity->getMarkRows(current_mark);
bool continue_reading = (current_mark != 0);
const auto & sample = reader->getColumns();

2
src/Storages/MergeTree/MergeTreeSettings.cpp
View File

@ -185,6 +185,8 @@ namespace ErrorCodes
DECLARE(UInt64, min_merge_bytes_to_use_direct_io, 10ULL * 1024 * 1024 * 1024, "Minimal amount of bytes to enable O_DIRECT in merge (0 - disabled).", 0) \
DECLARE(UInt64, index_granularity_bytes, 10 * 1024 * 1024, "Approximate amount of bytes in single granule (0 - disabled).", 0) \
DECLARE(UInt64, min_index_granularity_bytes, 1024, "Minimum amount of bytes in single granule.", 1024) \
DECLARE(Bool, use_const_adaptive_granularity, false, "KEK KEK KEK KEK KEK KEK", 0) \
DECLARE(Bool, enable_index_granularity_compression, true, "KEK KEK KEK KEK KEK KEK", 0) \
DECLARE(Int64, merge_with_ttl_timeout, 3600 * 4, "Minimal time in seconds, when merge with delete TTL can be repeated.", 0) \
DECLARE(Int64, merge_with_recompression_ttl_timeout, 3600 * 4, "Minimal time in seconds, when merge with recompression TTL can be repeated.", 0) \
DECLARE(Bool, ttl_only_drop_parts, false, "Only drop altogether the expired parts and not partially prune them.", 0) \

37
src/Storages/MergeTree/MergedBlockOutputStream.cpp
View File

@ -1,8 +1,10 @@
#include <optional>
#include <Storages/MergeTree/MergedBlockOutputStream.h>
#include <IO/HashingWriteBuffer.h>
#include <Interpreters/Context.h>
#include <Interpreters/MergeTreeTransaction.h>
#include <Parsers/queryToString.h>
#include "Common/Logger.h"
#include <Common/logger_useful.h>
#include <Core/Settings.h>
@ -15,6 +17,10 @@ namespace ErrorCodes
extern const int LOGICAL_ERROR;
}
namespace MergeTreeSetting
{
extern MergeTreeSettingsBool enable_index_granularity_compression;
}
MergedBlockOutputStream::MergedBlockOutputStream(
const MergeTreeMutableDataPartPtr & data_part,
@ -23,12 +29,12 @@ MergedBlockOutputStream::MergedBlockOutputStream(
const MergeTreeIndices & skip_indices,
const ColumnsStatistics & statistics,
CompressionCodecPtr default_codec_,
MergeTreeIndexGranularityPtr index_granularity_ptr,
TransactionID tid,
bool reset_columns_,
bool save_marks_in_cache,
bool blocks_are_granules_size,
const WriteSettings & write_settings_,
const MergeTreeIndexGranularity & computed_index_granularity)
const WriteSettings & write_settings_)
: IMergedBlockOutputStream(data_part->storage.getSettings(), data_part->getDataPartStoragePtr(), metadata_snapshot_, columns_list_, reset_columns_)
, columns_list(columns_list_)
, default_codec(default_codec_)
@ -53,11 +59,22 @@ MergedBlockOutputStream::MergedBlockOutputStream(
data_part->storeVersionMetadata();
writer = createMergeTreeDataPartWriter(data_part->getType(),
data_part->name, data_part->storage.getLogName(), data_part->getSerializations(),
data_part_storage, data_part->index_granularity_info,
storage_settings,
columns_list, data_part->getColumnPositions(), metadata_snapshot, data_part->storage.getVirtualsPtr(),
skip_indices, statistics, data_part->getMarksFileExtension(), default_codec, writer_settings, computed_index_granularity);
data_part->name,
data_part->storage.getLogName(),
data_part->getSerializations(),
data_part_storage,
data_part->index_granularity_info,
storage_settings,
columns_list,
data_part->getColumnPositions(),
metadata_snapshot,
data_part->storage.getVirtualsPtr(),
skip_indices,
statistics,
data_part->getMarksFileExtension(),
default_codec,
writer_settings,
std::move(index_granularity_ptr));
}
/// If data is pre-sorted.
@ -209,6 +226,12 @@ MergedBlockOutputStream::Finalizer MergedBlockOutputStream::finalizePartAsync(
new_part->index_granularity = writer->getIndexGranularity();
new_part->calculateColumnsAndSecondaryIndicesSizesOnDisk();
if ((*new_part->storage.getSettings())[MergeTreeSetting::enable_index_granularity_compression])
{
if (auto new_index_granularity = new_part->index_granularity->optimize())
new_part->index_granularity = std::move(new_index_granularity);
}
/// In mutation, existing_rows_count is already calculated in PartMergerWriter
/// In merge situation, lightweight deleted rows was physically deleted, existing_rows_count equals rows_count
if (!new_part->existing_rows_count.has_value())

4
src/Storages/MergeTree/MergedBlockOutputStream.h
View File

@ -22,12 +22,12 @@ public:
const MergeTreeIndices & skip_indices,
const ColumnsStatistics & statistics,
CompressionCodecPtr default_codec_,
MergeTreeIndexGranularityPtr index_granularity_ptr,
TransactionID tid,
bool reset_columns_ = false,
bool save_marks_in_cache = false,
bool blocks_are_granules_size = false,
const WriteSettings & write_settings = {},
const MergeTreeIndexGranularity & computed_index_granularity = {});
const WriteSettings & write_settings = {});
Block getHeader() const { return metadata_snapshot->getSampleBlock(); }

24
src/Storages/MergeTree/MergedColumnOnlyOutputStream.cpp
View File

@ -15,25 +15,25 @@ MergedColumnOnlyOutputStream::MergedColumnOnlyOutputStream(
const MergeTreeMutableDataPartPtr & data_part,
const StorageMetadataPtr & metadata_snapshot_,
const NamesAndTypesList & columns_list_,
CompressionCodecPtr default_codec,
const MergeTreeIndices & indices_to_recalc,
const ColumnsStatistics & stats_to_recalc_,
WrittenOffsetColumns * offset_columns_,
bool save_marks_in_cache,
const MergeTreeIndexGranularity & index_granularity,
const MergeTreeIndexGranularityInfo * index_granularity_info)
const ColumnsStatistics & stats_to_recalc,
CompressionCodecPtr default_codec,
MergeTreeIndexGranularityPtr index_granularity_ptr,
WrittenOffsetColumns * offset_columns,
bool save_marks_in_cache)
: IMergedBlockOutputStream(data_part->storage.getSettings(), data_part->getDataPartStoragePtr(), metadata_snapshot_, columns_list_, /*reset_columns=*/ true)
{
const auto & global_settings = data_part->storage.getContext()->getSettingsRef();
/// Granularity is never recomputed while writing only columns.
MergeTreeWriterSettings writer_settings(
global_settings,
data_part->storage.getContext()->getWriteSettings(),
storage_settings,
index_granularity_info ? index_granularity_info->mark_type.adaptive : data_part->storage.canUseAdaptiveGranularity(),
/* rewrite_primary_key = */ false,
data_part->index_granularity_info.mark_type.adaptive,
/*rewrite_primary_key=*/ false,
save_marks_in_cache,
/* blocks_are_granules_size = */ false);
/*blocks_are_granules_size=*/ false);
writer = createMergeTreeDataPartWriter(
data_part->getType(),
@ -45,17 +45,17 @@ MergedColumnOnlyOutputStream::MergedColumnOnlyOutputStream(
metadata_snapshot_,
data_part->storage.getVirtualsPtr(),
indices_to_recalc,
stats_to_recalc_,
stats_to_recalc,
data_part->getMarksFileExtension(),
default_codec,
writer_settings,
index_granularity);
std::move(index_granularity_ptr));
auto * writer_on_disk = dynamic_cast<MergeTreeDataPartWriterOnDisk *>(writer.get());
if (!writer_on_disk)
throw Exception(ErrorCodes::NOT_IMPLEMENTED, "MergedColumnOnlyOutputStream supports only parts stored on disk");
writer_on_disk->setWrittenOffsetColumns(offset_columns_);
writer_on_disk->setWrittenOffsetColumns(offset_columns);
}
void MergedColumnOnlyOutputStream::write(const Block & block)

13
src/Storages/MergeTree/MergedColumnOnlyOutputStream.h
View File

@ -18,13 +18,12 @@ public:
const MergeTreeMutableDataPartPtr & data_part,
const StorageMetadataPtr & metadata_snapshot_,
const NamesAndTypesList & columns_list_,
CompressionCodecPtr default_codec_,
const MergeTreeIndices & indices_to_recalc_,
const ColumnsStatistics & stats_to_recalc_,
WrittenOffsetColumns * offset_columns_ = nullptr,
bool save_marks_in_cache = false,
const MergeTreeIndexGranularity & index_granularity = {},
const MergeTreeIndexGranularityInfo * index_granularity_info_ = nullptr);
const MergeTreeIndices & indices_to_recalc,
const ColumnsStatistics & stats_to_recalc,
CompressionCodecPtr default_codec,
MergeTreeIndexGranularityPtr index_granularity_ptr,
WrittenOffsetColumns * offset_columns = nullptr,
bool save_marks_in_cache = false);
void write(const Block & block) override;

28
src/Storages/MergeTree/MutateTask.cpp
View File

@ -35,6 +35,7 @@
#include <DataTypes/DataTypeVariant.h>
#include <boost/algorithm/string/replace.hpp>
#include <Common/ProfileEventsScope.h>
#include "Storages/MergeTree/MergeTreeIndexGranularity.h"
#include <Core/ColumnsWithTypeAndName.h>
@ -1597,7 +1598,6 @@ private:
ctx->minmax_idx = std::make_shared<IMergeTreeDataPart::MinMaxIndex>();
MergeTreeIndexGranularity computed_granularity;
bool has_delete = false;
for (auto & command_for_interpreter : ctx->for_interpreter)
@ -1610,9 +1610,21 @@ private:
}
}
MergeTreeIndexGranularityPtr index_granularity_ptr;
/// Reuse source part granularity if mutation does not change number of rows
if (!has_delete && ctx->execute_ttl_type == ExecuteTTLType::NONE)
computed_granularity = ctx->source_part->index_granularity;
{
index_granularity_ptr = ctx->source_part->index_granularity;
}
else
{
index_granularity_ptr = createMergeTreeIndexGranularity(
ctx->new_data_part->rows_count,
ctx->new_data_part->getBytesUncompressedOnDisk(),
*ctx->data->getSettings(),
ctx->new_data_part->index_granularity_info,
/*blocks_are_granules=*/ false);
}
ctx->out = std::make_shared<MergedBlockOutputStream>(
ctx->new_data_part,
@ -1621,12 +1633,12 @@ private:
skip_indices,
stats_to_rewrite,
ctx->compression_codec,
std::move(index_granularity_ptr),
ctx->txn ? ctx->txn->tid : Tx::PrehistoricTID,
/*reset_columns=*/ true,
/*save_marks_in_cache=*/ false,
/*blocks_are_granules_size=*/ false,
ctx->context->getWriteSettings(),
computed_granularity);
ctx->context->getWriteSettings());
ctx->mutating_pipeline = QueryPipelineBuilder::getPipeline(std::move(*builder));
ctx->mutating_pipeline.setProgressCallback(ctx->progress_callback);
@ -1848,14 +1860,10 @@ private:
ctx->new_data_part,
ctx->metadata_snapshot,
ctx->updated_header.getNamesAndTypesList(),
ctx->compression_codec,
std::vector<MergeTreeIndexPtr>(ctx->indices_to_recalc.begin(), ctx->indices_to_recalc.end()),
ColumnsStatistics(ctx->stats_to_recalc.begin(), ctx->stats_to_recalc.end()),
nullptr,
/*save_marks_in_cache=*/ false,
ctx->source_part->index_granularity,
&ctx->source_part->index_granularity_info
);
ctx->compression_codec,
ctx->source_part->index_granularity);
ctx->mutating_pipeline = QueryPipelineBuilder::getPipeline(std::move(*builder));
ctx->mutating_pipeline.setProgressCallback(ctx->progress_callback);

2
src/Storages/MergeTree/RangesInDataPart.cpp
View File

@ -99,7 +99,7 @@ size_t RangesInDataPart::getMarksCount() const
size_t RangesInDataPart::getRowsCount() const
{
return data_part->index_granularity.getRowsCountInRanges(ranges);
return data_part->index_granularity->getRowsCountInRanges(ranges);
}

6
src/Storages/StorageMergeTreeIndex.cpp
View File

@ -63,7 +63,7 @@ protected:
marks_loader = createMarksLoader(part, MergeTreeDataPartCompact::DATA_FILE_NAME, part->getColumns().size());
size_t num_columns = header.columns();
size_t num_rows = index_granularity.getMarksCount();
size_t num_rows = index_granularity->getMarksCount();
const auto & part_name_column = StorageMergeTreeIndex::part_name_column;
const auto & mark_number_column = StorageMergeTreeIndex::mark_number_column;
@ -115,7 +115,7 @@ protected:
data.resize(num_rows);
for (size_t i = 0; i < num_rows; ++i)
data[i] = index_granularity.getMarkRows(i);
data[i] = index_granularity->getMarkRows(i);
result_columns[pos] = std::move(column);
}
@ -159,7 +159,7 @@ private:
{
size_t col_idx = 0;
bool has_marks_in_part = false;
size_t num_rows = part->index_granularity.getMarksCount();
size_t num_rows = part->index_granularity->getMarksCount();
if (isWidePart(part))
{