Exclude partition key columns from SummingMergeTree by nikitamikhaylov · Pull Request #12173 · ClickHouse/ClickHouse · GitHub
Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 0 additions & 11 deletions programs/copier/Internals.cpp
3 changes: 1 addition & 2 deletions programs/copier/Internals.h
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@
#include <Columns/ColumnString.h>
#include <Columns/ColumnsNumber.h>
#include <DataTypes/DataTypeString.h>
#include <DataTypes/NestedUtils.h>
#include <Parsers/ParserCreateQuery.h>
#include <Parsers/parseQuery.h>
#include <Parsers/ParserQuery.h>
Expand Down Expand Up @@ -197,8 +198,6 @@ ASTPtr extractPrimaryKey(const ASTPtr & storage_ast);

ASTPtr extractOrderBy(const ASTPtr & storage_ast);

String createCommaSeparatedStringFrom(const Names & names);

Names extractPrimaryKeyColumnNames(const ASTPtr & storage_ast);

String extractReplicatedTableZookeeperPath(const ASTPtr & storage_ast);
Expand Down
2 changes: 1 addition & 1 deletion programs/copier/TaskTableAndShard.h
Original file line number Diff line number Diff line change
Expand Up @@ -268,7 +268,7 @@ inline TaskTable::TaskTable(TaskCluster & parent, const Poco::Util::AbstractConf
ParserStorage parser_storage;
engine_push_ast = parseQuery(parser_storage, engine_push_str, 0, DBMS_DEFAULT_MAX_PARSER_DEPTH);
engine_push_partition_key_ast = extractPartitionKey(engine_push_ast);
primary_key_comma_separated = createCommaSeparatedStringFrom(extractPrimaryKeyColumnNames(engine_push_ast));
primary_key_comma_separated = Nested::createCommaSeparatedStringFrom(extractPrimaryKeyColumnNames(engine_push_ast));
engine_push_zk_path = extractReplicatedTableZookeeperPath(engine_push_ast);
}

Expand Down
11 changes: 11 additions & 0 deletions src/DataTypes/NestedUtils.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,17 @@ std::pair<std::string, std::string> splitName(const std::string & name)
return {{ begin, first_end }, { second_begin, end }};
}

std::string createCommaSeparatedStringFrom(const Names & names)
{
std::ostringstream ss;
if (!names.empty())
{
std::copy(names.begin(), std::prev(names.end()), std::ostream_iterator<std::string>(ss, ", "));
ss << names.back();
}
return ss.str();
}


