Influxdata
/
influxdb
mirror of https://github.com/influxdata/influxdb.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

test: Call the smaller inner hot_partitions_for_shard when only one shard is involved

pull/24376/head
Carol (Nichols || Goulding) 2022-09-16 10:33:51 -04:00
parent fcf9a9d589
commit 316ebfa8c1
No known key found for this signature in database
GPG Key ID: E907EE5A736F87D4
1 changed files with 38 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
compactor/src/hot.rs
View File

@ -643,12 +643,20 @@ mod tests {
.with_creation_time(hot_time_one_hour_ago);
partition6.create_parquet_file_catalog_record(pf6_2).await;
let query_times = query_times(compactor.catalog.time_provider());
// partition candidates: partitions with L0 and overlapped L1
let mut candidates = hot_partitions_to_compact(Arc::clone(&compactor))
.await
.unwrap();
let mut candidates = hot_partitions_for_shard(
Arc::clone(&compactor.catalog),
shard.shard.id,
&query_times,
1,
100,
)
.await
.unwrap();
assert_eq!(candidates.len(), 6);
candidates.sort_by_key(|c| c.candidate.partition_id);
candidates.sort_by_key(|c| c.partition_id);
{
let mut repos = compactor.catalog.repositories().await;
let skipped_compactions = repos.partitions().list_skipped_compactions().await.unwrap();
@ -658,6 +666,12 @@ mod tests {
);
}
let table_columns = compactor.table_columns(&candidates).await.unwrap();
let candidates = compactor
.add_info_to_partitions(&candidates, &table_columns)
.await
.unwrap();
// There are 3 rounds of parallel compaction:
//
// * Round 1: 3 candidates [P1, P2, P5] and total needed budget 13,500
@ -891,13 +905,30 @@ mod tests {
let count = catalog.count_level_0_files(shard.shard.id).await;
assert_eq!(count, 4);
let query_times = query_times(compactor.catalog.time_provider());
// ------------------------------------------------
// Compact
let mut partition_candidates = hot_partitions_to_compact(Arc::clone(&compactor))
.await
.unwrap();
let partition_candidates = hot_partitions_for_shard(
Arc::clone(&compactor.catalog),
shard.shard.id,
&query_times,
1,
1,
)
.await
.unwrap();
assert_eq!(partition_candidates.len(), 1);
let table_columns = compactor
.table_columns(&partition_candidates)
.await
.unwrap();
let mut partition_candidates = compactor
.add_info_to_partitions(&partition_candidates, &table_columns)
.await
.unwrap();
let partition = partition_candidates.pop().unwrap();
let parquet_files_for_compaction =