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/.gitignore b/.gitignore
index 81e8609f3334..486d0938034c 100644
--- a/.gitignore
+++ b/.gitignore
@@ -89,3 +89,5 @@ e2e_test/iceberg/metastore_db
 
 # mdbook
 book
+
+dhat-heap.json
diff --git a/Cargo.lock b/Cargo.lock
index cb581e4ac974..7afccde408ae 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4242,6 +4242,22 @@ dependencies = [
  "syn 2.0.87",
 ]
 
+[[package]]
+name = "dhat"
+version = "0.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "98cd11d84628e233de0ce467de10b8633f4ddaecafadefc86e13b84b8739b827"
+dependencies = [
+ "backtrace",
+ "lazy_static",
+ "mintex",
+ "parking_lot 0.12.1",
+ "rustc-hash",
+ "serde",
+ "serde_json",
+ "thousands",
+]
+
 [[package]]
 name = "dialoguer"
 version = "0.11.0"
@@ -7589,6 +7605,12 @@ dependencies = [
  "web-time",
 ]
 
+[[package]]
+name = "mintex"
+version = "0.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9bec4598fddb13cc7b528819e697852653252b760f1228b7642679bf2ff2cd07"
+
 [[package]]
 name = "mio"
 version = "0.8.11"
@@ -12154,6 +12176,7 @@ dependencies = [
  "cfg-if",
  "criterion",
  "delta_btree_map",
+ "dhat",
  "educe",
  "either",
  "enum-as-inner 0.6.0",
@@ -14344,6 +14367,12 @@ dependencies = [
  "syn 2.0.87",
 ]
 
+[[package]]
+name = "thousands"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3bf63baf9f5039dadc247375c29eb13706706cfde997d0330d05aa63a77d8820"
+
 [[package]]
 name = "thread-id"
 version = "4.2.0"
diff --git a/src/common/src/catalog/schema.rs b/src/common/src/catalog/schema.rs
index 9eccecfc2fc0..221d8aed5025 100644
--- a/src/common/src/catalog/schema.rs
+++ b/src/common/src/catalog/schema.rs
@@ -34,6 +34,17 @@ pub struct Field {
     pub type_name: String,
 }
 
+impl Field {
+    pub fn new(name: impl Into<String>, data_type: DataType) -> Self {
+        Self {
+            data_type,
+            name: name.into(),
+            sub_fields: vec![],
+            type_name: String::new(),
+        }
+    }
+}
+
 impl std::fmt::Debug for Field {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "{}:{:?}", self.name, self.data_type)
diff --git a/src/stream/Cargo.toml b/src/stream/Cargo.toml
index 0a8738b96a5d..3c30e0fccf6d 100644
--- a/src/stream/Cargo.toml
+++ b/src/stream/Cargo.toml
@@ -90,6 +90,7 @@ workspace-hack = { path = "../workspace-hack" }
 [dev-dependencies]
 assert_matches = "1"
 criterion = { workspace = true, features = ["async_tokio", "async"] }
+dhat =  "0.3"
 expect-test = "1"
 risingwave_expr_impl = { workspace = true }
 risingwave_hummock_sdk = { workspace = true }
@@ -99,6 +100,9 @@ risingwave_hummock_test = { path = "../storage/hummock_test", features = [
 serde_yaml = "0.9"
 tracing-test = "0.2"
 
+[features]
+dhat-heap = []
+
 [[bench]]
 name = "stream_hash_agg"
 harness = false
@@ -107,5 +111,13 @@ harness = false
 name = "bench_state_table"
 harness = false
 
+[[bench]]
+name = "stream_hash_join_rt"
+harness = false
+
+[[bench]]
+name = "stream_hash_join_mem"
+harness = false
+
 [lints]
 workspace = true
diff --git a/src/stream/benches/stream_hash_join_mem.rs b/src/stream/benches/stream_hash_join_mem.rs
new file mode 100644
index 000000000000..7f4275162ad9
--- /dev/null
+++ b/src/stream/benches/stream_hash_join_mem.rs
@@ -0,0 +1,55 @@
+// Copyright 2024 RisingWave Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#![feature(let_chains)]
+
+//! To run this benchmark you can use the following command:
+//! ```sh
+//! cargo bench --features dhat-heap --bench stream_hash_join
+//! ```
+//!
+//! You may also specify the amplification size, e.g. 40000
+//! ```sh
+//! cargo bench --features dhat-heap --bench stream_hash_join -- 40000
+//! ```
+
+use std::env;
+
+use risingwave_stream::executor::test_utils::hash_join_executor::*;
+
+risingwave_expr_impl::enable!();
+
+#[cfg(feature = "dhat-heap")]
+#[global_allocator]
+static ALLOC: dhat::Alloc = dhat::Alloc;
+
+#[tokio::main]
+async fn main() {
+    let args: Vec<_> = env::args().collect();
+    let amp = if let Some(raw_arg) = args.get(1)
+        && let Ok(arg) = raw_arg.parse()
+    {
+        arg
+    } else {
+        100_000
+    };
+    let (tx_l, tx_r, out) = setup_bench_stream_hash_join(amp).await;
+    {
+        // Start the profiler later, after we have ingested the data for hash join build-side.
+        #[cfg(feature = "dhat-heap")]
+        let _profiler = dhat::Profiler::new_heap();
+
+        handle_streams(amp, tx_l, tx_r, out).await;
+    }
+}
diff --git a/src/stream/benches/stream_hash_join_rt.rs b/src/stream/benches/stream_hash_join_rt.rs
new file mode 100644
index 000000000000..5beed531575e
--- /dev/null
+++ b/src/stream/benches/stream_hash_join_rt.rs
@@ -0,0 +1,47 @@
+// Copyright 2024 RisingWave Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#![feature(let_chains)]
+
+//! To run this benchmark you can use the following command:
+//! ```sh
+//! cargo bench --bench stream_hash_join_rt
+//! ```
+
+use criterion::{criterion_group, criterion_main, BatchSize, Criterion};
+use futures::executor::block_on;
+use risingwave_stream::executor::test_utils::hash_join_executor::*;
+use tokio::runtime::Runtime;
+
+risingwave_expr_impl::enable!();
+
+fn bench_hash_join(c: &mut Criterion) {
+    let mut group = c.benchmark_group("benchmark_hash_join");
+    group.sample_size(10);
+
+    let rt = Runtime::new().unwrap();
+    for amp in [10_000, 20_000, 30_000, 40_000, 100_000, 200_000, 400_000] {
+        let name = format!("hash_join_rt_{}", amp);
+        group.bench_function(&name, |b| {
+            b.to_async(&rt).iter_batched(
+                || block_on(setup_bench_stream_hash_join(amp)),
+                |(tx_l, tx_r, out)| handle_streams(amp, tx_l, tx_r, out),
+                BatchSize::SmallInput,
+            )
+        });
+    }
+}
+
+criterion_group!(benches, bench_hash_join);
+criterion_main!(benches);


Test patch:
diff --git a/src/stream/src/executor/test_utils.rs b/src/stream/src/executor/test_utils.rs
index 76301a2de509..80fca8d5583e 100644
--- a/src/stream/src/executor/test_utils.rs
+++ b/src/stream/src/executor/test_utils.rs
@@ -624,3 +624,217 @@ pub mod top_n_executor {
         .await
     }
 }
+
+pub mod hash_join_executor {
+    use std::sync::atomic::AtomicU64;
+    use std::sync::Arc;
+
+    use itertools::Itertools;
+    use risingwave_common::array::{I64Array, Op, StreamChunkTestExt};
+    use risingwave_common::catalog::{ColumnDesc, ColumnId, Field, TableId};
+    use risingwave_common::hash::Key128;
+    use risingwave_common::util::sort_util::OrderType;
+    use risingwave_storage::memory::MemoryStateStore;
+
+    use super::*;
+    use crate::common::table::test_utils::gen_pbtable;
+    use crate::executor::monitor::StreamingMetrics;
+    use crate::executor::prelude::StateTable;
+    use crate::executor::test_utils::{MessageSender, MockSource};
+    use crate::executor::{ActorContext, HashJoinExecutor, JoinParams, JoinType};
+
+    pub async fn create_in_memory_state_table(
+        mem_state: MemoryStateStore,
+        data_types: &[DataType],
+        order_types: &[OrderType],
+        pk_indices: &[usize],
+        table_id: u32,
+    ) -> (StateTable<MemoryStateStore>, StateTable<MemoryStateStore>) {
+        let column_descs = data_types
+            .iter()
+            .enumerate()
+            .map(|(id, data_type)| ColumnDesc::unnamed(ColumnId::new(id as i32), data_type.clone()))
+            .collect_vec();
+        let state_table = StateTable::from_table_catalog(
+            &gen_pbtable(
+                TableId::new(table_id),
+                column_descs,
+                order_types.to_vec(),
+                pk_indices.to_vec(),
+                0,
+            ),
+            mem_state.clone(),
+            None,
+        )
+        .await;
+
+        // Create degree table
+        let mut degree_table_column_descs = vec![];
+        pk_indices.iter().enumerate().for_each(|(pk_id, idx)| {
+            degree_table_column_descs.push(ColumnDesc::unnamed(
+                ColumnId::new(pk_id as i32),
+                data_types[*idx].clone(),
+            ))
+        });
+        degree_table_column_descs.push(ColumnDesc::unnamed(
+            ColumnId::new(pk_indices.len() as i32),
+            DataType::Int64,
+        ));
+        let degree_state_table = StateTable::from_table_catalog(
+            &gen_pbtable(
+                TableId::new(table_id + 1),
+                degree_table_column_descs,
+                order_types.to_vec(),
+                pk_indices.to_vec(),
+                0,
+            ),
+            mem_state,
+            None,
+        )
+        .await;
+        (state_table, degree_state_table)
+    }
+
+    /// 1. Refill state table of build side.
+    /// 2. Init executor.
+    /// 3. Push data to the probe side.
+    /// 4. Check memory utilization.
+    pub async fn setup_bench_stream_hash_join(
+        amp: usize,
+    ) -> (MessageSender, MessageSender, BoxedMessageStream) {
+        let fields = vec![DataType::Int64, DataType::Int64, DataType::Int64];
+        let orders = vec![OrderType::ascending(), OrderType::ascending()];
+        let state_store = MemoryStateStore::new();
+
+        // Probe side
+        let (lhs_state_table, lhs_degree_state_table) =
+            create_in_memory_state_table(state_store.clone(), &fields, &orders, &[0, 1], 0).await;
+
+        // Build side
+        let (mut rhs_state_table, rhs_degree_state_table) =
+            create_in_memory_state_table(state_store.clone(), &fields, &orders, &[0, 1], 2).await;
+
+        // Insert 100K records into the build side.
+        {
+            // Create column [0]: join key. Each record has the same value, to trigger join amplification.
+            let mut int64_jk_builder = DataType::Int64.create_array_builder(amp);
+            int64_jk_builder
+                .append_array(&I64Array::from_iter(vec![Some(200_000); amp].into_iter()).into());
+            let jk = int64_jk_builder.finish();
+
+            // Create column [1]: pk. The original pk will be here, it will be unique.
+            let mut int64_pk_data_chunk_builder = DataType::Int64.create_array_builder(amp);
+            let seq = I64Array::from_iter((0..amp as i64).map(Some));
+            int64_pk_data_chunk_builder.append_array(&I64Array::from(seq).into());
+            let pk = int64_pk_data_chunk_builder.finish();
+
+            // Create column [2]: value. This can be an arbitrary value, so just clone the pk column.
+            let values = pk.clone();
+
+            // Build the stream chunk.
+            let columns = vec![jk.into(), pk.into(), values.into()];
+            let ops = vec![Op::Insert; amp];
+            let stream_chunk = StreamChunk::new(ops, columns);
+
+            // Write to state table.
+            rhs_state_table.write_chunk(stream_chunk);
+        }
+
+        let schema = Schema::new(fields.iter().cloned().map(Field::unnamed).collect());
+
+        let (tx_l, source_l) = MockSource::channel();
+        let source_l = source_l.into_executor(schema.clone(), vec![1]);
+        let (tx_r, source_r) = MockSource::channel();
+        let source_r = source_r.into_executor(schema, vec![1]);
+
+        // Schema is the concatenation of the two source schemas.
+        // [lhs(jk):0, lhs(pk):1, lhs(value):2, rhs(jk):0, rhs(pk):1, rhs(value):2]
+        // [0,         1,         2,            3,         4,         5           ]
+        let schema: Vec<_> = [source_l.schema().fields(), source_r.schema().fields()]
+            .concat()
+            .into_iter()
+            .collect();
+        let schema_len = schema.len();
+        let info = ExecutorInfo {
+            schema: Schema { fields: schema },
+            pk_indices: vec![0, 1, 3, 4],
+            identity: "HashJoinExecutor".to_string(),
+        };
+
+        // join-key is [0], primary-key is [1].
+        let params_l = JoinParams::new(vec![0], vec![1]);
+        let params_r = JoinParams::new(vec![0], vec![1]);
+
+        let executor = HashJoinExecutor::<Key128, MemoryStateStore, { JoinType::Inner }>::new(
+            ActorContext::for_test(123),
+            info,
+            source_l,
+            source_r,
+            params_l,
+            params_r,
+            vec![false], // null-safe
+            (0..schema_len).collect_vec(),
+            None,   // condition, it is an eq join, we have no condition
+            vec![], // ineq pairs
+            lhs_state_table,
+            lhs_degree_state_table,
+            rhs_state_table,
+            rhs_degree_state_table,
+            Arc::new(AtomicU64::new(0)), // watermark epoch
+            false,                       // is_append_only
+            Arc::new(StreamingMetrics::unused()),
+            1024, // chunk_size
+            2048, // high_join_amplification_threshold
+        );
+        (tx_l, tx_r, executor.boxed().execute())
+    }
+
+    pub async fn handle_streams(
+        amp: usize,
+        mut tx_l: MessageSender,
+        mut tx_r: MessageSender,
+        mut stream: BoxedMessageStream,
+    ) {
+        // Init executors
+        tx_l.push_barrier(test_epoch(1), false);
+        tx_r.push_barrier(test_epoch(1), false);
+        // Push a single record into tx_l, matches 100K records in the build side.
+        let chunk = StreamChunk::from_pretty(
+            "  I      I I
+         + 200000 0 1",
+        );
+        tx_l.push_chunk(chunk);
+
+        match stream.next().await {
+            Some(Ok(Message::Barrier(b))) => {
+                assert_eq!(b.epoch.curr, test_epoch(1));
+            }
+            other => {
+                panic!("Expected a barrier, got {:?}", other);
+            }
+        }
+
+        let chunks = amp / 1024;
+        let remainder = amp % 1024;
+
+        for _ in 0..chunks {
+            match stream.next().await {
+                Some(Ok(Message::Chunk(c))) => {
+                    assert_eq!(c.cardinality(), 1024);
+                }
+                other => {
+                    panic!("Expected a barrier, got {:?}", other);
+                }
+            }
+        }
+
+        match stream.next().await {
+            Some(Ok(Message::Chunk(c))) => {
+                assert_eq!(c.cardinality(), remainder);
+            }
+            other => {
+                panic!("Expected a barrier, got {:?}", other);
+            }
+        }
+    }
+}