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

Fix bug in execution of TTL GROUP BY

Browse Source
This commit is contained in:
alesapin 2021-06-27 19:18:15 +03:00
parent bf85b09a16
commit c977c33d6d
9 changed files with 174 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
src/DataStreams/TTLAggregationAlgorithm.cpp
View File

@ -36,24 +36,32 @@ TTLAggregationAlgorithm::TTLAggregationAlgorithm(
storage_.getContext()->getTemporaryVolume(), settings.max_threads, settings.min_free_disk_space_for_temporary_data);
aggregator = std::make_unique<Aggregator>(params);
if (isMinTTLExpired())
new_ttl_info.finished = true;
}
void TTLAggregationAlgorithm::execute(Block & block)
{
if (!block)
{
if (!aggregation_result.empty())
{
MutableColumns result_columns = header.cloneEmptyColumns();
finalizeAggregates(result_columns);
block = header.cloneWithColumns(std::move(result_columns));
}
bool some_rows_were_aggregated = false;
MutableColumns result_columns = header.cloneEmptyColumns();
if (!block) /// Empty block -- no more data, but we may still have some accumulated rows
{
if (!aggregation_result.empty()) /// Still have some aggregated data, let's update TTL
{
finalizeAggregates(result_columns);
some_rows_were_aggregated = true;
}
else /// No block, all aggregated, just finish
{
return;
}
}
else
{
const auto & column_names = header.getNames();
MutableColumns result_columns = header.cloneEmptyColumns();
MutableColumns aggregate_columns = header.cloneEmptyColumns();
auto ttl_column = executeExpressionAndGetColumn(description.expression, block, description.result_column);
@ -84,7 +92,10 @@ void TTLAggregationAlgorithm::execute(Block & block)
if (!same_as_current)
{
if (rows_with_current_key)
{
some_rows_were_aggregated = true;
calculateAggregates(aggregate_columns, current_key_start, rows_with_current_key);
}
finalizeAggregates(result_columns);
current_key_start = rows_aggregated;
@ -104,7 +115,6 @@ void TTLAggregationAlgorithm::execute(Block & block)
}
else
{
new_ttl_info.update(cur_ttl);
for (const auto & name : column_names)
{
const IColumn * values_column = block.getByName(name).column.get();
@ -115,9 +125,21 @@ void TTLAggregationAlgorithm::execute(Block & block)
}
if (rows_with_current_key)
{
some_rows_were_aggregated = true;
calculateAggregates(aggregate_columns, current_key_start, rows_with_current_key);
}
}
block = header.cloneWithColumns(std::move(result_columns));
/// If some rows were aggregated we have to recalculate ttl info's
if (some_rows_were_aggregated)
{
auto ttl_column_after_aggregation = executeExpressionAndGetColumn(description.expression, block, description.result_column);
for (size_t i = 0; i < block.rows(); ++i)
new_ttl_info.update(getTimestampByIndex(ttl_column_after_aggregation.get(), i));
}
}
void TTLAggregationAlgorithm::calculateAggregates(const MutableColumns & aggregate_columns, size_t start_pos, size_t length)
@ -133,6 +155,7 @@ void TTLAggregationAlgorithm::calculateAggregates(const MutableColumns & aggrega
aggregator->executeOnBlock(aggregate_chunk, length, aggregation_result, key_columns,
columns_for_aggregator, no_more_keys);
}
void TTLAggregationAlgorithm::finalizeAggregates(MutableColumns & result_columns)
@ -140,6 +163,7 @@ void TTLAggregationAlgorithm::finalizeAggregates(MutableColumns & result_columns
if (!aggregation_result.empty())
{
auto aggregated_res = aggregator->convertToBlocks(aggregation_result, true, 1);
for (auto & agg_block : aggregated_res)
{
for (const auto & it : description.set_parts)

2
src/DataStreams/TTLColumnAlgorithm.cpp
View File

@ -21,6 +21,8 @@ TTLColumnAlgorithm::TTLColumnAlgorithm(
new_ttl_info = old_ttl_info;
is_fully_empty = false;
}
else
new_ttl_info.finished = true;
}
void TTLColumnAlgorithm::execute(Block & block)

2
src/DataStreams/TTLDeleteAlgorithm.cpp
View File

@ -9,6 +9,8 @@ TTLDeleteAlgorithm::TTLDeleteAlgorithm(
{
if (!isMinTTLExpired())
new_ttl_info = old_ttl_info;
else
new_ttl_info.finished = true;
}
void TTLDeleteAlgorithm::execute(Block & block)

51
src/Storages/MergeTree/MergeTreeDataPartTTLInfo.cpp
View File

@ -55,6 +55,10 @@ void MergeTreeDataPartTTLInfos::read(ReadBuffer & in)
MergeTreeDataPartTTLInfo ttl_info;
ttl_info.min = col["min"].getUInt();
ttl_info.max = col["max"].getUInt();
if (col.has("finished"))
ttl_info.finished = col["finished"].getUInt();
String name = col["name"].getString();
columns_ttl.emplace(name, ttl_info);
@ -67,6 +71,9 @@ void MergeTreeDataPartTTLInfos::read(ReadBuffer & in)
table_ttl.min = table["min"].getUInt();
table_ttl.max = table["max"].getUInt();
if (table.has("finished"))
table_ttl.finished = table["finished"].getUInt();
updatePartMinMaxTTL(table_ttl.min, table_ttl.max);
}
@ -77,6 +84,10 @@ void MergeTreeDataPartTTLInfos::read(ReadBuffer & in)
MergeTreeDataPartTTLInfo ttl_info;
ttl_info.min = elem["min"].getUInt();
ttl_info.max = elem["max"].getUInt();
if (elem.has("finished"))
ttl_info.finished = elem["finished"].getUInt();
String expression = elem["expression"].getString();
ttl_info_map.emplace(expression, ttl_info);
@ -126,6 +137,8 @@ void MergeTreeDataPartTTLInfos::write(WriteBuffer & out) const
writeIntText(it->second.min, out);
writeString(",\"max\":", out);
writeIntText(it->second.max, out);
writeString(R"(,"finished":)", out);
writeIntText(static_cast<uint8_t>(it->second.finished), out);
writeString("}", out);
}
writeString("]", out);
@ -138,6 +151,8 @@ void MergeTreeDataPartTTLInfos::write(WriteBuffer & out) const
writeIntText(table_ttl.min, out);
writeString(R"(,"max":)", out);
writeIntText(table_ttl.max, out);
writeString(R"(,"finished":)", out);
writeIntText(static_cast<uint8_t>(table_ttl.finished), out);
writeString("}", out);
}
@ -159,6 +174,8 @@ void MergeTreeDataPartTTLInfos::write(WriteBuffer & out) const
writeIntText(it->second.min, out);
writeString(R"(,"max":)", out);
writeIntText(it->second.max, out);
writeString(R"(,"finished":)", out);
writeIntText(static_cast<uint8_t>(it->second.finished), out);
writeString("}", out);
}
writeString("]", out);
@ -202,6 +219,39 @@ time_t MergeTreeDataPartTTLInfos::getMinimalMaxRecompressionTTL() const
return max;
}
bool MergeTreeDataPartTTLInfos::hasAnyNonFinishedTTLs() const
{
auto has_non_finished_ttl = [] (const TTLInfoMap & map) -> bool
{
for (const auto & [name, info] : map)
{
if (!info.finished)
return true;
}
return false;
};
if (!table_ttl.finished)
return true;
if (has_non_finished_ttl(columns_ttl))
return true;
if (has_non_finished_ttl(rows_where_ttl))
return true;
if (has_non_finished_ttl(moves_ttl))
return true;
if (has_non_finished_ttl(recompression_ttl))
return true;
if (has_non_finished_ttl(group_by_ttl))
return true;
return false;
}
std::optional<TTLDescription> selectTTLDescriptionForTTLInfos(const TTLDescriptions & descriptions, const TTLInfoMap & ttl_info_map, time_t current_time, bool use_max)
{
time_t best_ttl_time = 0;
@ -232,4 +282,5 @@ std::optional<TTLDescription> selectTTLDescriptionForTTLInfos(const TTLDescripti
return best_ttl_time ? *best_entry_it : std::optional<TTLDescription>();
}
}

