refactor: stable kitesql

Author: KKould

Cargo.lock

@@ -2,7 +2,7 @@
 
 [package]
 name          = "kite_sql"
-version       = "0.1.2"
+version       = "0.1.3"
 edition       = "2021"
 authors       = ["Kould <kould2333@gmail.com>", "Xwg <loloxwg@gmail.com>"]
 description   = "SQL as a Function for Rust"
@@ -60,6 +60,8 @@ sqlparser             = { version = "0.34", features = ["serde"] }
 thiserror             = { version = "1" }
 typetag               = { version = "0.2" }
 ulid                  = { version = "1", features = ["serde"] }
+genawaiter            = { version = "0.99" }
+rand                  = { version = "0.8" }
 
 # Feature: net
 async-trait           = { version = "0.1", optional = true }

Cargo.toml

@@ -1 +1 @@
-nightly-2024-10-10
+stable

rust-toolchain

@@ -28,9 +28,7 @@ use parking_lot::{RawRwLock, RwLock};
 use std::hash::RandomState;
 use std::marker::PhantomData;
 use std::mem;
-use std::ops::{Coroutine, CoroutineState};
 use std::path::PathBuf;
-use std::pin::Pin;
 use std::sync::atomic::AtomicUsize;
 use std::sync::Arc;
 
@@ -434,32 +432,19 @@ impl<S: Storage> DBTransaction<'_, S> {
 pub struct TransactionIter<'a> {
     executor: Executor<'a>,
     schema: SchemaRef,
-    is_over: bool,
 }
 
 impl<'a> TransactionIter<'a> {
     fn new(schema: SchemaRef, executor: Executor<'a>) -> Self {
-        Self {
-            executor,
-            schema,
-            is_over: false,
-        }
+        Self { executor, schema }
     }
 }
 
 impl Iterator for TransactionIter<'_> {
     type Item = Result<Tuple, DatabaseError>;
 
     fn next(&mut self) -> Option<Self::Item> {
-        if self.is_over {
-            return None;
-        }
-        if let CoroutineState::Yielded(tuple) = Pin::new(&mut self.executor).resume(()) {
-            Some(tuple)
-        } else {
-            self.is_over = true;
-            None
-        }
+        self.executor.next()
     }
 }
 

src/db.rs

@@ -1,5 +1,5 @@
 use crate::errors::DatabaseError;
-use crate::execution::{build_read, Executor, WriteExecutor};
+use crate::execution::{build_read, spawn_executor, Executor, WriteExecutor};
 use crate::planner::LogicalPlan;
 use crate::storage::{StatisticsMetaCache, TableCache, ViewCache};
 use crate::types::index::{Index, IndexType};
@@ -9,9 +9,6 @@ use crate::types::value::DataValue;
 use crate::{
     planner::operator::alter_table::add_column::AddColumnOperator, storage::Transaction, throw,
 };
