From c8401bd521b8f622de971939cbe559752355ede3 Mon Sep 17 00:00:00 2001
From: Noel Kwan <noelkwan1998@gmail.com>
Date: Fri, 9 Aug 2024 00:25:26 +0800
Subject: [PATCH 1/4] fix row count + add test

---
 ...ase_approx_percentile_merge_normal_agg.slt | 93 +++++++++++++++++++
 src/stream/src/executor/row_merge.rs          | 16 +++-
 2 files changed, 105 insertions(+), 4 deletions(-)
 create mode 100644 e2e_test/streaming/aggregate/two_phase_approx_percentile_merge_normal_agg.slt

diff --git a/e2e_test/streaming/aggregate/two_phase_approx_percentile_merge_normal_agg.slt b/e2e_test/streaming/aggregate/two_phase_approx_percentile_merge_normal_agg.slt
new file mode 100644
index 0000000000000..299fa02e5bdfd
--- /dev/null
+++ b/e2e_test/streaming/aggregate/two_phase_approx_percentile_merge_normal_agg.slt
@@ -0,0 +1,93 @@
+# Single phase approx percentile
+statement ok
+create table t(p_col double, grp_col int);
+
+statement ok
+insert into t select a, 1 from generate_series(-1000, 1000) t(a);
+
+statement ok
+flush;
+
+query I
+select
+  percentile_cont(0.01) within group (order by p_col) as p01,
+  sum(p_col),
+  percentile_cont(0.5) within group (order by p_col) as p50,
+  count(*),
+  percentile_cont(0.99) within group (order by p_col) as p99
+from t;
+----
+-980 0 0 2001 980
+
+statement ok
+create materialized view m1 as
+ select
+     approx_percentile(0.01, 0.01) within group (order by p_col) as p01,
+     sum(p_col),
+     approx_percentile(0.5, 0.01) within group (order by p_col) as p50,
+     count(*),
+     approx_percentile(0.99, 0.01) within group (order by p_col) as p99
+ from t;
+
+query I
+select * from m1;
+----
+-982.5779489474152 0 0 2001 982.5779489474152
+
+# Test state encode / decode
+onlyif can-use-recover
+statement ok
+recover;
+
+onlyif can-use-recover
+sleep 10s
+
+query I
+select * from m1;
+----
+-982.5779489474152 0 0 2001 982.5779489474152
+
+# Test state encode / decode
+onlyif can-use-recover
+statement ok
+recover;
+
+onlyif can-use-recover
+sleep 10s
+
+query I
+select * from m1;
+----
+-982.5779489474152 0 0 2001 982.5779489474152
+
+# Test 0<x<1 values
+statement ok
+insert into t select 0.001, 1 from generate_series(1, 500);
+
+statement ok
+insert into t select 0.0001, 1 from generate_series(1, 501);
+
+statement ok
+flush;
+
+query I
+select * from m1;
+----
+-963.1209598593477 0.5501000000000007 0.00009999833511933609 3002 963.1209598593477
+
+query I
+select
+  percentile_cont(0.01) within group (order by p_col) as p01,
+  sum(p_col),
+  percentile_cont(0.5) within group (order by p_col) as p50,
+  count(*),
+  percentile_cont(0.99) within group (order by p_col) as p99
+from t;
+----
+-969.99 0.5500999999513363 0.0001 3002 969.9899999999998
+
+statement ok
+drop materialized view m1;
+
+statement ok
+drop table t;
\ No newline at end of file
diff --git a/src/stream/src/executor/row_merge.rs b/src/stream/src/executor/row_merge.rs
index b19b9bdcca8c8..326246cce6abf 100644
--- a/src/stream/src/executor/row_merge.rs
+++ b/src/stream/src/executor/row_merge.rs
@@ -161,15 +161,23 @@ impl RowMergeExecutor {
         for (i, (op, lhs_row)) in lhs_chunk.rows().enumerate() {
             ops.push(op);
             for (j, d) in lhs_row.iter().enumerate() {
-                let out_index = lhs_mapping.map(j);
-                merged_rows[i][out_index] = d.to_owned_datum();
+                // NOTE(kwannoel): Unnecessary columns will not have a mapping,
+                // for instance extra row count column.
+                // those can be skipped here.
+                if let Some(out_index) = lhs_mapping.try_map(j) {
+                    merged_rows[i][out_index] = d.to_owned_datum();
+                }
             }
         }
 
         for (i, (_, rhs_row)) in rhs_chunk.rows().enumerate() {
             for (j, d) in rhs_row.iter().enumerate() {
-                let out_index = rhs_mapping.map(j);
-                merged_rows[i][out_index] = d.to_owned_datum();
+                // NOTE(kwannoel): Unnecessary columns will not have a mapping,
+                // for instance extra row count column.
+                // those can be skipped here.
+                if let Some(out_index) = rhs_mapping.try_map(j) {
+                    merged_rows[i][out_index] = d.to_owned_datum();
+                }
             }
         }
         let mut builder = DataChunkBuilder::new(data_types.to_vec(), cardinality);

From df3549d64eb890f806b16805504e52af5580b9ae Mon Sep 17 00:00:00 2001
From: Noel Kwan <noelkwan1998@gmail.com>
Date: Fri, 9 Aug 2024 10:13:16 +0800
Subject: [PATCH 2/4] fix

---
 .../two_phase_approx_percentile_merge_normal_agg.slt          | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/e2e_test/streaming/aggregate/two_phase_approx_percentile_merge_normal_agg.slt b/e2e_test/streaming/aggregate/two_phase_approx_percentile_merge_normal_agg.slt
index 299fa02e5bdfd..423a24530b757 100644
--- a/e2e_test/streaming/aggregate/two_phase_approx_percentile_merge_normal_agg.slt
+++ b/e2e_test/streaming/aggregate/two_phase_approx_percentile_merge_normal_agg.slt
@@ -78,13 +78,13 @@ select * from m1;
 query I
 select
   percentile_cont(0.01) within group (order by p_col) as p01,
-  sum(p_col),
+  round(sum(p_col), 2),
   percentile_cont(0.5) within group (order by p_col) as p50,
   count(*),
   percentile_cont(0.99) within group (order by p_col) as p99
 from t;
 ----
--969.99 0.5500999999513363 0.0001 3002 969.9899999999998
+-969.99 0.55 0.0001 3002 969.9899999999998
 
 statement ok
 drop materialized view m1;

From 8183af86f81e873322ef3c1c2bf28ca48f79fd9f Mon Sep 17 00:00:00 2001
From: Noel Kwan <noelkwan1998@gmail.com>
Date: Fri, 9 Aug 2024 10:31:55 +0800
Subject: [PATCH 3/4] use round digit

---
 .../aggregate/two_phase_approx_percentile_merge_normal_agg.slt  | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/e2e_test/streaming/aggregate/two_phase_approx_percentile_merge_normal_agg.slt b/e2e_test/streaming/aggregate/two_phase_approx_percentile_merge_normal_agg.slt
index 423a24530b757..93063e2bde689 100644
--- a/e2e_test/streaming/aggregate/two_phase_approx_percentile_merge_normal_agg.slt
+++ b/e2e_test/streaming/aggregate/two_phase_approx_percentile_merge_normal_agg.slt
@@ -78,7 +78,7 @@ select * from m1;
 query I
 select
   percentile_cont(0.01) within group (order by p_col) as p01,
-  round(sum(p_col), 2),
+  round_digit(sum(p_col), 2),
   percentile_cont(0.5) within group (order by p_col) as p50,
   count(*),
   percentile_cont(0.99) within group (order by p_col) as p99

From ff45bd91b2e8f9d8cd7c75e67656938e50d783c7 Mon Sep 17 00:00:00 2001
From: Noel Kwan <noelkwan1998@gmail.com>
Date: Fri, 9 Aug 2024 10:59:40 +0800
Subject: [PATCH 4/4] fix

---
 .../two_phase_approx_percentile_merge_normal_agg.slt          | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/e2e_test/streaming/aggregate/two_phase_approx_percentile_merge_normal_agg.slt b/e2e_test/streaming/aggregate/two_phase_approx_percentile_merge_normal_agg.slt
index 93063e2bde689..959e3684a9fd2 100644
--- a/e2e_test/streaming/aggregate/two_phase_approx_percentile_merge_normal_agg.slt
+++ b/e2e_test/streaming/aggregate/two_phase_approx_percentile_merge_normal_agg.slt
@@ -78,13 +78,13 @@ select * from m1;
 query I
 select
   percentile_cont(0.01) within group (order by p_col) as p01,
-  round_digit(sum(p_col), 2),
+  round(sum(p_col) * 100),
   percentile_cont(0.5) within group (order by p_col) as p50,
   count(*),
   percentile_cont(0.99) within group (order by p_col) as p99
 from t;
 ----
--969.99 0.55 0.0001 3002 969.9899999999998
+-969.99 55 0.0001 3002 969.9899999999998
 
 statement ok
 drop materialized view m1;