fix: tpcc segfault caused by bump-backed executor drop order

Author: KKould

src/execution/dql/join/hash_join.rs

@@ -39,6 +39,7 @@ use std::mem::transmute;
 use std::sync::Arc;
 
 pub struct HashJoin {
+    state: HashJoinState,
     ty: JoinType,
     on_left_keys: Vec<ScalarExpression>,
     on_right_keys: Vec<ScalarExpression>,
@@ -51,7 +52,6 @@ pub struct HashJoin {
     left_input: ExecId,
     right_input: ExecId,
     bump: Box<Bump>,
-    state: HashJoinState,
     init_error: Option<DatabaseError>,
 }
 
@@ -109,6 +109,7 @@ impl From<(JoinOperator, LogicalPlan, LogicalPlan)> for HashJoin {
         let right_schema_len = full_schema_ref.len() - left_schema_len;
 
         HashJoin {
+            state: HashJoinState::Build,
             ty: join_type,
             on_left_keys,
             on_right_keys,
@@ -124,7 +125,6 @@ impl From<(JoinOperator, LogicalPlan, LogicalPlan)> for HashJoin {
             left_input: 0,
             right_input: 0,
             bump: Box::<Bump>::default(),
-            state: HashJoinState::Build,
             init_error,
         }
     }

src/execution/dql/top_k.rs

@@ -89,14 +89,14 @@ fn top_sort<'a>(
 }
 
 pub struct TopK {
+    output: Option<std::iter::Skip<BTreeSetIntoIter<CmpItem<'static>>>>,
     arena: Box<Bump>,
     sort_fields: Vec<SortField>,
     limit: usize,
     offset: Option<usize>,
     input_schema: SchemaRef,
     input_plan: Option<LogicalPlan>,
     input: ExecId,
-    output: Option<std::iter::Skip<BTreeSetIntoIter<CmpItem<'static>>>>,
 }
 
 impl From<(TopKOperator, LogicalPlan)> for TopK {
@@ -111,14 +111,14 @@ impl From<(TopKOperator, LogicalPlan)> for TopK {
         ): (TopKOperator, LogicalPlan),
     ) -> Self {
         TopK {
+            output: None,
             arena: Box::<Bump>::default(),
             sort_fields,
             limit,
             offset,
             input_schema: input.output_schema().clone(),
             input_plan: Some(input),
             input: 0,
-            output: None,
         }
     }
 }

src/orm/dql.rs

@@ -152,7 +152,9 @@ impl<'a, S: Storage> DBTransaction<'a, S> {
     }
 
     /// Fetches all rows for a model inside the current transaction.
-    pub fn fetch<M: Model>(&mut self) -> Result<OrmIter<TransactionIter<'_>, M>, DatabaseError> {
+    pub fn fetch<M: Model>(
+        &mut self,
+    ) -> Result<OrmIter<TransactionIter<'_, S::TransactionType<'a>>, M>, DatabaseError> {
         orm_list::<_, M>(self)
     }
 
@@ -164,7 +166,8 @@ impl<'a, S: Storage> DBTransaction<'a, S> {
     /// Lists all table names inside the current transaction.
     pub fn show_tables(
         &mut self,
-    ) -> Result<ProjectValueIter<TransactionIter<'_>, String>, DatabaseError> {
+    ) -> Result<ProjectValueIter<TransactionIter<'_, S::TransactionType<'a>>, String>, DatabaseError>
+    {
         Ok(ProjectValueIter::new(
             self.execute(&orm_show_tables_statement(), &[])?,
         ))
@@ -173,7 +176,8 @@ impl<'a, S: Storage> DBTransaction<'a, S> {
     /// Lists all view names inside the current transaction.
     pub fn show_views(
         &mut self,
-    ) -> Result<ProjectValueIter<TransactionIter<'_>, String>, DatabaseError> {
+    ) -> Result<ProjectValueIter<TransactionIter<'_, S::TransactionType<'a>>, String>, DatabaseError>
+    {
         Ok(ProjectValueIter::new(
             self.execute(&orm_show_views_statement(), &[])?,
         ))
@@ -182,7 +186,8 @@ impl<'a, S: Storage> DBTransaction<'a, S> {
     /// Describes the schema of the model table inside the current transaction.
     pub fn describe<M: Model>(
         &mut self,
-    ) -> Result<OrmIter<TransactionIter<'_>, DescribeColumn>, DatabaseError> {
+    ) -> Result<OrmIter<TransactionIter<'_, S::TransactionType<'a>>, DescribeColumn>, DatabaseError>
+    {
         Ok(self
             .execute(&orm_describe_statement(M::table_name()), &[])?
             .orm::<DescribeColumn>())

src/orm/mod.rs

@@ -1365,7 +1365,7 @@ impl<'a, S: Storage> StatementSource for &'a Database<S> {
 }
 
 impl<'a, 'tx, S: Storage> StatementSource for &'a mut DBTransaction<'tx, S> {
-    type Iter = TransactionIter<'a>;
+    type Iter = TransactionIter<'a, S::TransactionType<'tx>>;
 
     fn execute_statement<A: AsRef<[(&'static str, DataValue)]>>(
         self,