Skip to content

Commit 6064b76

Browse files
KKouldKKould
committed
feat(query): refresh topn statistics on append
1 parent fcf7671 commit 6064b76

35 files changed

Lines changed: 1154 additions & 237 deletions

‎src/query/ee/src/storages/fuse/operations/virtual_columns.rs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -318,6 +318,7 @@ pub async fn commit_refresh_virtual_column(
318318
block_meta: Arc::unwrap_or_clone(block_meta.clone()),
319319
draft_virtual_block_meta: Some(result.draft_virtual_block_meta.clone()),
320320
column_hlls: result.column_hlls.clone().map(BlockHLLState::Serialized),
321+
column_top_n: None,
321322
};
322323
let entry = MutationLogEntry::ReplacedBlock {
323324
index: BlockMetaIndex {
@@ -551,6 +552,7 @@ async fn prepare_vacuum_virtual_column_mutations(
551552
block_meta: new_block_meta,
552553
draft_virtual_block_meta: None,
553554
column_hlls: column_hlls.map(BlockHLLState::Serialized),
555+
column_top_n: None,
554556
}),
555557
});
556558

‎src/query/expression/src/values.rs‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -116,6 +116,8 @@ use crate::with_opaque_size;
116116
use crate::with_opaque_size_mapped;
117117
use crate::with_opaque_type;
118118

119+
pub const LARGE_STRING_BYTES_THRESHOLD: usize = 256;
120+
119121
#[derive(Debug, Clone, PartialEq, EnumAsInner)]
120122
pub enum Value<T: AccessType> {
121123
Scalar(T::Scalar),
@@ -1913,7 +1915,7 @@ impl Column {
19131915
}
19141916
}
19151917

1916-
/// Checks if the average length of a string column exceeds 256 bytes.
1918+
/// Checks if the average length of a string column exceeds LARGE_STRING_BYTES_THRESHOLD bytes.
19171919
/// If it does, the bloom index for the column will not be established.
19181920
pub fn check_large_string(&self) -> bool {
19191921
let (inner, len) = if let Column::Nullable(c) = self {
@@ -1923,7 +1925,7 @@ impl Column {
19231925
};
19241926
if let Column::String(v) = inner {
19251927
let bytes_per_row = v.total_bytes_len() / len.max(1);
1926-
if bytes_per_row > 256 {
1928+
if bytes_per_row > LARGE_STRING_BYTES_THRESHOLD {
19271929
return true;
19281930
}
19291931
}

‎src/query/service/src/interpreters/interpreter_table_drop_column.rs‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ use databend_common_sql::binder::validate_table_indexes_not_referencing_columns;
2626
use databend_common_sql::plans::DropTableColumnPlan;
2727
use databend_common_storages_basic::view_table::VIEW_ENGINE;
2828
use databend_common_storages_stream::stream_table::STREAM_ENGINE;
29+
use databend_storages_common_table_meta::table::OPT_KEY_ANALYZE_TOP_N_COLUMNS;
2930
use databend_storages_common_table_meta::table::OPT_KEY_APPROX_DISTINCT_COLUMNS;
3031
use databend_storages_common_table_meta::table::OPT_KEY_BLOOM_INDEX_COLUMNS;
3132

@@ -159,6 +160,16 @@ impl Interpreter for DropTableColumnInterpreter {
159160
}
160161
}
161162
}
163+
if let Some(value) = opts.get_mut(OPT_KEY_ANALYZE_TOP_N_COLUMNS) {
164+
if let ApproxDistinctColumns::Specify(mut cols) =
165+
value.parse::<ApproxDistinctColumns>()?
166+
{
167+
if let Some(pos) = cols.iter().position(|x| *x == self.plan.column) {
168+
cols.remove(pos);
169+
*value = cols.join(",");
170+
}
171+
}
172+
}
162173
let new_schema = new_table_meta.schema.as_ref().clone();
163174

164175
let dropped_column_ids = field.column_ids().into_iter().collect();

‎src/query/service/src/interpreters/interpreter_table_modify_column.rs‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,7 @@ use databend_storages_common_index::RangeIndex;
6161
use databend_storages_common_table_meta::meta::SnapshotId;
6262
use databend_storages_common_table_meta::meta::TableMetaTimestamps;
6363
use databend_storages_common_table_meta::readers::snapshot_reader::TableSnapshotAccessor;
64+
use databend_storages_common_table_meta::table::OPT_KEY_ANALYZE_TOP_N_COLUMNS;
6465
use databend_storages_common_table_meta::table::OPT_KEY_APPROX_DISTINCT_COLUMNS;
6566
use databend_storages_common_table_meta::table::OPT_KEY_BLOOM_INDEX_COLUMNS;
6667

@@ -308,6 +309,12 @@ impl ModifyTableColumnInterpreter {
308309
approx_distinct_cols = cols;
309310
}
310311
}
312+
let mut analyze_top_n_cols = vec![];
313+
if let Some(v) = table_info.options().get(OPT_KEY_ANALYZE_TOP_N_COLUMNS) {
314+
if let ApproxDistinctColumns::Specify(cols) = v.parse::<ApproxDistinctColumns>()? {
315+
analyze_top_n_cols = cols;
316+
}
317+
}
311318