std::string extractTableName(const std::string & nested_name)
{
Expand Down
2 changes: 2 additions & 0 deletions src/DataTypes/NestedUtils.h
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,8 @@ namespace Nested

std::pair<std::string, std::string> splitName(const std::string & name);

std::string createCommaSeparatedStringFrom(const Names & names);

/// Returns the prefix of the name to the first '.'. Or the name is unchanged if there is no dot.
std::string extractTableName(const std::string & nested_name);

Expand Down
16 changes: 12 additions & 4 deletions src/Processors/Merges/Algorithms/SummingSortedAlgorithm.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,12 @@ static bool isInPrimaryKey(const SortDescription & description, const std::strin
return false;
}

static bool isInPartitionKey(const std::string & column_name, const Names & partition_key_columns)
{
auto is_in_partition_key = std::find(partition_key_columns.begin(), partition_key_columns.end(), column_name);
return is_in_partition_key != partition_key_columns.end();
}

/// Returns true if merge result is not empty
static bool mergeMap(const SummingSortedAlgorithm::MapDescription & desc,
Row & row, const ColumnRawPtrs & raw_columns, size_t row_number)
Expand Down Expand Up @@ -181,7 +187,8 @@ static bool mergeMap(const SummingSortedAlgorithm::MapDescription & desc,
static SummingSortedAlgorithm::ColumnsDefinition defineColumns(
const Block & header,
const SortDescription & description,
const Names & column_names_to_sum)
const Names & column_names_to_sum,
const Names & partition_key_columns)
{
size_t num_columns = header.columns();
SummingSortedAlgorithm::ColumnsDefinition def;
Expand Down Expand Up @@ -223,8 +230,8 @@ static SummingSortedAlgorithm::ColumnsDefinition defineColumns(
continue;
}

/// Are they inside the PK?
if (isInPrimaryKey(description, column.name, i))
/// Are they inside the primary key or partiton key?
if (isInPrimaryKey(description, column.name, i) || isInPartitionKey(column.name, partition_key_columns))
{
def.column_numbers_not_to_aggregate.push_back(i);
continue;
Expand Down Expand Up @@ -617,9 +624,10 @@ SummingSortedAlgorithm::SummingSortedAlgorithm(
const Block & header, size_t num_inputs,
SortDescription description_,
const Names & column_names_to_sum,
const Names & partition_key_columns,
size_t max_block_size)
: IMergingAlgorithmWithDelayedChunk(num_inputs, std::move(description_))
, columns_definition(defineColumns(header, description, column_names_to_sum))
, columns_definition(defineColumns(header, description, column_names_to_sum, partition_key_columns))
, merged_data(getMergedDataColumns(header, columns_definition), max_block_size, columns_definition)
{
}
Expand Down
2 changes: 2 additions & 0 deletions src/Processors/Merges/Algorithms/SummingSortedAlgorithm.h
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,8 @@ class SummingSortedAlgorithm final : public IMergingAlgorithmWithDelayedChunk
SortDescription description_,
/// List of columns to be summed. If empty, all numeric columns that are not in the description are taken.
const Names & column_names_to_sum,
/// List of partition key columns. They have to be excluded.
const Names & partition_key_columns,
size_t max_block_size);

void initialize(Inputs inputs) override;
Expand Down
2 changes: 2 additions & 0 deletions src/Processors/Merges/SummingSortedTransform.h
Original file line number Diff line number Diff line change
Expand Up @@ -16,13 +16,15 @@ class SummingSortedTransform final : public IMergingTransform<SummingSortedAlgor
SortDescription description_,
/// List of columns to be summed. If empty, all numeric columns that are not in the description are taken.
const Names & column_names_to_sum,
const Names & partition_key_columns,
size_t max_block_size)
: IMergingTransform(
num_inputs, header, header, true,
header,
num_inputs,
std::move(description_),
column_names_to_sum,
partition_key_columns,
max_block_size)
{
}
Expand Down
23 changes: 20 additions & 3 deletions src/Storages/MergeTree/MergeTreeData.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,7 @@ namespace ErrorCodes
extern const int UNKNOWN_DISK;
extern const int NOT_ENOUGH_SPACE;
extern const int ALTER_OF_COLUMN_IS_FORBIDDEN;
extern const int BAD_ARGUMENTS;
}


Expand Down Expand Up @@ -169,7 +170,7 @@ MergeTreeData::MergeTreeData(
const auto settings = getSettings();

/// NOTE: using the same columns list as is read when performing actual merges.
merging_params.check(metadata_.getColumns().getAllPhysical());
merging_params.check(metadata_);

if (metadata_.sampling_key.definition_ast != nullptr)
{
Expand Down Expand Up @@ -521,8 +522,10 @@ void MergeTreeData::checkStoragePolicy(const StoragePolicyPtr & new_storage_poli
}


void MergeTreeData::MergingParams::check(const NamesAndTypesList & columns) const
void MergeTreeData::MergingParams::check(const StorageInMemoryMetadata & metadata) const
{
const auto columns = metadata.getColumns().getAllPhysical();

if (!sign_column.empty() && mode != MergingParams::Collapsing && mode != MergingParams::VersionedCollapsing)
throw Exception("Sign column for MergeTree cannot be specified in modes except Collapsing or VersionedCollapsing.",
ErrorCodes::LOGICAL_ERROR);
Expand Down Expand Up @@ -607,6 +610,21 @@ void MergeTreeData::MergingParams::check(const NamesAndTypesList & columns) cons
throw Exception(
"Column " + column_to_sum + " listed in columns to sum does not exist in table declaration.", ErrorCodes::NO_SUCH_COLUMN_IN_TABLE);
}

/// Check that summing columns are not in partition key.
if (metadata.isPartitionKeyDefined())
{
auto partition_key_columns = metadata.getPartitionKey().column_names;

Names names_intersection;
std::set_intersection(columns_to_sum.begin(), columns_to_sum.end(),
partition_key_columns.begin(), partition_key_columns.end(),
std::back_inserter(names_intersection));

if (!names_intersection.empty())
throw Exception("Colums: " + Nested::createCommaSeparatedStringFrom(names_intersection) +
" listed both in colums to sum and in partition key. That is not allowed.", ErrorCodes::BAD_ARGUMENTS);
}
}

if (mode == MergingParams::Replacing)
Expand Down Expand Up @@ -638,7 +656,6 @@ String MergeTreeData::MergingParams::getModeName() const
__builtin_unreachable();
}


