PortableSpiller

Author: forsaken628

src/query/service/src/servers/http/v1/query/execute_state.rs

@@ -50,12 +50,14 @@ use crate::sessions::QueryAffect;
 use crate::sessions::QueryContext;
 use crate::sessions::Session;
 use crate::sessions::TableContext;
-use crate::spillers::Spiller;
+use crate::spillers::PortableSpiller;
+use crate::spillers::PortableSpillerRef;
 use crate::spillers::SpillerConfig;
 use crate::spillers::SpillerDiskConfig;
-use crate::spillers::SpillerRef;
 use crate::spillers::SpillerType;
 
+type Sender = SizedChannelSender<PortableSpillerRef>;
+
 pub struct ExecutionError;
 
 #[derive(Serialize, Deserialize, Debug, Copy, Clone, PartialEq, Eq)]
@@ -120,7 +122,7 @@ impl ExecuteState {
 
 pub struct ExecuteStarting {
     pub(crate) ctx: Arc<QueryContext>,
-    pub(crate) sender: SizedChannelSender<SpillerRef>,
+    pub(crate) sender: Sender,
 }
 
 pub struct ExecuteRunning {
@@ -361,7 +363,7 @@ impl ExecuteState {
         sql: String,
         session: Arc<Session>,
         ctx: Arc<QueryContext>,
-        mut block_sender: SizedChannelSender<SpillerRef>,
+        mut block_sender: Sender,
     ) -> Result<(), ExecutionError> {
         let make_error = || format!("failed to start query: {sql}");
 
@@ -426,7 +428,7 @@ impl ExecuteState {
         interpreter: Arc<dyn Interpreter>,
         schema: DataSchemaRef,
         ctx: Arc<QueryContext>,
-        mut block_sender: SizedChannelSender<SpillerRef>,
+        mut block_sender: Sender,
         executor: Arc<Mutex<Executor>>,
     ) -> Result<(), ExecutionError> {
         let make_error = || format!("failed to execute {}", interpreter.name());
@@ -471,7 +473,7 @@ impl ExecuteState {
     }
 
     async fn send_data_block(
-        sender: &mut SizedChannelSender<SpillerRef>,
+        sender: &mut Sender,
         executor: &Arc<Mutex<Executor>>,
         block: DataBlock,
     ) -> Result<()> {
@@ -484,10 +486,7 @@ impl ExecuteState {
         }
     }
 
-    fn apply_settings(
-        ctx: &Arc<QueryContext>,
-        block_sender: &mut SizedChannelSender<SpillerRef>,
-    ) -> Result<()> {
+    fn apply_settings(ctx: &Arc<QueryContext>, block_sender: &mut Sender) -> Result<()> {
         let settings = ctx.get_settings();
 
         let spiller = if settings.get_enable_result_set_spilling()? {
@@ -507,7 +506,7 @@ impl ExecuteState {
                 use_parquet: settings.get_spilling_file_format()?.is_parquet(),
             };
             let op = DataOperator::instance().spill_operator();
-            Some(Spiller::create(ctx.clone(), op, config)?.into())
+            Some(PortableSpiller::create(op, config)?)
         } else {
             None
         };

src/query/service/src/servers/http/v1/query/http_query.rs

@@ -74,7 +74,7 @@ use crate::servers::http::v1::QueryStats;
 use crate::sessions::QueryAffect;
 use crate::sessions::Session;
 use crate::sessions::TableContext;
-use crate::spillers::SpillerRef;
+use crate::spillers::PortableSpillerRef;
 
 fn default_as_true() -> bool {
     true
@@ -612,7 +612,8 @@ impl HttpQuery {
             })
         };
 
-        let (sender, receiver) = sized_spsc::<SpillerRef>(req.pagination.max_rows_in_buffer);
+        let (sender, receiver) =
+            sized_spsc::<PortableSpillerRef>(req.pagination.max_rows_in_buffer);
 
         let executor = Arc::new(Mutex::new(Executor {
             query_id: query_id.clone(),

src/query/service/src/servers/http/v1/query/page_manager.rs

@@ -20,7 +20,7 @@ use databend_common_exception::Result;
 use super::blocks_serializer::BlocksSerializer;
 use crate::servers::http::v1::query::sized_spsc::SizedChannelReceiver;
 use crate::servers::http::v1::query::sized_spsc::Wait;
-use crate::spillers::SpillerRef;
+use crate::spillers::PortableSpillerRef;
 
 #[derive(Clone)]
 pub struct Page {
@@ -38,13 +38,13 @@ pub struct PageManager {
     total_pages: usize,
     end: bool,
     last_page: Option<Page>,
-    block_receiver: SizedChannelReceiver<SpillerRef>,
+    block_receiver: SizedChannelReceiver<PortableSpillerRef>,
 }
 
 impl PageManager {
     pub fn new(
         max_rows_per_page: usize,
-        block_receiver: SizedChannelReceiver<SpillerRef>,
+        block_receiver: SizedChannelReceiver<PortableSpillerRef>,
     ) -> PageManager {
         PageManager {
             total_rows: 0,