1.2.307 hotfix 20240430

Cargo.toml

@@ -46,7 +46,6 @@ members = [
     "src/query/pipeline/sinks",
     "src/query/pipeline/sources",
     "src/query/pipeline/transforms",
-    "src/query/profile",
     "src/query/settings",
     "src/query/sql",
     "src/query/storages/common/blocks",

src/common/cloud_control/proto/task.proto

@@ -33,6 +33,7 @@ message CreateTaskRequest {
   // DAG
   repeated string after = 11; //
   optional string when_condition = 12;
+  map<string, string> session_parameters = 13;
 }
 
 message TaskError {
@@ -90,6 +91,7 @@ message Task {
   optional string last_suspended_at = 16;
   repeated string after = 17;
   optional string when_condition = 18;
+  map<string, string> session_parameters = 19;
 }
 
 
@@ -134,6 +136,8 @@ message AlterTaskRequest {
   optional string when_condition = 11;
   repeated string add_after = 12;
   repeated string remove_after = 13;
+  bool set_session_parameters = 14;
+  map<string, string> session_parameters = 15;
 }
 
 message AlterTaskResponse {
@@ -179,6 +183,7 @@ message TaskRun {
   string query_id = 17;
   string condition_text = 18;
   string root_task_id = 19;
+  map<string, string> session_parameters = 20;
 }
 
 message ShowTaskRunsResponse {

src/common/cloud_control/src/task_utils.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::BTreeMap;
 use std::fmt::Display;
 use std::fmt::Formatter;
 
@@ -77,6 +78,7 @@ pub struct Task {
     pub created_at: DateTime<Utc>,
     pub updated_at: DateTime<Utc>,
     pub last_suspended_at: Option<DateTime<Utc>>,
+    pub session_params: BTreeMap<String, String>,
 }
 
 pub fn format_schedule_options(s: &ScheduleOptions) -> Result<String> {
@@ -203,6 +205,7 @@ impl TryFrom<crate::pb::Task> for Task {
             status,
             created_at,
             updated_at,
+            session_params: value.session_parameters,
         };
         Ok(t)
     }
@@ -227,6 +230,7 @@ pub struct TaskRun {
     pub error_code: i64,
     pub error_message: Option<String>,
     pub root_task_id: String,
+    pub session_params: BTreeMap<String, String>,
 }
 
 // convert from crate::pb::taskRun to struct taskRun
@@ -305,6 +309,7 @@ impl TryFrom<crate::pb::TaskRun> for TaskRun {
             scheduled_at,
             completed_at,
             root_task_id: value.root_task_id,
+            session_params: value.session_parameters,
         };
         Ok(tr)
     }

src/common/cloud_control/tests/it/task_client.rs

@@ -82,6 +82,7 @@ impl TaskService for MockTaskService {
                 last_suspended_at: None,
                 after: vec![],
                 when_condition: None,
+                session_parameters: Default::default(),
             }),
             error: None,
         }))
@@ -196,6 +197,7 @@ async fn test_task_client_success_cases() -> Result<()> {
         if_not_exist: false,
         after: vec![],
         when_condition: None,
+        session_parameters: Default::default(),
     });
 
     let response = client.create_task(request).await?;

src/common/hashtable/src/hashjoin_hashtable.rs

@@ -209,56 +209,80 @@ where
         count
     }
 