Int64 MergeTreeData::getMaxBlockNumber() const
{
auto lock = lockParts();
Expand Down
2 changes: 1 addition & 1 deletion src/Storages/MergeTree/MergeTreeData.h
Original file line number Diff line number Diff line change
Expand Up @@ -301,7 +301,7 @@ class MergeTreeData : public IStorage
Graphite::Params graphite_params;

/// Check that needed columns are present and have correct types.
void check(const NamesAndTypesList & columns) const;
void check(const StorageInMemoryMetadata & metadata) const;

String getModeName() const;
};
Expand Down
4 changes: 3 additions & 1 deletion src/Storages/MergeTree/MergeTreeDataMergerMutator.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -739,6 +739,8 @@ MergeTreeData::MutableDataPartPtr MergeTreeDataMergerMutator::mergePartsToTempor
size_t sort_columns_size = sort_columns.size();
sort_description.reserve(sort_columns_size);

Names partition_key_columns = metadata_snapshot->getPartitionKey().column_names;

Block header = pipes.at(0).getHeader();
for (size_t i = 0; i < sort_columns_size; ++i)
sort_description.emplace_back(header.getPositionByName(sort_columns[i]), 1, 1);
Expand Down Expand Up @@ -767,7 +769,7 @@ MergeTreeData::MutableDataPartPtr MergeTreeDataMergerMutator::mergePartsToTempor

case MergeTreeData::MergingParams::Summing:
merged_transform = std::make_unique<SummingSortedTransform>(
header, pipes.size(), sort_description, data.merging_params.columns_to_sum, merge_block_size);
header, pipes.size(), sort_description, data.merging_params.columns_to_sum, partition_key_columns, merge_block_size);
break;

case MergeTreeData::MergingParams::Aggregating:
Expand Down
4 changes: 3 additions & 1 deletion src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -1160,6 +1160,8 @@ Pipes MergeTreeDataSelectExecutor::spreadMarkRangesAmongStreamsFinal(
size_t sort_columns_size = sort_columns.size();
sort_description.reserve(sort_columns_size);

Names partition_key_columns = metadata_snapshot->getPartitionKey().column_names;

Block header = pipes.at(0).getHeader();
for (size_t i = 0; i < sort_columns_size; ++i)
sort_description.emplace_back(header.getPositionByName(sort_columns[i]), 1, 1);
Expand All @@ -1180,7 +1182,7 @@ Pipes MergeTreeDataSelectExecutor::spreadMarkRangesAmongStreamsFinal(

case MergeTreeData::MergingParams::Summing:
return std::make_shared<SummingSortedTransform>(header, pipes.size(),
sort_description, data.merging_params.columns_to_sum, max_block_size);
sort_description, data.merging_params.columns_to_sum, partition_key_columns, max_block_size);

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We also need a test with SELECT FINAL?


case MergeTreeData::MergingParams::Aggregating:
return std::make_shared<AggregatingSortedTransform>(header, pipes.size(),
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
0 17
1 17
2 17
3 17
4 17
5 17
6 17
7 17
8 17
9 17
10 17
11 17
12 17
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
DROP TABLE IF EXISTS tt_01373;

CREATE TABLE tt_01373
(a Int64, d Int64, val Int64)
ENGINE = SummingMergeTree PARTITION BY (a) ORDER BY (d);

INSERT INTO tt_01373 SELECT number%13, number%17, 1 from numbers(1000000);

OPTIMIZE TABLE tt_01373 FINAL;

SELECT a, count() FROM tt_01373 GROUP BY a ORDER BY a;

DROP TABLE IF EXISTS tt_01373;