Merge pull request #9090 from sundy-li/fix-agg2

mergify[bot] · web-flow · commit 69eb9d6b58d4 · 2022-12-06T07:25:04.000Z
fix(query): fix state merge bug for aggregation
diff --git a/src/query/functions-v2/src/aggregates/adaptors/aggregate_null_unary_adaptor.rs b/src/query/functions-v2/src/aggregates/adaptors/aggregate_null_unary_adaptor.rs
@@ -97,7 +97,7 @@ impl<const NULLABLE_RESULT: bool> AggregateFunction for AggregateNullUnaryAdapto
     #[inline]
     fn state_layout(&self) -> Layout {
         let layout = self.nested.state_layout();
-        let add = usize::from(NULLABLE_RESULT);
+        let add = if NULLABLE_RESULT { layout.align() } else { 0 };
         Layout::from_size_align(layout.size() + add, layout.align()).unwrap()
     }
 
@@ -182,9 +182,9 @@ impl<const NULLABLE_RESULT: bool> AggregateFunction for AggregateNullUnaryAdapto
 
     fn deserialize(&self, place: StateAddr, reader: &mut &[u8]) -> Result<()> {
         if NULLABLE_RESULT {
+            let flag = reader[reader.len() - 1];
             self.nested
                 .deserialize(place, &mut &reader[..reader.len() - 1])?;
-            let flag = reader[reader.len() - 1];
             self.set_flag(place, flag);
         } else {
             self.nested.deserialize(place, reader)?;
diff --git a/src/query/functions-v2/src/aggregates/adaptors/aggregate_null_variadic_adaptor.rs b/src/query/functions-v2/src/aggregates/adaptors/aggregate_null_variadic_adaptor.rs
@@ -98,7 +98,7 @@ impl<const NULLABLE_RESULT: bool> AggregateFunction
     #[inline]
     fn state_layout(&self) -> Layout {
         let layout = self.nested.state_layout();
-        let add = usize::from(NULLABLE_RESULT);
+        let add = if NULLABLE_RESULT { layout.align() } else { 0 };
         Layout::from_size_align(layout.size() + add, layout.align()).unwrap()
     }
 
@@ -184,9 +184,9 @@ impl<const NULLABLE_RESULT: bool> AggregateFunction
 
     fn deserialize(&self, place: StateAddr, reader: &mut &[u8]) -> Result<()> {
         if NULLABLE_RESULT {
+            let flag = reader[reader.len() - 1];
             self.nested
                 .deserialize(place, &mut &reader[..reader.len() - 1])?;
-            let flag = reader[reader.len() - 1];
             self.set_flag(place, flag);
         } else {
             self.nested.deserialize(place, reader)?;
diff --git a/src/query/functions-v2/src/aggregates/adaptors/aggregate_ornull_adaptor.rs b/src/query/functions-v2/src/aggregates/adaptors/aggregate_ornull_adaptor.rs
@@ -90,7 +90,7 @@ impl AggregateFunction for AggregateFunctionOrNullAdaptor {
     #[inline]
     fn state_layout(&self) -> std::alloc::Layout {
         let layout = self.inner.state_layout();
-        Layout::from_size_align(layout.size() + 1, layout.align()).unwrap()
+        Layout::from_size_align(layout.size() + layout.align(), layout.align()).unwrap()
     }
 
     #[inline]
@@ -174,17 +174,17 @@ impl AggregateFunction for AggregateFunctionOrNullAdaptor {
 
     #[inline]
     fn deserialize(&self, place: StateAddr, reader: &mut &[u8]) -> Result<()> {
+        let flag = reader[reader.len() - 1];
         self.inner
             .deserialize(place, &mut &reader[..reader.len() - 1])?;
-        let flag = reader[reader.len() - 1];
         self.set_flag(place, flag);
         Ok(())
     }
 
     fn merge(&self, place: StateAddr, rhs: StateAddr) -> Result<()> {
         self.inner.merge(place, rhs)?;
-        let flag = self.get_flag(place) + self.get_flag(rhs);
-        self.set_flag(place, flag);
+        let flag = self.get_flag(place) > 0 || self.get_flag(rhs) > 0;
+        self.set_flag(place, u8::from(flag));
         Ok(())
     }
 
diff --git a/src/query/functions-v2/src/aggregates/aggregate_combinator_distinct.rs b/src/query/functions-v2/src/aggregates/aggregate_combinator_distinct.rs
@@ -70,6 +70,7 @@ where State: DistinctStateFunc
 
     fn state_layout(&self) -> Layout {
         let layout = Layout::new::<State>();
+
         let netesed = self.nested.state_layout();
         Layout::from_size_align(layout.size() + netesed.size(), layout.align()).unwrap()
     }
diff --git a/src/query/functions/src/aggregates/adaptors/aggregate_null_unary_adaptor.rs b/src/query/functions/src/aggregates/adaptors/aggregate_null_unary_adaptor.rs
@@ -94,7 +94,7 @@ impl<const NULLABLE_RESULT: bool> AggregateFunction for AggregateNullUnaryAdapto
     #[inline]
     fn state_layout(&self) -> Layout {
         let layout = self.nested.state_layout();
-        let add = usize::from(NULLABLE_RESULT);
+        let add = if NULLABLE_RESULT { layout.align() } else { 0 };
         Layout::from_size_align(layout.size() + add, layout.align()).unwrap()
     }
 
@@ -183,9 +183,9 @@ impl<const NULLABLE_RESULT: bool> AggregateFunction for AggregateNullUnaryAdapto
 
     fn deserialize(&self, place: StateAddr, reader: &mut &[u8]) -> Result<()> {
         if NULLABLE_RESULT {
+            let flag = reader[reader.len() - 1];
             self.nested
                 .deserialize(place, &mut &reader[..reader.len() - 1])?;
-            let flag = reader[reader.len() - 1];
             self.set_flag(place, flag);
         } else {
             self.nested.deserialize(place, reader)?;
diff --git a/src/query/functions/src/aggregates/adaptors/aggregate_null_variadic_adaptor.rs b/src/query/functions/src/aggregates/adaptors/aggregate_null_variadic_adaptor.rs
@@ -95,7 +95,7 @@ impl<const NULLABLE_RESULT: bool> AggregateFunction
     #[inline]
     fn state_layout(&self) -> Layout {
         let layout = self.nested.state_layout();
-        let add = usize::from(NULLABLE_RESULT);
+        let add = if NULLABLE_RESULT { layout.align() } else { 0 };
         Layout::from_size_align(layout.size() + add, layout.align()).unwrap()
     }
 
@@ -198,9 +198,9 @@ impl<const NULLABLE_RESULT: bool> AggregateFunction
 
     fn deserialize(&self, place: StateAddr, reader: &mut &[u8]) -> Result<()> {
         if NULLABLE_RESULT {
+            let flag = reader[reader.len() - 1];
             self.nested
                 .deserialize(place, &mut &reader[..reader.len() - 1])?;
-            let flag = reader[reader.len() - 1];
             self.set_flag(place, flag);
         } else {
             self.nested.deserialize(place, reader)?;
diff --git a/src/query/functions/src/aggregates/adaptors/aggregate_ornull_adaptor.rs b/src/query/functions/src/aggregates/adaptors/aggregate_ornull_adaptor.rs
@@ -92,7 +92,7 @@ impl AggregateFunction for AggregateFunctionOrNullAdaptor {
     #[inline]
     fn state_layout(&self) -> std::alloc::Layout {
         let layout = self.inner.state_layout();
-        Layout::from_size_align(layout.size() + 1, layout.align()).unwrap()
+        Layout::from_size_align(layout.size() + layout.align(), layout.align()).unwrap()
     }
 
     #[inline]
@@ -178,17 +178,17 @@ impl AggregateFunction for AggregateFunctionOrNullAdaptor {
 
     #[inline]
     fn deserialize(&self, place: StateAddr, reader: &mut &[u8]) -> Result<()> {
+        let flag = reader[reader.len() - 1];
         self.inner
             .deserialize(place, &mut &reader[..reader.len() - 1])?;
-        let flag = reader[reader.len() - 1];
         self.set_flag(place, flag);
         Ok(())
     }
 
     fn merge(&self, place: StateAddr, rhs: StateAddr) -> Result<()> {
         self.inner.merge(place, rhs)?;
-        let flag = self.get_flag(place) + self.get_flag(rhs);
-        self.set_flag(place, flag);
+        let flag = self.get_flag(place) > 0 || self.get_flag(rhs) > 0;
+        self.set_flag(place, u8::from(flag));
         Ok(())
     }
 
diff --git a/src/query/functions/src/aggregates/aggregate_combinator_distinct.rs b/src/query/functions/src/aggregates/aggregate_combinator_distinct.rs
@@ -66,6 +66,7 @@ where State: DistinctStateFunc
 
     fn state_layout(&self) -> Layout {
         let layout = Layout::new::<State>();
+
         let netesed = self.nested.state_layout();
         Layout::from_size_align(layout.size() + netesed.size(), layout.align()).unwrap()
     }

Original file line number	Diff line number	Diff line change
`@@ -70,6 +70,7 @@ where State: DistinctStateFunc`
`70`	`70`
`71`	`71`	`fn state_layout(&self) -> Layout {`
`72`	`72`	`let layout = Layout::new::<State>();`
	`73`	`+`
`73`	`74`	`let netesed = self.nested.state_layout();`
`74`	`75`	`Layout::from_size_align(layout.size() + netesed.size(), layout.align()).unwrap()`
`75`	`76`	`}`
Original file line number	Diff line number	Diff line change
`@@ -66,6 +66,7 @@ where State: DistinctStateFunc`
`66`	`66`
`67`	`67`	`fn state_layout(&self) -> Layout {`
`68`	`68`	`let layout = Layout::new::<State>();`
	`69`	`+`
`69`	`70`	`let netesed = self.nested.state_layout();`
`70`	`71`	`Layout::from_size_align(layout.size() + netesed.size(), layout.align()).unwrap()`
`71`	`72`	`}`