Repository: risingwavelabs/risingwave (language: python)

Issue:
Sqlsmith: Sql feature generation
As we add more features, these should be tested via sqlsmith as well.

SQL Features:
- [ ] Generate other table functions: `generate_series`, `unnest`. See: https://github.com/singularity-data/risingwave/issues/4540
- [ ] Generate array_agg, jsonb_agg, jsonb_object_agg
- [ ] Generate select from System Tables
- [ ] Generate `NATURAL JOIN`
- [ ] Generate `JOIN` with `USING`
- [ ] Generate query w/o `GROUP BY`
- [ ] Randomly toggle session variables
- [ ] Generate `pow`/`exp`.
- [ ] Generate TopN by group: https://www.risingwave.dev/docs/current/sql-pattern-topn/
- [ ] Generate jsonb exprs, see: https://github.com/risingwavelabs/risingwave/issues/7714
- [ ] #7132
- [x] Generate order-by-clauses with `NULLS { FIRST | LAST }` (will be done in https://github.com/risingwavelabs/risingwave/pull/10864/commits/000cbc11cbc04f20abfd127b1bf4b3fd24e75c44)
- [ ] Generate `max`,`min`,`count` for `timestamptz`
- [ ] Generate `bit_and`, `bit_or` and `big_xor` in batch queries, and `bit_xor` in streaming queries.
- [ ] Generate array range access.
- [ ] https://github.com/risingwavelabs/risingwave/issues/10241
- [ ] Generate window functions other than `row_number`, `rank`.
- [ ] Generate `ALTER` DDL statements.\
- [ ] https://github.com/risingwavelabs/risingwave/issues/10516
- [ ] Generate temporal filter in sqlsmith


Discussion / hints:

Fix patch:
diff --git a/src/meta/src/hummock/compaction/selector/level_selector.rs b/src/meta/src/hummock/compaction/selector/level_selector.rs
index 321a3a65f713..59c24c3cc287 100644
--- a/src/meta/src/hummock/compaction/selector/level_selector.rs
+++ b/src/meta/src/hummock/compaction/selector/level_selector.rs
@@ -205,13 +205,26 @@ impl DynamicLevelSelectorCore {
     ) -> SelectContext {
         let mut ctx = self.calculate_level_base_size(levels);
 
-        let idle_file_count = levels
+        let l0_file_count = levels
             .l0
             .sub_levels
             .iter()
-            .map(|level| level.table_infos.len())
-            .sum::<usize>()
-            - handlers[0].get_pending_file_count();
+            .map(|sub_level| sub_level.table_infos.len())
+            .sum::<usize>();
+
+        let idle_file_count = match l0_file_count.checked_sub(handlers[0].pending_file_count()) {
+            Some(count) => count,
+            None => {
+                // If the number of files in L0 is less than the number of pending files, it means
+                // that may be encountered some issue, we can work around it.
+                tracing::warn!(
+                    "The number of files in L0 {} is less than the number of pending files {} group {} pending_tasks_ids {:?} compacting_files {:?}",
+                    l0_file_count, handlers[0].pending_file_count(), levels.group_id, handlers[0].pending_tasks_ids(), handlers[0].compacting_files()
+                );
+
+                0
+            }
+        };
 
         if idle_file_count > 0 {
             // trigger l0 compaction when the number of files is too large.
@@ -254,7 +267,7 @@ impl DynamicLevelSelectorCore {
                 })
                 .map(|level| level.total_file_size)
                 .sum::<u64>()
-                .saturating_sub(handlers[0].get_pending_output_file_size(ctx.base_level as u32));
+                .saturating_sub(handlers[0].pending_output_file_size(ctx.base_level as u32));
             let base_level_size = levels.get_level(ctx.base_level).total_file_size;
             let base_level_sst_count = levels.get_level(ctx.base_level).table_infos.len() as u64;
 
@@ -305,7 +318,7 @@ impl DynamicLevelSelectorCore {
                 continue;
             }
             let output_file_size =
-                handlers[level_idx].get_pending_output_file_size(level.level_idx + 1);
+                handlers[level_idx].pending_output_file_size(level.level_idx + 1);
             let total_size = level.total_file_size.saturating_sub(output_file_size);
             if total_size == 0 {
                 continue;
diff --git a/src/meta/src/hummock/level_handler.rs b/src/meta/src/hummock/level_handler.rs
index 7ebc55eb1d3a..e9c46c2cf40f 100644
--- a/src/meta/src/hummock/level_handler.rs
+++ b/src/meta/src/hummock/level_handler.rs
@@ -104,18 +104,18 @@ impl LevelHandler {
         });
     }
 
-    pub fn get_pending_file_count(&self) -> usize {
+    pub fn pending_file_count(&self) -> usize {
         self.compacting_files.len()
     }
 
-    pub fn get_pending_file_size(&self) -> u64 {
+    pub fn pending_file_size(&self) -> u64 {
         self.pending_tasks
             .iter()
             .map(|task| task.total_file_size)
             .sum::<u64>()
     }
 
-    pub fn get_pending_output_file_size(&self, target_level: u32) -> u64 {
+    pub fn pending_output_file_size(&self, target_level: u32) -> u64 {
         self.pending_tasks
             .iter()
             .filter(|task| task.target_level == target_level)
@@ -130,9 +130,13 @@ impl LevelHandler {
             .collect_vec()
     }
 
-    pub fn get_pending_tasks(&self) -> &[RunningCompactTask] {
+    pub fn pending_tasks(&self) -> &[RunningCompactTask] {
         &self.pending_tasks
     }
+
+    pub fn compacting_files(&self) -> &HashMap<HummockSstableId, HummockCompactionTaskId> {
+        &self.compacting_files
+    }
 }
 
 impl From<&LevelHandler> for risingwave_pb::hummock::LevelHandler {
diff --git a/src/meta/src/hummock/manager/commit_epoch.rs b/src/meta/src/hummock/manager/commit_epoch.rs
index 4dbb1c71b35d..91bd1857fc19 100644
--- a/src/meta/src/hummock/manager/commit_epoch.rs
+++ b/src/meta/src/hummock/manager/commit_epoch.rs
@@ -18,9 +18,10 @@ use std::sync::Arc;
 use itertools::Itertools;
 use risingwave_common::catalog::TableId;
 use risingwave_common::config::default::compaction_config;
+use risingwave_common::hash::VirtualNode;
 use risingwave_common::system_param::reader::SystemParamsRead;
 use risingwave_hummock_sdk::change_log::ChangeLogDelta;
-use risingwave_hummock_sdk::compaction_group::group_split::split_sst_with_table_ids;
+use risingwave_hummock_sdk::compaction_group::group_split::{self, split_sst};
 use risingwave_hummock_sdk::sstable_info::SstableInfo;
 use risingwave_hummock_sdk::table_stats::{
     add_prost_table_stats_map, purge_prost_table_stats, to_prost_table_stats_map, PbTableStatsMap,
@@ -375,10 +376,18 @@ impl HummockManager {
 
         for (mut sst, group_table_ids) in sst_to_cg_vec {
             let len = group_table_ids.len();
+            let mut group_table_ids = group_table_ids.into_iter().collect_vec();
+            group_table_ids.sort_by_key(|(_, table_ids)| table_ids[0]);
             for (index, (group_id, match_ids)) in group_table_ids.into_iter().enumerate() {
                 if sst.sst_info.table_ids == match_ids {
                     // The SST contains all the tables in the group should be last key
-                    assert!(index == len - 1);
+                    assert!(
+                        index == len - 1,
+                        "SST should be the last key in the group {} index {} len {}",
+                        group_id,
+                        index,
+                        len
+                    );
                     commit_sstables
                         .entry(group_id)
                         .or_default()
@@ -395,20 +404,39 @@ impl HummockManager {
                     })
                     .sum();
 
-                // TODO(li0k): replace with `split_sst`
-                let (modified_sst_info, branch_sst) = split_sst_with_table_ids(
-                    &sst.sst_info,
+                let sstable_info = sst.sst_info;
+                let sst_id = sstable_info.sst_id;
+                let object_id = sstable_info.object_id;
+                let right_sst_size = origin_sst_size.checked_sub(new_sst_size).unwrap_or_else(|| {
+                    panic!(
+                        "group_id {} sst_id {} object_id {} is not split correctly, new_sst_size {} is larger than origin_sst_size {} match_ids {:?} table_stats {:?}",
+                        group_id, sst_id, object_id, new_sst_size, origin_sst_size, match_ids, sst.table_stats
+                    );
+                });
+                let (left, right) = split_sst(
+                    sstable_info,
                     &mut new_sst_id,
-                    origin_sst_size - new_sst_size,
+                    group_split::build_split_key(*match_ids.last().unwrap() + 1, VirtualNode::ZERO),
                     new_sst_size,
-                    match_ids,
+                    right_sst_size,
                 );
-                sst.sst_info = modified_sst_info;
 
                 commit_sstables
                     .entry(group_id)
                     .or_default()
-                    .push(branch_sst);
+                    .push(left.unwrap_or_else(|| {
+                        panic!(
+                            "group_id {} sst_id {} object_id {} is not split correctly, left part is missing ",
+                            group_id, sst_id, object_id
+                        );
+                    }));
+
+                sst.sst_info = right.unwrap_or_else(|| {
+                    panic!(
+                        "group_id {} sst_id {} object_id {} is not split correctly, right part is missing ",
+                        group_id, sst_id, object_id
+                    );
+                });
             }
         }
 
diff --git a/src/meta/src/hummock/manager/compaction/compaction_group_schedule.rs b/src/meta/src/hummock/manager/compaction/compaction_group_schedule.rs
index 8504a75fe9e2..b98129ace4e7 100644
--- a/src/meta/src/hummock/manager/compaction/compaction_group_schedule.rs
+++ b/src/meta/src/hummock/manager/compaction/compaction_group_schedule.rs
@@ -324,10 +324,9 @@ impl HummockManager {
                 }
             });
 
-        drop(versioning_guard);
-        drop(compaction_guard);
         if !canceled_tasks.is_empty() {
-            self.report_compact_tasks(canceled_tasks).await?;
+            self.report_compact_tasks_impl(canceled_tasks, compaction_guard, versioning_guard)
+                .await?;
         }
 
         self.metrics
@@ -559,10 +558,9 @@ impl HummockManager {
                 }
             });
 
-        drop(versioning_guard);
-        drop(compaction_guard);
         if !canceled_tasks.is_empty() {
-            self.report_compact_tasks(canceled_tasks).await?;
+            self.report_compact_tasks_impl(canceled_tasks, compaction_guard, versioning_guard)
+                .await?;
         }
 
         self.metrics
diff --git a/src/meta/src/hummock/manager/compaction/mod.rs b/src/meta/src/hummock/manager/compaction/mod.rs
index bfa5e0ec97a5..14413159faa7 100644
--- a/src/meta/src/hummock/manager/compaction/mod.rs
+++ b/src/meta/src/hummock/manager/compaction/mod.rs
@@ -69,6 +69,7 @@ use thiserror_ext::AsReport;
 use tokio::sync::mpsc::error::SendError;
 use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver};
 use tokio::sync::oneshot::{Receiver as OneShotReceiver, Sender};
+use tokio::sync::RwLockWriteGuard;
 use tokio::task::JoinHandle;
 use tonic::Streaming;
 use tracing::warn;
@@ -1099,6 +1100,14 @@ impl HummockManager {
         Ok(rets[0])
     }
 
+    pub async fn report_compact_tasks(&self, report_tasks: Vec<ReportTask>) -> Result<Vec<bool>> {
+        let compaction_guard = self.compaction.write().await;
+        let versioning_guard = self.versioning.write().await;
+
+        self.report_compact_tasks_impl(report_tasks, compaction_guard, versioning_guard)
+            .await
+    }
+
     /// Finishes or cancels a compaction task, according to `task_status`.
     ///
     /// If `context_id` is not None, its validity will be checked when writing meta store.
@@ -1106,10 +1115,14 @@ impl HummockManager {
     ///
     /// Return Ok(false) indicates either the task is not found,
     /// or the task is not owned by `context_id` when `context_id` is not None.
-    pub async fn report_compact_tasks(&self, report_tasks: Vec<ReportTask>) -> Result<Vec<bool>> {
-        let mut guard = self.compaction.write().await;
+    pub async fn report_compact_tasks_impl(
+        &self,
+        report_tasks: Vec<ReportTask>,
+        mut compaction_guard: RwLockWriteGuard<'_, Compaction>,
+        mut versioning_guard: RwLockWriteGuard<'_, Versioning>,
+    ) -> Result<Vec<bool>> {
         let deterministic_mode = self.env.opts.compaction_deterministic_test;
-        let compaction: &mut Compaction = &mut guard;
+        let compaction: &mut Compaction = &mut compaction_guard;
         let start_time = Instant::now();
         let original_keys = compaction.compaction_statuses.keys().cloned().collect_vec();
         let mut compact_statuses = BTreeMapTransaction::new(&mut compaction.compaction_statuses);
@@ -1117,7 +1130,6 @@ impl HummockManager {
         let mut compact_task_assignment =
             BTreeMapTransaction::new(&mut compaction.compact_task_assignment);
         // The compaction task is finished.
-        let mut versioning_guard = self.versioning.write().await;
         let versioning: &mut Versioning = &mut versioning_guard;
         let _timer = start_measure_real_process_timer!(self, "report_compact_tasks");
 
diff --git a/src/meta/src/hummock/manager/timer_task.rs b/src/meta/src/hummock/manager/timer_task.rs
index cd948500aaa6..8855de039182 100644
--- a/src/meta/src/hummock/manager/timer_task.rs
+++ b/src/meta/src/hummock/manager/timer_task.rs
@@ -526,7 +526,7 @@ impl HummockManager {
             for group_id in slowdown_groups.keys() {
                 if let Some(status) = compaction_guard.compaction_statuses.get(group_id) {
                     for (idx, level_handler) in status.level_handlers.iter().enumerate() {
-                        let tasks = level_handler.get_pending_tasks().to_vec();
+                        let tasks = level_handler.pending_tasks().to_vec();
                         if tasks.is_empty() {
                             continue;
                         }
diff --git a/src/meta/src/hummock/metrics_utils.rs b/src/meta/src/hummock/metrics_utils.rs
index a5b6aea1270f..924e858c3840 100644
--- a/src/meta/src/hummock/metrics_utils.rs
+++ b/src/meta/src/hummock/metrics_utils.rs
@@ -169,13 +169,13 @@ pub fn trigger_sst_stat(
             .with_label_values(&[&level_label])
             .set(level_sst_size(idx) as i64);
         if let Some(compact_status) = compact_status {
-            let compact_cnt = compact_status.level_handlers[idx].get_pending_file_count();
+            let compact_cnt = compact_status.level_handlers[idx].pending_file_count();
             metrics
                 .level_compact_cnt
                 .with_label_values(&[&level_label])
                 .set(compact_cnt as i64);
 
-            let compacting_task = compact_status.level_handlers[idx].get_pending_tasks();
+            let compacting_task = compact_status.level_handlers[idx].pending_tasks();
             let mut pending_task_ids: HashSet<u64> = HashSet::default();
             for task in compacting_task {
                 if pending_task_ids.contains(&task.task_id) {
@@ -298,7 +298,7 @@ pub fn trigger_sst_stat(
             for (idx, level_handler) in enumerate(compact_status.level_handlers.iter()) {
                 let sst_num = level_sst_cnt(idx);
                 let sst_size = level_sst_size(idx);
-                let compact_cnt = level_handler.get_pending_file_count();
+                let compact_cnt = level_handler.pending_file_count();
                 tracing::info!(
                     "Level {} has {} SSTs, the total size of which is {}KB, while {} of those are being compacted to bottom levels",
                     idx,


Test patch:
diff --git a/src/storage/hummock_test/src/hummock_storage_tests.rs b/src/storage/hummock_test/src/hummock_storage_tests.rs
index fa95238e49d2..74aac44a1eac 100644
--- a/src/storage/hummock_test/src/hummock_storage_tests.rs
+++ b/src/storage/hummock_test/src/hummock_storage_tests.rs
@@ -29,6 +29,7 @@ use risingwave_common::util::epoch::{test_epoch, EpochExt, INVALID_EPOCH};
 use risingwave_hummock_sdk::key::{
     gen_key_from_bytes, prefixed_range_with_vnode, FullKey, TableKey, UserKey, TABLE_PREFIX_LEN,
 };
+use risingwave_hummock_sdk::key_range::KeyRange;
 use risingwave_hummock_sdk::sstable_info::{SstableInfo, SstableInfoInner};
 use risingwave_hummock_sdk::table_stats::TableStats;
 use risingwave_hummock_sdk::table_watermark::{
@@ -2663,15 +2664,28 @@ async fn test_commit_multi_epoch() {
         };
 
     let epoch1 = initial_epoch.next_epoch();
-    let sst1_epoch1: SstableInfo = SstableInfoInner {
-        sst_id: 11,
-        object_id: 1,
-        table_ids: vec![existing_table_id.table_id],
-        file_size: 100,
-        sst_size: 100,
-        ..Default::default()
-    }
-    .into();
+    let sst1_epoch1: SstableInfo = {
+        let table_key_l = gen_key_from_str(VirtualNode::ZERO, "key_1");
+        let table_key_r = gen_key_from_str(VirtualNode::ZERO, "key_2");
+        SstableInfoInner {
+            sst_id: 11,
+            object_id: 1,
+            table_ids: vec![existing_table_id.table_id],
+            file_size: 100,
+            sst_size: 100,
+            key_range: KeyRange {
+                left: FullKey::for_test(existing_table_id, table_key_l, epoch1)
+                    .encode()
+                    .into(),
+                right: FullKey::for_test(existing_table_id, table_key_r, epoch1)
+                    .encode()
+                    .into(),
+                right_exclusive: false,
+            },
+            ..Default::default()
+        }
+        .into()
+    };
 
     commit_epoch(
         epoch1,
@@ -2705,17 +2719,29 @@ async fn test_commit_multi_epoch() {
         assert_eq!(cg_id, info.compaction_group_id);
     }
 
-    let sst1_epoch2: SstableInfo = SstableInfoInner {
-        sst_id: 22,
-        object_id: 2,
-        table_ids: vec![existing_table_id.table_id],
-        file_size: 100,
-        sst_size: 100,
-        ..Default::default()
-    }
-    .into();
-
     let epoch2 = epoch1.next_epoch();
+    let sst1_epoch2: SstableInfo = {
+        let table_key_l = gen_key_from_str(VirtualNode::ZERO, "key_1");
+        let table_key_r = gen_key_from_str(VirtualNode::ZERO, "key_2");
+        SstableInfoInner {
+            sst_id: 22,
+            object_id: 2,
+            table_ids: vec![existing_table_id.table_id],
+            file_size: 100,
+            sst_size: 100,
+            key_range: KeyRange {
+                left: FullKey::for_test(existing_table_id, table_key_l, epoch2)
+                    .encode()
+                    .into(),
+                right: FullKey::for_test(existing_table_id, table_key_r, epoch2)
+                    .encode()
+                    .into(),
+                right_exclusive: false,
+            },
+            ..Default::default()
+        }
+        .into()
+    };
 
     commit_epoch(epoch2, sst1_epoch2.clone(), vec![], &[existing_table_id]).await;
 
@@ -2742,15 +2768,28 @@ async fn test_commit_multi_epoch() {
 
     let new_table_id = TableId::new(2);
 
-    let sst2_epoch1: SstableInfo = SstableInfoInner {
-        sst_id: 33,
-        object_id: 3,
-        table_ids: vec![new_table_id.table_id],
-        file_size: 100,
-        sst_size: 100,
-        ..Default::default()
-    }
-    .into();
+    let sst2_epoch1: SstableInfo = {
+        let table_key_l = gen_key_from_str(VirtualNode::ZERO, "key_1");
+        let table_key_r = gen_key_from_str(VirtualNode::ZERO, "key_2");
+        SstableInfoInner {
+            sst_id: 33,
+            object_id: 3,
+            table_ids: vec![new_table_id.table_id],
+            file_size: 100,
+            sst_size: 100,
+            key_range: KeyRange {
+                left: FullKey::for_test(new_table_id, table_key_l, epoch1)
+                    .encode()
+                    .into(),
+                right: FullKey::for_test(new_table_id, table_key_r, epoch1)
+                    .encode()
+                    .into(),
+                right_exclusive: false,
+            },
+            ..Default::default()
+        }
+        .into()
+    };
 
     commit_epoch(
         epoch1,
@@ -2782,15 +2821,28 @@ async fn test_commit_multi_epoch() {
         new_cg_id
     };
 
-    let sst2_epoch2: SstableInfo = SstableInfoInner {
-        sst_id: 44,
-        object_id: 4,
-        table_ids: vec![new_table_id.table_id],
-        file_size: 100,
-        sst_size: 100,
-        ..Default::default()
-    }
-    .into();
+    let sst2_epoch2: SstableInfo = {
+        let table_key_l = gen_key_from_str(VirtualNode::ZERO, "key_1");
+        let table_key_r = gen_key_from_str(VirtualNode::ZERO, "key_2");
+        SstableInfoInner {
+            sst_id: 44,
+            object_id: 4,
+            table_ids: vec![new_table_id.table_id],
+            file_size: 100,
+            sst_size: 100,
+            key_range: KeyRange {
+                left: FullKey::for_test(new_table_id, table_key_l, epoch2)
+                    .encode()
+                    .into(),
+                right: FullKey::for_test(new_table_id, table_key_r, epoch2)
+                    .encode()
+                    .into(),
+                right_exclusive: false,
+            },
+            ..Default::default()
+        }
+        .into()
+    };
 
     commit_epoch(epoch2, sst2_epoch2.clone(), vec![], &[new_table_id]).await;
 
@@ -2814,15 +2866,28 @@ async fn test_commit_multi_epoch() {
 
     let epoch3 = epoch2.next_epoch();
 
-    let sst_epoch3: SstableInfo = SstableInfoInner {
-        sst_id: 55,
-        object_id: 5,
-        table_ids: vec![existing_table_id.table_id, new_table_id.table_id],
-        file_size: 100,
-        sst_size: 100,
-        ..Default::default()
-    }
-    .into();
+    let sst_epoch3: SstableInfo = {
+        let table_key_l = gen_key_from_str(VirtualNode::ZERO, "key_1");
+        let table_key_r = gen_key_from_str(VirtualNode::ZERO, "key_2");
+        SstableInfoInner {
+            sst_id: 55,
+            object_id: 5,
+            table_ids: vec![existing_table_id.table_id, new_table_id.table_id],
+            file_size: 100,
+            sst_size: 100,
+            key_range: KeyRange {
+                left: FullKey::for_test(existing_table_id, table_key_l, epoch3)
+                    .encode()
+                    .into(),
+                right: FullKey::for_test(new_table_id, table_key_r, epoch3)
+                    .encode()
+                    .into(),
+                right_exclusive: false,
+            },
+            ..Default::default()
+        }
+        .into()
+    };
 
     commit_epoch(
         epoch3,