9
src/Storages/MergeTree/MergeTreeDataPartTTLInfo.h
View File

@ -14,6 +14,11 @@ struct MergeTreeDataPartTTLInfo
time_t min = 0;
time_t max = 0;
/// This TTL was computed on completely expired part. It doesn't make sense
/// to select such parts for TTL again. But make sense to recalcuate TTL
/// again for merge with multiple parts.
bool finished = false;
void update(time_t time)
{
if (time && (!min || time < min))
@ -28,6 +33,7 @@ struct MergeTreeDataPartTTLInfo
min = other_info.min;
max = std::max(other_info.max, max);
finished &= other_info.finished;
}
};
@ -60,6 +66,9 @@ struct MergeTreeDataPartTTLInfos
void write(WriteBuffer & out) const;
void update(const MergeTreeDataPartTTLInfos & other_infos);
/// Has any TTLs which are not calculated on completely expired parts.
bool hasAnyNonFinishedTTLs() const;
void updatePartMinMaxTTL(time_t time_min, time_t time_max)
{
if (time_min && (!part_min_ttl || time_min < part_min_ttl))

3
src/Storages/MergeTree/TTLMergeSelector.cpp
View File

@ -111,6 +111,9 @@ bool TTLDeleteMergeSelector::isTTLAlreadySatisfied(const IMergeSelector::Part &
if (only_drop_parts)
return false;
if (!part.ttl_infos->hasAnyNonFinishedTTLs())
return false;
return !part.shall_participate_in_merges;
}

4
src/Storages/StorageReplicatedMergeTree.cpp
View File

@ -551,6 +551,10 @@ void StorageReplicatedMergeTree::waitMutationToFinishOnReplicas(
break;
}
/// This replica inactive, don't check anything
if (!inactive_replicas.empty() && inactive_replicas.count(replica))
break;
/// It maybe already removed from zk, but local in-memory mutations
/// state was not updated.
if (!getZooKeeper()->exists(fs::path(zookeeper_path) / "mutations" / mutation_id))

13
tests/integration/test_ttl_replicated/test.py
View File

@ -351,6 +351,7 @@ def test_ttl_compatibility(started_cluster, node_left, node_right, num_run):
ENGINE = ReplicatedMergeTree('/clickhouse/tables/test/test_ttl_delete_{suff}', '{replica}')
ORDER BY id PARTITION BY toDayOfMonth(date)
TTL date + INTERVAL 3 SECOND
SETTINGS max_number_of_merges_with_ttl_in_pool=100, max_replicated_merges_with_ttl_in_queue=100
'''.format(suff=num_run, replica=node.name))
node.query(
@ -359,6 +360,7 @@ def test_ttl_compatibility(started_cluster, node_left, node_right, num_run):
ENGINE = ReplicatedMergeTree('/clickhouse/tables/test/test_ttl_group_by_{suff}', '{replica}')
ORDER BY id PARTITION BY toDayOfMonth(date)
TTL date + INTERVAL 3 SECOND GROUP BY id SET val = sum(val)
SETTINGS max_number_of_merges_with_ttl_in_pool=100, max_replicated_merges_with_ttl_in_queue=100
'''.format(suff=num_run, replica=node.name))
node.query(
@ -367,6 +369,7 @@ def test_ttl_compatibility(started_cluster, node_left, node_right, num_run):
ENGINE = ReplicatedMergeTree('/clickhouse/tables/test/test_ttl_where_{suff}', '{replica}')
ORDER BY id PARTITION BY toDayOfMonth(date)
TTL date + INTERVAL 3 SECOND DELETE WHERE id % 2 = 1
SETTINGS max_number_of_merges_with_ttl_in_pool=100, max_replicated_merges_with_ttl_in_queue=100
'''.format(suff=num_run, replica=node.name))
node_left.query("INSERT INTO test_ttl_delete VALUES (now(), 1)")
@ -397,9 +400,9 @@ def test_ttl_compatibility(started_cluster, node_left, node_right, num_run):
node_right.query("OPTIMIZE TABLE test_ttl_group_by FINAL")
node_right.query("OPTIMIZE TABLE test_ttl_where FINAL")
exec_query_with_retry(node_left, "SYSTEM SYNC REPLICA test_ttl_delete")
node_left.query("SYSTEM SYNC REPLICA test_ttl_group_by", timeout=20)
node_left.query("SYSTEM SYNC REPLICA test_ttl_where", timeout=20)
exec_query_with_retry(node_left, "OPTIMIZE TABLE test_ttl_delete FINAL")
node_left.query("OPTIMIZE TABLE test_ttl_group_by FINAL", timeout=20)
node_left.query("OPTIMIZE TABLE test_ttl_where FINAL", timeout=20)
# After OPTIMIZE TABLE, it is not guaranteed that everything is merged.
# Possible scenario (for test_ttl_group_by):
@ -414,6 +417,10 @@ def test_ttl_compatibility(started_cluster, node_left, node_right, num_run):
node_right.query("SYSTEM SYNC REPLICA test_ttl_group_by", timeout=20)
node_right.query("SYSTEM SYNC REPLICA test_ttl_where", timeout=20)
exec_query_with_retry(node_left, "SYSTEM SYNC REPLICA test_ttl_delete")
node_left.query("SYSTEM SYNC REPLICA test_ttl_group_by", timeout=20)
node_left.query("SYSTEM SYNC REPLICA test_ttl_where", timeout=20)
assert node_left.query("SELECT id FROM test_ttl_delete ORDER BY id") == "2\n4\n"
assert node_right.query("SELECT id FROM test_ttl_delete ORDER BY id") == "2\n4\n"

5
tests/integration/test_version_update_after_mutation/test.py
View File

@ -79,7 +79,10 @@ def test_upgrade_while_mutation(start_cluster):
node3.restart_with_latest_version(signal=9)
exec_query_with_retry(node3, "ALTER TABLE mt1 DELETE WHERE id > 100000", settings={"mutations_sync": "2"})
# wait replica became active
exec_query_with_retry(node3, "SYSTEM RESTART REPLICA mt1")
node3.query("ALTER TABLE mt1 DELETE WHERE id > 100000", settings={"mutations_sync": "2"})
# will delete nothing, but previous async mutation will finish with this query
assert_eq_with_retry(node3, "SELECT COUNT() from mt1", "50000\n")