resolve conflict

README.md

@@ -22,9 +22,9 @@ There are two ways to install RisingWave: use a pre-built package or compile fro
 
 ```shell
 # Download the pre-built binary
-wget https://github.com/risingwavelabs/risingwave/releases/download/v0.1.14/risingwave-v0.1.14-x86_64-unknown-linux.tar.gz
+wget https://github.com/risingwavelabs/risingwave/releases/download/v0.1.15/risingwave-v0.1.15-x86_64-unknown-linux.tar.gz
 # Unzip the binary
-tar xvf risingwave-v0.1.14-x86_64-unknown-linux.tar.gz
+tar xvf risingwave-v0.1.15-x86_64-unknown-linux.tar.gz
 # Start RisingWave in single-binary playground mode
 ./risingwave playground
 ```
@@ -33,7 +33,7 @@ tar xvf risingwave-v0.1.14-x86_64-unknown-linux.tar.gz
 
 ```shell
 # Start RisingWave in single-binary playground mode
-docker run -it --pull=always -p 4566:4566 -p 5691:5691 ghcr.io/risingwavelabs/risingwave:v0.1.14 playground
+docker run -it --pull=always -p 4566:4566 -p 5691:5691 ghcr.io/risingwavelabs/risingwave:v0.1.15 playground
 ```
 
 **Compile from Source with [RiseDev](docs/developer-guide.md#set-up-the-development-environment) (Linux and macOS)**

ci/workflows/main.yml

@@ -226,15 +226,15 @@ steps:
     retry: *auto-retry
 
   - label: "scaling test (deterministic simulation)"
-    command: "MADSIM_TEST_NUM=30 timeout 25m ci/scripts/deterministic-scale-test.sh"
+    command: "MADSIM_TEST_NUM=30 timeout 30m ci/scripts/deterministic-scale-test.sh"
     depends_on: "build-simulation"
     plugins:
       - gencer/cache#v2.4.10: *cargo-cache
       - docker-compose#v4.9.0:
           run: rw-build-env
           config: ci/docker-compose.yml
           mount-buildkite-agent: true
-    timeout_in_minutes: 30
+    timeout_in_minutes: 35
     retry: *auto-retry
 
   - label: "end-to-end test (deterministic simulation)"

e2e_test/batch/functions/now.slt.part

@@ -0,0 +1,17 @@
+statement ok
+create table tz (a timestamp with time zone);
+
+statement ok
+insert into tz values(now());
+
+statement ok
+create table t (a timestamp);
+
+statement ok
+insert into t values(now());
+
+statement ok
+drop table tz
+
+statement ok
+drop table t

e2e_test/batch/join/issue_7158.slt

@@ -0,0 +1,50 @@
+# https://github.com/risingwavelabs/risingwave/issues/7158
+
+statement ok
+CREATE TABLE t1 (x INT);
+
+statement ok
+CREATE TABLE t2 (x INT, y BIGINT, PRIMARY KEY (y));
+
+statement ok
+CREATE MATERIALIZED VIEW m AS SELECT 0::int AS x FROM t1;
+
+# row seq scan panics
+query I
+SELECT
+    1
+FROM
+    m JOIN t2 ON m.x = t2.y
+GROUP BY
+    m.x,
+    t2.y
+HAVING
+    1 = m.x;
+----
+
+# lookup join panics at runtime!
+statement ok
+INSERT INTO t1 VALUES (1);
+
+statement ok
+INSERT INTO t2 VALUES (1, 1);
+
+statement ok
+FLUSH;
+
+query I
+SELECT
+    1
+FROM
+    t1 JOIN t2 ON t1.x = t2.y AND t2.y=1;
+----
+1
+
+statement ok
+DROP MATERIALIZED VIEW m;
+
+statement ok
+DROP TABLE t2;
+
+statement ok
+DROP TABLE t1;

e2e_test/ddl/table.slt

@@ -126,6 +126,13 @@ statement error
 create table t (v1 varchar collate "en_US");
 
 # Test create-table-as 
+statement error alias must be specified
+create table t as select 1;
+
+# FIXME: this should be supported
+statement error alias must be specified
+create table t(v1) as select 1;
+
 statement ok
 create table t (v1 int,v2 int);
 
@@ -188,14 +195,8 @@ insert into t values (1);
 statement ok
 insert into t values (2);
 
-statement error 
-create table n1 as select sum(v1) from t;
-
-statement error 
-create table n1(v1) as select sum(v1) from t;
-
 statement ok
-create table n1 as select sum(v1) as sum from t;
+create table n1 as select sum(v1) from t;
 
 statement ok
 flush;

grafana/risingwave-dashboard.dashboard.py

@@ -1301,6 +1301,16 @@ def section_streaming_actors(outer_panels):
                         ),
                     ],
                 ),