-    // Using hashes to probe hash table and converting them in-place to pointers for memory reuse.
-    fn early_filtering_probe(&self, hashes: &mut [u64], bitmap: Option<Bitmap>) -> usize {
+    // Perform early filtering probe, store matched indexes in `matched_selection` and store unmatched indexes
+    // in `unmatched_selection`, return the number of matched and unmatched indexes.
+    fn early_filtering_probe(
+        &self,
+        hashes: &mut [u64],
+        bitmap: Option<Bitmap>,
+        matched_selection: &mut [u32],
+        unmatched_selection: &mut [u32],
+    ) -> (usize, usize) {
         let mut valids = None;
         if let Some(bitmap) = bitmap {
             if bitmap.unset_bits() == bitmap.len() {
-                hashes.iter_mut().for_each(|hash| {
-                    *hash = 0;
-                });
-                return 0;
+                unmatched_selection
+                    .iter_mut()
+                    .enumerate()
+                    .for_each(|(idx, val)| {
+                        *val = idx as u32;
+                    });
+                return (0, hashes.len());
             } else if bitmap.unset_bits() > 0 {
                 valids = Some(bitmap);
             }
         }
-        let mut count = 0;
+        let mut matched_idx = 0;
+        let mut unmatched_idx = 0;
         match valids {
             Some(valids) => {
-                valids
-                    .iter()
-                    .zip(hashes.iter_mut())
-                    .for_each(|(valid, hash)| {
+                valids.iter().zip(hashes.iter_mut().enumerate()).for_each(
+                    |(valid, (idx, hash))| {
                         if valid {
                             let header = self.pointers[(*hash >> self.hash_shift) as usize];
                             if header != 0 && early_filtering(header, *hash) {
                                 *hash = remove_header_tag(header);
-                                count += 1;
+                                unsafe {
+                                    *matched_selection.get_unchecked_mut(matched_idx) = idx as u32
+                                };
+                                matched_idx += 1;
                             } else {
-                                *hash = 0;
+                                unsafe {
+                                    *unmatched_selection.get_unchecked_mut(unmatched_idx) =
+                                        idx as u32
+                                };
+                                unmatched_idx += 1;
                             }
                         } else {
-                            *hash = 0;
+                            unsafe {
+                                *unmatched_selection.get_unchecked_mut(unmatched_idx) = idx as u32
+                            };
+                            unmatched_idx += 1;
                         }
-                    });
+                    },
+                );
             }
             None => {
-                hashes.iter_mut().for_each(|hash| {
+                hashes.iter_mut().enumerate().for_each(|(idx, hash)| {
                     let header = self.pointers[(*hash >> self.hash_shift) as usize];
                     if header != 0 && early_filtering(header, *hash) {
                         *hash = remove_header_tag(header);
-                        count += 1;
+                        unsafe { *matched_selection.get_unchecked_mut(matched_idx) = idx as u32 };
+                        matched_idx += 1;
                     } else {
-                        *hash = 0;
+                        unsafe {
+                            *unmatched_selection.get_unchecked_mut(unmatched_idx) = idx as u32
+                        };
+                        unmatched_idx += 1;
                     }
                 });
             }
         }
-        count
+        (matched_idx, unmatched_idx)
     }
 
-    // Using hashes to probe hash table and converting them in-place to pointers for memory reuse.
-    fn early_filtering_probe_with_selection(
+    // Perform early filtering probe and store matched indexes in `selection`, return the number of matched indexes.
+    fn early_filtering_matched_probe(
         &self,
         hashes: &mut [u64],
         bitmap: Option<Bitmap>,

src/common/hashtable/src/hashjoin_string_hashtable.rs

@@ -138,53 +138,77 @@ where A: Allocator + Clone + 'static
         count
     }
 
-    // Using hashes to probe hash table and converting them in-place to pointers for memory reuse.
-    fn early_filtering_probe(&self, hashes: &mut [u64], bitmap: Option<Bitmap>) -> usize {
+    // Perform early filtering probe, store matched indexes in `matched_selection` and store unmatched indexes
+    // in `unmatched_selection`, return the number of matched and unmatched indexes.
+    fn early_filtering_probe(
+        &self,
+        hashes: &mut [u64],
+        bitmap: Option<Bitmap>,
+        matched_selection: &mut [u32],
+        unmatched_selection: &mut [u32],
+    ) -> (usize, usize) {
         let mut valids = None;
         if let Some(bitmap) = bitmap {
             if bitmap.unset_bits() == bitmap.len() {
-                hashes.iter_mut().for_each(|hash| {
-                    *hash = 0;
-                });
-                return 0;
+                unmatched_selection
+                    .iter_mut()
+                    .enumerate()
+                    .for_each(|(idx, val)| {
+                        *val = idx as u32;
+                    });
+                return (0, hashes.len());
             } else if bitmap.unset_bits() > 0 {
                 valids = Some(bitmap);
             }
         }
-        let mut count = 0;
+        let mut matched_idx = 0;
+        let mut unmatched_idx = 0;
         match valids {
             Some(valids) => {
                 hashes.iter_mut().enumerate().for_each(|(idx, hash)| {
                     if unsafe { valids.get_bit_unchecked(idx) } {
                         let header = self.pointers[(*hash >> self.hash_shift) as usize];
                         if header != 0 && early_filtering(header, *hash) {
                             *hash = remove_header_tag(header);
-                            count += 1;
+                            unsafe {
+                                *matched_selection.get_unchecked_mut(matched_idx) = idx as u32
+                            };
+                            matched_idx += 1;
                         } else {
-                            *hash = 0;
+                            unsafe {
+                                *unmatched_selection.get_unchecked_mut(unmatched_idx) = idx as u32
+                            };
+                            unmatched_idx += 1;
                         }
                     } else {
-                        *hash = 0;
-                    };
+                        unsafe {
+                            *unmatched_selection.get_unchecked_mut(unmatched_idx) = idx as u32
+                        };
+                        unmatched_idx += 1;
+                    }
                 });
             }
             None => {
-                hashes.iter_mut().for_each(|hash| {
+                hashes.iter_mut().enumerate().for_each(|(idx, hash)| {
                     let header = self.pointers[(*hash >> self.hash_shift) as usize];
                     if header != 0 && early_filtering(header, *hash) {
                         *hash = remove_header_tag(header);
-                        count += 1;
+                        unsafe { *matched_selection.get_unchecked_mut(matched_idx) = idx as u32 };
+                        matched_idx += 1;
                     } else {
-                        *hash = 0;
+                        unsafe {
+                            *unmatched_selection.get_unchecked_mut(unmatched_idx) = idx as u32
+                        };
+                        unmatched_idx += 1;
                     }
                 });
             }
         }
-        count
+        (matched_idx, unmatched_idx)
     }
 
-    // Using hashes to probe hash table and converting them in-place to pointers for memory reuse.
-    fn early_filtering_probe_with_selection(
+    // Perform early filtering probe and store matched indexes in `selection`, return the number of matched indexes.
+    fn early_filtering_matched_probe(
         &self,
         hashes: &mut [u64],
         bitmap: Option<Bitmap>,