-use std::ops::Coroutine;
-use std::ops::CoroutineState;
-use std::pin::Pin;
 
 pub struct AddColumn {
     op: AddColumnOperator,
@@ -30,73 +27,83 @@ impl<'a, T: Transaction + 'a> WriteExecutor<'a, T> for AddColumn {
         cache: (&'a TableCache, &'a ViewCache, &'a StatisticsMetaCache),
         transaction: *mut T,
     ) -> Executor<'a> {
-        Box::new(
-            #[coroutine]
-            move || {
-                let AddColumnOperator {
-                    table_name,
-                    column,
-                    if_not_exists,
-                } = &self.op;
+        spawn_executor(move |co| async move {
+            let AddColumnOperator {
+                table_name,
+                column,
+                if_not_exists,
+            } = &self.op;
 
-                let mut unique_values = column.desc().is_unique().then(Vec::new);
-                let mut tuples = Vec::new();
-                let schema = self.input.output_schema();
-                let mut types = Vec::with_capacity(schema.len() + 1);
+            let mut unique_values = column.desc().is_unique().then(Vec::new);
+            let mut tuples = Vec::new();
+            let schema = self.input.output_schema();
+            let mut types = Vec::with_capacity(schema.len() + 1);
 
-                for column_ref in schema.iter() {
-                    types.push(column_ref.datatype().clone());
-                }
-                types.push(column.datatype().clone());
+            for column_ref in schema.iter() {
+                types.push(column_ref.datatype().clone());
+            }
+            types.push(column.datatype().clone());
 
-                let mut coroutine = build_read(self.input, cache, transaction);
+            let mut coroutine = build_read(self.input, cache, transaction);
 
-                while let CoroutineState::Yielded(tuple) = Pin::new(&mut coroutine).resume(()) {
-                    let mut tuple: Tuple = throw!(tuple);
+            while let Some(tuple) = coroutine.next() {
+                let mut tuple: Tuple = throw!(co, tuple);
 
-                    if let Some(value) = throw!(column.default_value()) {
-                        if let Some(unique_values) = &mut unique_values {
-                            unique_values.push((
-                                throw!(tuple.pk.clone().ok_or(DatabaseError::PrimaryKeyNotFound)),
-                                value.clone(),
-                            ));
-                        }
-                        tuple.values.push(value);
-                    } else {
-                        tuple.values.push(DataValue::Null);
+                if let Some(value) = throw!(co, column.default_value()) {
+                    if let Some(unique_values) = &mut unique_values {
+                        unique_values.push((
+                            throw!(
+                                co,
+                                tuple.pk.clone().ok_or(DatabaseError::PrimaryKeyNotFound)
+                            ),
+                            value.clone(),
+                        ));
                     }
-                    tuples.push(tuple);
+                    tuple.values.push(value);
+                } else {
+                    tuple.values.push(DataValue::Null);
                 }
-                drop(coroutine);
+                tuples.push(tuple);
+            }
+            drop(coroutine);
 
-                for tuple in tuples {
-                    throw!(unsafe { &mut (*transaction) }
-                        .append_tuple(table_name, tuple, &types, true));
-                }
-                let col_id = throw!(unsafe { &mut (*transaction) }.add_column(
+            for tuple in tuples {
+                throw!(
+                    co,
+                    unsafe { &mut (*transaction) }.append_tuple(table_name, tuple, &types, true)
+                );
+            }
+            let col_id = throw!(
+                co,
+                unsafe { &mut (*transaction) }.add_column(
                     cache.0,
                     table_name,
                     column,
                     *if_not_exists
-                ));
+                )
+            );
 
-                // Unique Index
-                if let (Some(unique_values), Some(unique_meta)) = (
-                    unique_values,
-                    throw!(unsafe { &mut (*transaction) }.table(cache.0, table_name.clone()))
-                        .and_then(|table| table.get_unique_index(&col_id))
-                        .cloned(),
-                ) {
-                    for (tuple_id, value) in unique_values {
-                        let index = Index::new(unique_meta.id, &value, IndexType::Unique);
-                        throw!(
-                            unsafe { &mut (*transaction) }.add_index(table_name, index, &tuple_id)
-                        );
-                    }
+            // Unique Index
+            if let (Some(unique_values), Some(unique_meta)) = (
+                unique_values,
+                throw!(
+                    co,
+                    unsafe { &mut (*transaction) }.table(cache.0, table_name.clone())
+                )
+                .and_then(|table| table.get_unique_index(&col_id))
+                .cloned(),
+            ) {
+                for (tuple_id, value) in unique_values {
+                    let index = Index::new(unique_meta.id, &value, IndexType::Unique);
+                    throw!(
+                        co,
+                        unsafe { &mut (*transaction) }.add_index(table_name, index, &tuple_id)
+                    );
                 }
+            }
 
-                yield Ok(TupleBuilder::build_result("1".to_string()));
-            },
-        )
+            co.yield_(Ok(TupleBuilder::build_result("1".to_string())))
+                .await;
+        })
     }
 }