Merge branch 'main' into xx/lifetime

Makefile.toml

@@ -8,6 +8,7 @@ extend = [
   { path = "src/risedevtool/redis.toml" },
   { path = "src/risedevtool/risedev-components.toml" },
   { path = "src/frontend/planner_test/planner_test.toml" },
+  { path = "src/tests/compaction_test/Makefile.toml" },
 ]
 
 env_files = ["./risedev-components.user.env"]
@@ -447,15 +448,6 @@ cargo nextest run "$@"
 """
 description = "Run unit tests"
 
-[tasks.compaction-test]
-category = "RiseDev - Test"
-description = "Run hummock compaction e2e deterministic test"
-script = """
-#!/bin/bash
-set -e
-cargo run --bin compaction-test -- "$@"
-"""
-
 [tasks.sbuild]
 category = "RiseDev - Build in simulation mode"
 description = "Build in simulation mode"

ci/scripts/cron-e2e-test.sh

@@ -6,4 +6,5 @@ set -euo pipefail
 source ci/scripts/common.env.sh
 export RUN_SQLSMITH=1; # fuzz tests
 export SQLSMITH_COUNT=1000; # fuzz with 1000 queries
+export RUN_COMPACTION=1;
 source ci/scripts/run-e2e-test.sh

ci/scripts/pr.env.sh

@@ -11,6 +11,9 @@ MATCHES="ci/scripts/pr.env.sh\
 CHANGED=$(git diff --name-only origin/main | grep "$MATCHES")
 set -e
 
+# Don't run e2e compaction test in PR build
+export RUN_COMPACTION=0;
+
 if [[ -n "$CHANGED" ]]; then
     echo "Changes to Sqlsmith source files detected.";
     export RUN_SQLSMITH=1;

ci/scripts/run-e2e-test.sh

@@ -70,6 +70,35 @@ sqllogictest -p 4566 -d dev -e postgres-extended './e2e_test/extended_query/**/*
 echo "--- Kill cluster"
 cargo make ci-kill
 
+if [[ "$RUN_COMPACTION" -eq "1" ]]; then
+    echo "--- e2e, ci-compaction-test, nexmark_q7"
+    cargo make clean-data
+    cargo make ci-start ci-compaction-test
+    # Please make sure the regression is expected before increasing the timeout.
+    sqllogictest -p 4566 -d dev './e2e_test/compaction/ingest_rows.slt'
+
+    # We should ingest about 100 version deltas before the test
+    echo "--- Wait for data ingestion"
+    # Poll the current version id until we have around 100 version deltas
+    delta_log_cnt=0
+    while [ $delta_log_cnt -le 95 ]
+    do
+        delta_log_cnt="$(./risedev ctl hummock list-version | grep -w '^ *id:' | grep -o '[0-9]\+' | head -n 1)"
+        echo "Current version $delta_log_cnt"
+        sleep 5
+    done
+
+    echo "--- Pause source and disable commit new epochs"
+    ./risedev ctl meta pause
+    ./risedev ctl hummock disable-commit-epoch
+
+    echo "--- Start to run compaction test"
+    cargo run -r --bin compaction-test -- --ci-mode true --state-store hummock+minio://hummockadmin:hummockadmin@127.0.0.1:9301/hummock001
+
+    echo "--- Kill cluster"
+    cargo make ci-kill
+fi
+
 if [[ "$RUN_SQLSMITH" -eq "1" ]]; then
     echo "--- e2e, ci-3cn-1fe, fuzzing"
     buildkite-agent artifact download sqlsmith-"$profile" target/debug/

ci/workflows/main-cron.yml

@@ -21,15 +21,15 @@ auto-retry: &auto-retry
 
 steps:
   - label: "build"
-    command: "ci/scripts/build.sh -t ci-dev -p ci-dev"
+    command: "ci/scripts/build.sh -t ci-release -p ci-release"
     key: "build"
     plugins:
       - gencer/cache#v2.4.10: *cargo-cache
       - docker-compose#v3.9.0:
           run: rw-build-env
           config: ci/docker-compose.yml
           mount-buildkite-agent: true
-    timeout_in_minutes: 10
+    timeout_in_minutes: 20
     retry: *auto-retry
 
   - label: "build (deterministic simulation)"
@@ -56,8 +56,8 @@ steps:
     timeout_in_minutes: 10
     retry: *auto-retry
 
-  - label: "end-to-end test"
-    command: "ci/scripts/cron-e2e-test.sh -p ci-dev"
+  - label: "end-to-end test (release mode)"
+    command: "ci/scripts/cron-e2e-test.sh -p ci-release"
     depends_on:
       - "build"
       - "docslt"
@@ -71,8 +71,8 @@ steps:
     timeout_in_minutes: 30
     retry: *auto-retry
 
-  - label: "end-to-end source test"
-    command: "ci/scripts/e2e-source-test.sh -p ci-dev"
+  - label: "end-to-end source test (release mode)"
+    command: "ci/scripts/e2e-source-test.sh -p ci-release"
     depends_on: "build"
     plugins:
       - gencer/cache#v2.4.10: *cargo-cache

e2e_test/compaction/ingest_rows.slt

@@ -0,0 +1,49 @@
+statement ok
+CREATE SOURCE person
+    (id INTEGER, name VARCHAR, "email_address" VARCHAR, "credit_card" VARCHAR, city VARCHAR, state VARCHAR, "date_time" TIMESTAMP)
+with (
+  connector = 'nexmark',
+  nexmark.table.type = 'Person',
+  nexmark.split.num = '12',
+  nexmark.min.event.gap.in.ns = '0'
+) ROW FORMAT JSON;
+
+statement ok
+CREATE SOURCE auction (id INTEGER, "item_name" VARCHAR, description VARCHAR, "initial_bid" INTEGER, reserve INTEGER, "date_time" TIMESTAMP, expires TIMESTAMP, seller INTEGER, category INTEGER)
+with (
+  connector = 'nexmark',
+  nexmark.table.type = 'Auction',
+  nexmark.split.num = '12',
+  nexmark.min.event.gap.in.ns = '0'
+) ROW FORMAT JSON;
+
+statement ok
+CREATE SOURCE bid (auction INTEGER, bidder INTEGER, price INTEGER, "date_time" TIMESTAMP)
+with (
+  connector = 'nexmark',
+  nexmark.table.type = 'Bid',
+  nexmark.split.num = '12',
+  nexmark.min.event.gap.in.ns = '0'
+) ROW FORMAT JSON;
+
+statement ok
+CREATE MATERIALIZED VIEW nexmark_q7 AS
+SELECT
+  B.auction,
+  B.price,
+  B.bidder,
+  B.date_time
+FROM
+  bid B
+JOIN (
+  SELECT
+    MAX(price) AS maxprice,
+    window_end as date_time
+  FROM
+    TUMBLE(bid, date_time, INTERVAL '10' SECOND)
+  GROUP BY
+    window_end
+) B1 ON B.price = B1.maxprice
+WHERE
+  B.date_time BETWEEN B1.date_time - INTERVAL '10' SECOND
+  AND B1.date_time;

proto/stream_plan.proto

@@ -395,6 +395,14 @@ message ProjectSetNode {
   repeated expr.ProjectSetSelectItem select_list = 1;
 }
 
+// Sorts inputs and outputs ordered data based on watermark.
+message SortNode {
+  // Persists data above watermark.
+  catalog.Table state_table = 1;
+  // Column index of watermark to perform sorting.
+  uint32 sort_column_index = 2;
+}
+
 message StreamNode {
   oneof node_body {
     SourceNode source = 100;
@@ -422,6 +430,7 @@ message StreamNode {
     DynamicFilterNode dynamic_filter = 122;
     ProjectSetNode project_set = 123;
     GroupTopNNode group_top_n = 124;
+    SortNode sort = 125;
   }
   // The id for the operator. This is local per mview.
   // TODO: should better be a uint32.

risedev.yml

@@ -439,6 +439,19 @@ risedev:
     - use: compactor
     - use: redis
 
+  ci-compaction-test:
+    - use: minio
+    - use: etcd
+      unsafe-no-fsync: true
+    - use: meta-node
+      unsafe-disable-recovery: true
+      periodic-compaction-interval-sec: 99999
+      enable-compaction-deterministic: true
+    - use: compute-node
+      enable-tiered-cache: true
+    - use: frontend
+    - use: compactor
+
 compose:
   risingwave: "ghcr.io/risingwavelabs/risingwave:latest"
   prometheus: "prom/prometheus:latest"

src/bench/file_cache_bench/bench.rs

@@ -23,13 +23,14 @@ use prometheus::Registry;
 use rand::{Rng, SeedableRng};
 use risingwave_storage::hummock::file_cache::cache::{FileCache, FileCacheOptions};
 use risingwave_storage::hummock::file_cache::metrics::FileCacheMetrics;
+use risingwave_storage::hummock::file_cache::store::FsType;
 use risingwave_storage::hummock::{TieredCacheKey, TieredCacheValue};
 use tokio::sync::oneshot;
 use tracing::Instrument;
 
 use crate::analyze::{analyze, monitor, Hook, Metrics};
 use crate::rate::RateLimiter;
-use crate::utils::{dev_stat_path, iostat};
+use crate::utils::{dev_stat_path, file_stat_path, iostat};
 use crate::Args;
 
 #[derive(PartialEq, Eq, Hash, Clone, Debug)]
@@ -74,7 +75,17 @@ pub async fn run(args: Args, stop: oneshot::Receiver<()>) {
             .await
             .unwrap();
 
-    let iostat_path = dev_stat_path(&args.path);
+    let iostat_path = match cache.fs_type() {
+        FsType::Tmpfs => panic!("file cache bench is not supported on tmpfs"),
+        FsType::Btrfs => {
+            if args.iostat_dev.is_empty() {
+                panic!("cannot decide which block device to monitor for btrfs, please specify device name with \'--iostat-dev\'");
+            } else {
+                dev_stat_path(&args.iostat_dev)
+            }
+        }
+        _ => file_stat_path(&args.path),
+    };
 
     let iostat_start = iostat(&iostat_path);
     let metrics_dump_start = metrics.dump();

src/bench/file_cache_bench/main.rs

@@ -77,6 +77,10 @@ pub struct Args {
     /// Endpoint for jaeger, only valid when feature `trace` is enabled.
     #[clap(long, default_value = "http://127.0.0.1:14268/api/traces")]
     jaeger_endpoint: String,
+    /// Some filesystem (e.g. btrfs) can span across multiple block devices and it's hard to decide
+    /// which device to moitor. Use this argument to specify which block device to monitor.
+    #[clap(long, default_value = "")]
+    iostat_dev: String,
 }
 
 #[cfg(target_os = "linux")]

src/bench/file_cache_bench/utils.rs

@@ -20,7 +20,7 @@ use nix::sys::stat::stat;
 
 /// Given a normal file path, returns the containing block device static file path (of the
 /// partition).
-pub fn dev_stat_path(path: impl AsRef<Path>) -> PathBuf {
+pub fn file_stat_path(path: impl AsRef<Path>) -> PathBuf {
     let st_dev = stat(path.as_ref()).unwrap().st_dev;
 
     let major = unsafe { libc::major(st_dev) };
@@ -30,7 +30,10 @@ pub fn dev_stat_path(path: impl AsRef<Path>) -> PathBuf {
 
     let linkname = readlink(&dev).unwrap();
     let devname = Path::new(linkname.as_os_str()).file_name().unwrap();
+    dev_stat_path(devname.to_str().unwrap())
+}
 
+pub fn dev_stat_path(devname: &str) -> PathBuf {
     let classpath = Path::new("/sys/class/block").join(devname);
     let devclass = readlink(&classpath).unwrap();
 

src/cmd/Cargo.toml

@@ -54,7 +54,3 @@ path = "src/bin/compactor.rs"
 [[bin]]
 name = "risectl"
 path = "src/bin/ctl.rs"
-
-[[bin]]
-name = "compaction-test"
-path = "src/bin/compaction_test.rs"