312319
let mut table_info = table.get_table_info().clone();
313320
table_info.meta.fill_field_comments();
@@ -345,6 +352,14 @@ impl ModifyTableColumnInterpreter {
345352
field.data_type
346353
)));
347354
}
355+
if analyze_top_n_cols.iter().any(|v| v.as_str() == field.name)
356+
&& !RangeIndex::supported_table_type(&field.data_type)
357+
{
358+
return Err(ErrorCode::TableOptionInvalid(format!(
359+
"Unsupported data type '{}' for analyze top n columns",
360+
field.data_type
361+
)));
362+
}
348363
// If the column is inverted index column, the type can't be changed.
349364
if !table_info.meta.indexes.is_empty() {
350365
for (index_name, index) in &table_info.meta.indexes {

‎src/query/service/src/interpreters/interpreter_table_rename_column.rs‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ use databend_common_sql::plans::RenameTableColumnPlan;
2323
use databend_common_storages_basic::view_table::VIEW_ENGINE;
2424
use databend_common_storages_iceberg::table::ICEBERG_ENGINE;
2525
use databend_common_storages_stream::stream_table::STREAM_ENGINE;
26+
use databend_storages_common_table_meta::table::OPT_KEY_ANALYZE_TOP_N_COLUMNS;
2627
use databend_storages_common_table_meta::table::OPT_KEY_APPROX_DISTINCT_COLUMNS;
2728
use databend_storages_common_table_meta::table::OPT_KEY_BLOOM_INDEX_COLUMNS;
2829

@@ -129,6 +130,14 @@ impl Interpreter for RenameTableColumnInterpreter {
129130
&self.plan.new_column,
130131
)?;
131132
}
133+
if let Some(value) = opts.get_mut(OPT_KEY_ANALYZE_TOP_N_COLUMNS) {
134+
rename_column_in_comma_separated_ident(
135+
self.ctx.as_ref(),
136+
value,
137+
&self.plan.old_column,
138+
&self.plan.new_column,
139+
)?;
140+
}
132141

133142
let mut new_cluster_key = None;
134143
if let Some((_, cluster_key)) = table.cluster_key_meta() {

‎src/query/service/src/physical_plans/physical_commit_sink.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -179,6 +179,7 @@ impl IPhysicalPlan for CommitSink {
179179
block_meta: Arc::unwrap_or_clone(block_meta),
180180
draft_virtual_block_meta: None,
181181
column_hlls: column_hlls.map(BlockHLLState::Serialized),
182+
column_top_n: None,
182183
})
183184
})
184185
.collect::<Vec<Arc<ExtendedBlockMeta>>>();

‎src/query/service/src/physical_plans/physical_mutation_source.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -144,6 +144,7 @@ impl IPhysicalPlan for MutationSource {
144144
virtual_schema: None,
145145
virtual_schema_mode: VirtualSchemaMode::Merge,
146146
hll: HashMap::new(),
147+
top_n: HashMap::new(),
147148
};
148149
let block = DataBlock::empty_with_meta(Box::new(meta));
149150
OneBlockSource::create(output, block)

‎src/query/sql/src/planner/plans/scan.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -546,6 +546,7 @@ mod tests {
546546
fn test_sampled_scan_clears_top_n_stats() -> Result<()> {
547547
let column = Symbol::new(1);
548548
let top_n = ColumnTopN {
549+
capacity: 1,
549550
values: vec![ColumnTopNEntry {
550551
scalar: Scalar::Number(NumberScalar::UInt64(42)),
551552
count: 37,

‎src/query/sql/tests/it/optimizer/collect_statistics.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@ async fn test_collect_statistics_skips_top_n_for_change_scan() -> Result<()> {
3939
ctx.configure_for_optimizer_case(true)?;
4040

4141
let top_n = ColumnTopN {
42+
capacity: 1,
4243
values: vec![ColumnTopNEntry {
4344
scalar: Scalar::Number(NumberScalar::UInt64(1)),
4445
count: 90,

‎src/query/sql/tests/it/optimizer/selectivity.rs‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -284,6 +284,7 @@ fn test_selectivity_comparison_outcomes() -> Result<()> {
284284
histogram: None,
285285
})]);
286286
let top_n = TopNSet::from_iter([(Symbol::new(0), ColumnTopN {
287+
capacity: 1,
287288
values: vec![ColumnTopNEntry {
288289
scalar: Scalar::Number(NumberScalar::UInt64(42)),
289290
count: 37,
@@ -311,6 +312,7 @@ fn test_selectivity_comparison_outcomes() -> Result<()> {
311312
"TopN equality estimates should compose with AND filters.",
312313
)?;
313314
let constrained_top_n = TopNSet::from_iter([(Symbol::new(0), ColumnTopN {
315+
capacity: 2,
314316
values: vec![
315317
ColumnTopNEntry {
316318
scalar: Scalar::Number(NumberScalar::UInt64(1)),
@@ -347,6 +349,7 @@ fn test_selectivity_comparison_outcomes() -> Result<()> {
347349
"Approximate TopN frequencies should use the count upper bound for equality and the lower bound for inequality.",
348350
)?;
349351
let approximate_top_n = TopNSet::from_iter([(Symbol::new(0), ColumnTopN {
352+
capacity: 1,
350353
values: vec![ColumnTopNEntry {
351354
scalar: Scalar::Number(NumberScalar::UInt64(42)),
352355
count: 100,
@@ -380,6 +383,7 @@ fn test_selectivity_comparison_outcomes() -> Result<()> {
380383
histogram: None,
381384
})]);
382385
let fallback_top_n = TopNSet::from_iter([(Symbol::new(0), ColumnTopN {
386+
capacity: 1,
383387
values: vec![ColumnTopNEntry {
384388
scalar: Scalar::Number(NumberScalar::UInt64(42)),
385389
count: 500,
@@ -413,6 +417,7 @@ fn test_selectivity_comparison_outcomes() -> Result<()> {
413417
histogram: None,
414418
})]);
415419
let nullable_top_n = TopNSet::from_iter([(Symbol::new(0), ColumnTopN {
420+
capacity: 1,
416421
values: vec![ColumnTopNEntry {
417422
scalar: Scalar::Number(NumberScalar::UInt64(1)),
418423
count: 10,

0 commit comments

Comments
 (0)