+                panels.timeseries_percentage(
+                    "Join Actor Match Duration Per Second",
+                    "",
+                    [
+                        panels.target(
+                            f"rate({metric('stream_join_match_duration_ns')}[$__rate_interval]) / 1000000000",
+                            "{{actor_id}}.{{side}}",
+                        ),
+                    ],
+                ),
                 panels.timeseries_count(
                     "Join Cached Entries",
                     "",

proto/expr.proto

@@ -110,9 +110,9 @@ message ExprNode {
     ARRAY_CAT = 531;
     ARRAY_APPEND = 532;
     ARRAY_PREPEND = 533;
-    // Search operator and Search ARGument
-    SEARCH = 998;
-    SARG = 999;
+
+    // Non-pure functions below (> 600)
+    // ------------------------
     // Internal functions
     VNODE = 1101;
     // Non-deterministic functions

src/common/src/array/data_chunk.rs

@@ -484,6 +484,7 @@ impl DataChunkTestExt for DataChunk {
                 "F" => DataType::Float64,
                 "f" => DataType::Float32,
                 "TS" => DataType::Timestamp,
+                "TSZ" => DataType::Timestamptz,
                 "T" => DataType::Varchar,
                 array if array.starts_with('{') && array.ends_with('}') => {
                     DataType::Struct(Arc::new(StructType {

src/common/src/array/stream_chunk.rs

@@ -336,6 +336,7 @@ impl StreamChunkTestExt for StreamChunk {
     /// //     f: f32
     /// //     T: str
     /// //    TS: Timestamp
+    /// //   TSZ: Timestamptz
     /// // {i,f}: struct
     /// ```
     fn from_pretty(s: &str) -> Self {

src/ctl/Cargo.toml

@@ -39,6 +39,7 @@ tokio = { version = "0.2", package = "madsim-tokio", features = [
     "fs",
 ] }
 tracing = "0.1"
+uuid = { version = "1", features = ["v4"] }
 
 [target.'cfg(not(madsim))'.dependencies]
 workspace-hack = { path = "../workspace-hack" }

src/ctl/src/cmd_impl/bench.rs

@@ -25,7 +25,7 @@ use size::Size;
 use tokio::task::JoinHandle;
 
 use super::table::{get_table_catalog, make_state_table};
-use crate::common::HummockServiceOpts;
+use crate::CtlContext;
 
 #[derive(Subcommand)]
 pub enum BenchCommands {
@@ -71,9 +71,9 @@ impl InterestedMetrics {
     }
 }
 
-pub async fn do_bench(cmd: BenchCommands) -> Result<()> {
-    let mut hummock_opts = HummockServiceOpts::from_env()?;
-    let (meta, hummock, metrics) = hummock_opts.create_hummock_store_with_metrics().await?;
+pub async fn do_bench(context: &CtlContext, cmd: BenchCommands) -> Result<()> {
+    let meta = context.meta_client().await?;
+    let (hummock, metrics) = context.hummock_store_with_metrics().await?;
     let next_cnt = Arc::new(AtomicU64::new(0));
     let iter_cnt = Arc::new(AtomicU64::new(0));
     match cmd {
@@ -130,6 +130,5 @@ pub async fn do_bench(cmd: BenchCommands) -> Result<()> {
         }
     }
 
-    hummock_opts.shutdown().await;
     Ok(())
 }

src/ctl/src/cmd_impl/hummock/compaction_group.rs

@@ -15,22 +15,21 @@
 use risingwave_hummock_sdk::CompactionGroupId;
 use risingwave_pb::hummock::rise_ctl_update_compaction_config_request::mutable_config::MutableConfig;
 
-use crate::common::MetaServiceOpts;
+use crate::CtlContext;
 
-pub async fn list_compaction_group() -> anyhow::Result<()> {
-    let meta_opts = MetaServiceOpts::from_env()?;
-    let meta_client = meta_opts.create_meta_client().await?;
+pub async fn list_compaction_group(context: &CtlContext) -> anyhow::Result<()> {
+    let meta_client = context.meta_client().await?;
     let result = meta_client.risectl_list_compaction_group().await?;
     println!("{:#?}", result);
     Ok(())
 }
 
 pub async fn update_compaction_config(
+    context: &CtlContext,
     ids: Vec<CompactionGroupId>,
     configs: Vec<MutableConfig>,
 ) -> anyhow::Result<()> {
-    let meta_opts = MetaServiceOpts::from_env()?;
-    let meta_client = meta_opts.create_meta_client().await?;
+    let meta_client = context.meta_client().await?;
     meta_client
         .risectl_update_compaction_config(ids.as_slice(), configs.as_slice())
         .await?;

src/ctl/src/cmd_impl/hummock/disable_commit_epoch.rs

@@ -12,11 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::common::MetaServiceOpts;
+use crate::CtlContext;
 
-pub async fn disable_commit_epoch() -> anyhow::Result<()> {
-    let meta_opts = MetaServiceOpts::from_env()?;
-    let meta_client = meta_opts.create_meta_client().await?;
+pub async fn disable_commit_epoch(context: &CtlContext) -> anyhow::Result<()> {
+    let meta_client = context.meta_client().await?;
     let version = meta_client.disable_commit_epoch().await?;
     println!(
         "Disabled.\

src/ctl/src/cmd_impl/hummock/list_kv.rs

@@ -17,11 +17,10 @@ use core::ops::Bound::Unbounded;
 use risingwave_common::catalog::TableId;
 use risingwave_storage::store::{ReadOptions, StateStoreReadExt};
 
-use crate::common::HummockServiceOpts;
+use crate::CtlContext;
 
-pub async fn list_kv(epoch: u64, table_id: u32) -> anyhow::Result<()> {
-    let mut hummock_opts = HummockServiceOpts::from_env()?;
-    let (_meta_client, hummock) = hummock_opts.create_hummock_store().await?;
+pub async fn list_kv(context: &CtlContext, epoch: u64, table_id: u32) -> anyhow::Result<()> {
+    let hummock = context.hummock_store().await?;
     if epoch == u64::MAX {
         tracing::info!("using u64::MAX as epoch");
     }
@@ -47,6 +46,5 @@ pub async fn list_kv(epoch: u64, table_id: u32) -> anyhow::Result<()> {
         let print_string = format!("[t{}]", k.user_key.table_id.table_id());
         println!("{} {:?} => {:?}", print_string, k, v)
     }
-    hummock_opts.shutdown().await;
     Ok(())
 }

src/ctl/src/cmd_impl/hummock/list_version.rs

@@ -15,19 +15,17 @@
 use risingwave_pb::hummock::{PinnedSnapshotsSummary, PinnedVersionsSummary};
 use risingwave_rpc_client::HummockMetaClient;
 
-use crate::common::MetaServiceOpts;
+use crate::CtlContext;
 
-pub async fn list_version() -> anyhow::Result<()> {
-    let meta_opts = MetaServiceOpts::from_env()?;
-    let meta_client = meta_opts.create_meta_client().await?;
+pub async fn list_version(context: &CtlContext) -> anyhow::Result<()> {
+    let meta_client = context.meta_client().await?;
     let version = meta_client.get_current_version().await?;
     println!("{:#?}", version);
     Ok(())
 }
 
-pub async fn list_pinned_versions() -> anyhow::Result<()> {
-    let meta_opts = MetaServiceOpts::from_env()?;
-    let meta_client = meta_opts.create_meta_client().await?;
+pub async fn list_pinned_versions(context: &CtlContext) -> anyhow::Result<()> {
+    let meta_client = context.meta_client().await?;
     let PinnedVersionsSummary {
         mut pinned_versions,
         workers,
@@ -56,9 +54,8 @@ pub async fn list_pinned_versions() -> anyhow::Result<()> {
     Ok(())
 }
 
-pub async fn list_pinned_snapshots() -> anyhow::Result<()> {
-    let meta_opts = MetaServiceOpts::from_env()?;
-    let meta_client = meta_opts.create_meta_client().await?;
+pub async fn list_pinned_snapshots(context: &CtlContext) -> anyhow::Result<()> {
+    let meta_client = context.meta_client().await?;
     let PinnedSnapshotsSummary {
         mut pinned_snapshots,
         workers,

src/ctl/src/cmd_impl/hummock/list_version_deltas.rs

@@ -12,11 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::common::MetaServiceOpts;
+use crate::CtlContext;
 
-pub async fn list_version_deltas(start_id: u64, num_epochs: u32) -> anyhow::Result<()> {
-    let meta_opts = MetaServiceOpts::from_env()?;
-    let meta_client = meta_opts.create_meta_client().await?;
+pub async fn list_version_deltas(
+    context: &CtlContext,
+    start_id: u64,
+    num_epochs: u32,
+) -> anyhow::Result<()> {
+    let meta_client = context.meta_client().await?;
     let resp = meta_client
         .list_version_deltas(start_id, num_epochs, u64::MAX)
         .await?;