@@ -236,33 +236,45 @@ mod tests {
236236 } ;
237237
238238 use crate :: analytical_cost:: { ExecutionMultiplicity , PhysicalDagNode , PhysicalOperator } ;
239- use crate :: analytical_statistics:: { EdgeStatistics , OperatorStatistics , SourceCoverage } ;
239+ use crate :: analytical_statistics:: {
240+ EdgeStatistics , OperatorStatistics , SourceCoverage , UnaryEdgeStatistics ,
241+ } ;
240242 use crate :: replacement:: ReplacementStrategy ;
241243
242244 fn edge ( rows : u64 , bytes : u64 ) -> EdgeStatistics {
243245 EdgeStatistics { rows, bytes }
244246 }
245247
246- fn statistics (
247- source_scan_bytes : u64 ,
248- inputs : Vec < EdgeStatistics > ,
249- output : EdgeStatistics ,
250- ) -> OperatorStatistics {
251- OperatorStatistics {
252- source_scan_bytes,
253- inputs,
248+ fn unary_edges ( input : EdgeStatistics , output : EdgeStatistics ) -> UnaryEdgeStatistics {
249+ UnaryEdgeStatistics {
250+ input,
254251 output,
255- group_count : None ,
256- key_bytes : None ,
257- aggregate_value_bytes : None ,
258- k : None ,
259- topk_output_offset : None ,
260- limit_rows_consumed : None ,
261- hash_join_build_side : None ,
262252 promql : None ,
263253 }
264254 }
265255
256+ fn scan_statistics ( source_read_bytes : u64 , edge : EdgeStatistics ) -> OperatorStatistics {
257+ OperatorStatistics :: Scan {
258+ edges : unary_edges ( edge, edge) ,
259+ source_read_bytes,
260+ }
261+ }
262+
263+ fn aggregate_statistics ( input : EdgeStatistics , output : EdgeStatistics ) -> OperatorStatistics {
264+ OperatorStatistics :: HashAggregate {
265+ edges : unary_edges ( input, output) ,
266+ group_count : 1 ,
267+ key_bytes : 0 ,
268+ accumulator_bytes_per_group : 8 ,
269+ }
270+ }
271+
272+ fn pass_through_statistics ( edge : EdgeStatistics ) -> OperatorStatistics {
273+ OperatorStatistics :: PassThrough {
274+ edges : unary_edges ( edge, edge) ,
275+ }
276+ }
277+
266278 fn query ( ) -> Rc < QueryExpr > {
267279 Rc :: new ( QueryExpr :: Aggregate {
268280 reduction : Reduction :: by ( vec ! [ ] ) ,
@@ -299,8 +311,9 @@ mod tests {
299311 source: Source :: Table {
300312 table_ref: "events" . into( ) ,
301313 } ,
302- snapshot_id : "snapshot-1" . into( ) ,
314+ source_snapshot_id : "snapshot-1" . into( ) ,
303315 predicates: vec![ ] ,
316+ info_matchers: vec![ ] ,
304317 } ] ,
305318 }
306319 }
@@ -323,16 +336,9 @@ mod tests {
323336 }
324337
325338 fn summary_dag ( & self , scope : & ComparisonScope ) -> PhysicalDag {
326- let scan_statistics = statistics (
327- self . candidate_scan_bytes ,
328- vec ! [ edge( 100 , 800 ) ] ,
329- edge ( 100 , 800 ) ,
330- ) ;
331- let mut aggregate_statistics = statistics ( 0 , vec ! [ edge( 100 , 800 ) ] , edge ( 1 , 8 ) ) ;
332- aggregate_statistics. group_count = Some ( 1 ) ;
333- aggregate_statistics. key_bytes = Some ( 8 ) ;
334- aggregate_statistics. aggregate_value_bytes = Some ( 8 ) ;
335- let read_statistics = statistics ( 0 , vec ! [ edge( 1 , 8 ) ] , edge ( 1 , 8 ) ) ;
339+ let scan_statistics = scan_statistics ( self . candidate_scan_bytes , edge ( 100 , 800 ) ) ;
340+ let aggregate_statistics = aggregate_statistics ( edge ( 100 , 800 ) , edge ( 1 , 8 ) ) ;
341+ let read_statistics = pass_through_statistics ( edge ( 1 , 8 ) ) ;
336342 let evidence = HashMap :: from ( [
337343 (
338344 "candidate-scan" . into ( ) ,
@@ -372,7 +378,10 @@ mod tests {
372378 } ,
373379 PhysicalDagNode {
374380 id: "candidate-state" . into( ) ,
375- operator: PhysicalOperator :: HashAggregate ,
381+ operator: PhysicalOperator :: HashAggregate {
382+ grouping_key_count: 0 ,
383+ accumulator_count: 1 ,
384+ } ,
376385 children: vec![ "candidate-scan" . into( ) ] ,
377386 source_coverage: None ,
378387 output_buffer_bytes: 8 ,
@@ -415,22 +424,17 @@ mod tests {
415424 assert_eq ! ( snapshot. version, "test-snapshot-1" ) ;
416425 self . raw_evidence_calls
417426 . set ( self . raw_evidence_calls . get ( ) + 1 ) ;
418- let ( physical_id, mut statistics) = match request. operator {
419- PhysicalOperator :: Scan => (
420- "raw-scan" ,
421- statistics ( 800 , vec ! [ edge ( 100 , 800 ) ] , edge ( 100 , 800 ) ) ,
422- ) ,
423- PhysicalOperator :: HashAggregate => (
427+ let ( physical_id, statistics) = match request. operator {
428+ PhysicalOperator :: Scan => ( "raw-scan" , scan_statistics ( 800 , edge ( 100 , 800 ) ) ) ,
429+ PhysicalOperator :: HashAggregate {
430+ grouping_key_count : 0 ,
431+ accumulator_count : 1 ,
432+ } => (
424433 "raw-aggregate" ,
425- statistics ( 0 , vec ! [ edge( 100 , 800 ) ] , edge ( 1 , 8 ) ) ,
434+ aggregate_statistics ( edge ( 100 , 800 ) , edge ( 1 , 8 ) ) ,
426435 ) ,
427436 _ => return Err ( AnalyticalCostError :: UnsupportedQueryOperator ) ,
428437 } ;
429- if request. operator == PhysicalOperator :: HashAggregate {
430- statistics. group_count = Some ( 1 ) ;
431- statistics. key_bytes = Some ( 8 ) ;
432- statistics. aggregate_value_bytes = Some ( 8 ) ;
433- }
434438 Ok ( PhysicalNodeEvidence {
435439 physical_id : physical_id. into ( ) ,
436440 output_buffer_bytes : 8 ,
@@ -522,7 +526,11 @@ mod tests {
522526 target : & TargetSubDAG < ' _ > ,
523527 ) -> Result < PhysicalDag , AnalyticalCostError > {
524528 let mut dag = self . 0 . summary_physical_dag ( snapshot, summary, target) ?;
525- dag. nodes [ 0 ] . source_coverage . as_mut ( ) . unwrap ( ) . snapshot_id = "other" . into ( ) ;
529+ dag. nodes [ 0 ]
530+ . source_coverage
531+ . as_mut ( )
532+ . unwrap ( )
533+ . source_snapshot_id = "other" . into ( ) ;
526534 Ok ( dag)
527535 }
528536 }
0 commit comments