feat(flow): render reduce (#3769)

* feat: render reduce * fix: distinct input dedup&change per review * chore: typos * fix: eval_distinct_core&per review * chore: typos * docs: more comment on accums internal * chore: per review * chore: rebase to main
GreptimeTeam · Apr 25, 2024 · 2d0f493 · 2d0f493
1 parent bba3108
commit 2d0f493
Show file tree

Hide file tree

Showing 2 changed files with 1,081 additions and 7 deletions.
diff --git a/src/flow/src/compute/render.rs b/src/flow/src/compute/render.rs
@@ -44,6 +44,7 @@ use crate::repr::{self, DiffRow, KeyValDiffRow, Row};
 use crate::utils::{ArrangeHandler, ArrangeReader, ArrangeWriter, Arrangement};
 
 mod map;
+mod reduce;
 
 /// The Context for build a Operator with id of `GlobalId`
 pub struct Context<'referred, 'df> {
@@ -104,13 +105,10 @@ impl<'referred, 'df> Context<'referred, 'df> {
             Plan::Let { id, value, body } => self.eval_let(id, value, body),
             Plan::Mfp { input, mfp } => self.render_mfp(input, mfp),
             Plan::Reduce {
-                input: _,
-                key_val_plan: _,
-                reduce_plan: _,
-            } => NotImplementedSnafu {
-                reason: "Reduce is still WIP".to_string(),
-            }
-            .fail(),
+                input,
+                key_val_plan,
+                reduce_plan,
+            } => self.render_reduce(input, key_val_plan, reduce_plan),
             Plan::Join { .. } => NotImplementedSnafu {
                 reason: "Join is still WIP".to_string(),
             }