Merge branch 'main' into xxh/commit_checkpoint_interval_sink_decouple

ci/build-ci-image.sh

@@ -10,7 +10,7 @@ cat ../rust-toolchain
 # shellcheck disable=SC2155
 
 # REMEMBER TO ALSO UPDATE ci/docker-compose.yml
-export BUILD_ENV_VERSION=v20240812
+export BUILD_ENV_VERSION=v20240911
 
 export BUILD_TAG="public.ecr.aws/w1p7b4n3/rw-build-env:${BUILD_ENV_VERSION}"
 

ci/docker-compose.yml

@@ -71,7 +71,7 @@ services:
       retries: 5
 
   source-test-env:
-    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240812
+    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240911
     depends_on:
       - mysql
       - sqlserver-server
@@ -85,7 +85,7 @@ services:
       - ..:/risingwave
 
   sink-test-env:
-    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240812
+    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240911
     depends_on:
       - mysql
       - db
@@ -108,12 +108,12 @@ services:
 
 
   rw-build-env:
-    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240812
+    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240911
     volumes:
       - ..:/risingwave
 
   ci-flamegraph-env:
-    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240812
+    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240911
     # NOTE(kwannoel): This is used in order to permit
     # syscalls for `nperf` (perf_event_open),
     # so it can do CPU profiling.
@@ -124,7 +124,7 @@ services:
       - ..:/risingwave
 
   regress-test-env:
-    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240812
+    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240911
     depends_on:
       db:
         condition: service_healthy

ci/rust-toolchain

@@ -4,4 +4,4 @@
 # 3. (optional) **follow the instructions in lints/README.md** to update the toolchain and dependencies for lints
 
 [toolchain]
-channel = "nightly-2024-06-06"
+channel = "nightly-2024-07-19"

lints/Cargo.toml

@@ -14,7 +14,7 @@ path = "ui/format_error.rs"
 # See `README.md` before bumping the version.
 # Remember to update the version in `ci/Dockerfile` as well.
 [dependencies]
-clippy_utils = { git = "https://github.com/risingwavelabs/clippy", rev = "5e2a7c6adebdb0478ee6d5b67ab4ee94153b2997" }
+clippy_utils = { git = "https://github.com/risingwavelabs/clippy", rev = "61e1d2fd7062e46ccf1237707ee6da5aac018f70" }
 dylint_linting = "3.1.0"
 itertools = "0.12"
 

lints/rust-toolchain

@@ -1,5 +1,5 @@
 # See `README.md` before bumping the version.
 
 [toolchain]
-channel = "nightly-2024-06-06"
+channel = "nightly-2024-07-19"
 components = ["llvm-tools-preview", "rustc-dev"]

src/batch/src/executor/join/hash_join.rs

@@ -162,9 +162,8 @@ impl<'a> Iterator for RowIdIter<'a> {
     type Item = RowId;
 
     fn next(&mut self) -> Option<Self::Item> {
-        self.current_row_id.map(|row_id| {
-            self.current_row_id = self.next_row_id[row_id];
-            row_id
+        self.current_row_id.inspect(|row_id| {
+            self.current_row_id = self.next_row_id[*row_id];
         })
     }
 }

src/batch/src/lib.rs

@@ -20,7 +20,6 @@
 #![feature(coroutines)]
 #![feature(proc_macro_hygiene, stmt_expr_attributes)]
 #![feature(iterator_try_collect)]
-#![feature(lint_reasons)]
 #![feature(is_sorted)]
 #![recursion_limit = "256"]
 #![feature(let_chains)]

src/common/benches/bench_data_chunk_encoding.rs

@@ -55,7 +55,7 @@ fn bench_data_chunk_encoding(c: &mut Criterion) {
         for null_ratio in NULL_RATIOS {
             for chunk_size in CHUNK_SIZES {
                 let chunk = rand_chunk::gen_chunk(&case.data_types, *chunk_size, SEED, *null_ratio);
-                let mut group = c.benchmark_group(&format!(
+                let mut group = c.benchmark_group(format!(
                     "data chunk encoding: {}, {} rows, Pr[null]={}",
                     case.name, chunk_size, null_ratio
                 ));

src/common/benches/bench_sequencer.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![feature(lint_reasons)]
-
 use std::cell::RefCell;
 use std::hint::black_box;
 use std::sync::atomic::{AtomicUsize, Ordering};

src/common/common_service/src/lib.rs

@@ -14,7 +14,6 @@
 
 // This is a stub lib.rs.
 
-#![feature(lint_reasons)]
 #![feature(impl_trait_in_assoc_type)]
 #![feature(error_generic_member_access)]
 

src/common/src/lib.rs

@@ -23,7 +23,6 @@
 #![feature(test)]
 #![feature(trusted_len)]
 #![feature(allocator_api)]
-#![feature(lint_reasons)]
 #![feature(coroutines)]
 #![feature(map_try_insert)]
 #![feature(error_generic_member_access)]

src/compute/src/lib.rs

@@ -16,7 +16,6 @@
 #![feature(coroutines)]
 #![feature(type_alias_impl_trait)]
 #![feature(let_chains)]
-#![feature(lint_reasons)]
 #![feature(impl_trait_in_assoc_type)]
 #![cfg_attr(coverage, feature(coverage_attribute))]
 
@@ -103,8 +102,9 @@ pub struct ComputeNodeOpts {
     pub role: Role,
 
     /// Used for control the metrics level, similar to log level.
-    /// 0 = disable metrics
-    /// >0 = enable metrics
+    ///
+    /// level = 0: disable metrics
+    /// level > 0: enable metrics
     #[clap(long, hide = true, env = "RW_METRICS_LEVEL")]
     #[override_opts(path = server.metrics_level)]
     pub metrics_level: Option<MetricLevel>,

src/connector/codec/src/lib.rs

@@ -21,7 +21,6 @@
 #![feature(stmt_expr_attributes)]
 #![feature(box_patterns)]
 #![feature(trait_alias)]
-#![feature(lint_reasons)]
 #![feature(let_chains)]
 #![feature(box_into_inner)]
 #![feature(type_alias_impl_trait)]

src/connector/src/lib.rs

@@ -19,7 +19,6 @@
 #![feature(stmt_expr_attributes)]
 #![feature(box_patterns)]
 #![feature(trait_alias)]
-#![feature(lint_reasons)]
 #![feature(let_chains)]
 #![feature(box_into_inner)]
 #![feature(type_alias_impl_trait)]

src/connector/src/parser/plain_parser.rs

@@ -297,10 +297,9 @@ mod tests {
             .unwrap()
             .into_iter()
             .filter(|c| c.cardinality() > 0)
-            .map(|c| {
+            .inspect(|c| {
                 // 5 data messages in a single chunk
                 assert_eq!(5, c.cardinality());
-                c
             })
             .collect_vec();
 

src/connector/src/sink/google_pubsub.rs

@@ -14,19 +14,15 @@
 
 use std::collections::BTreeMap;
 
-use anyhow::{anyhow, Context};
-use futures::future::try_join_all;
-use futures::prelude::future::FutureExt;
-use futures::prelude::TryFuture;
-use futures::TryFutureExt;
+use anyhow::anyhow;
 use google_cloud_gax::conn::Environment;
 use google_cloud_googleapis::pubsub::v1::PubsubMessage;
 use google_cloud_pubsub::apiv1;
 use google_cloud_pubsub::client::google_cloud_auth::credentials::CredentialsFile;
 use google_cloud_pubsub::client::google_cloud_auth::project;
 use google_cloud_pubsub::client::google_cloud_auth::token::DefaultTokenSourceProvider;
 use google_cloud_pubsub::client::{Client, ClientConfig};
-use google_cloud_pubsub::publisher::{Awaiter, Publisher};
+use google_cloud_pubsub::publisher::Publisher;
 use risingwave_common::array::StreamChunk;
 use risingwave_common::catalog::Schema;
 use serde_derive::Deserialize;
@@ -46,19 +42,33 @@ use crate::dispatch_sink_formatter_str_key_impl;
 pub const PUBSUB_SINK: &str = "google_pubsub";
 const PUBSUB_SEND_FUTURE_BUFFER_MAX_SIZE: usize = 65536;
 
-fn may_delivery_future(awaiter: Vec<Awaiter>) -> GooglePubSubSinkDeliveryFuture {
-    try_join_all(awaiter.into_iter().map(|awaiter| {
-        awaiter.get().map(|result| {
-            result
-                .context("Google Pub/Sub sink error")
-                .map_err(SinkError::GooglePubSub)
-                .map(|_| ())
-        })
-    }))
-    .map_ok(|_: Vec<()>| ())
-    .boxed()
+mod delivery_future {
+    use anyhow::Context;
+    use futures::future::try_join_all;
+    use futures::{FutureExt, TryFuture, TryFutureExt};
+    use google_cloud_pubsub::publisher::Awaiter;
+
+    use crate::sink::SinkError;
+
+    pub type GooglePubSubSinkDeliveryFuture =
+        impl TryFuture<Ok = (), Error = SinkError> + Unpin + 'static;
+
+    pub(super) fn may_delivery_future(awaiter: Vec<Awaiter>) -> GooglePubSubSinkDeliveryFuture {
+        try_join_all(awaiter.into_iter().map(|awaiter| {
+            awaiter.get().map(|result| {
+                result
+                    .context("Google Pub/Sub sink error")
+                    .map_err(SinkError::GooglePubSub)
+                    .map(|_| ())
+            })
+        }))
+        .map_ok(|_: Vec<()>| ())
+        .boxed()
+    }
 }
 
+use delivery_future::*;
+
 #[serde_as]
 #[derive(Clone, Debug, Deserialize, WithOptions)]
 pub struct GooglePubSubConfig {
@@ -172,9 +182,6 @@ struct GooglePubSubPayloadWriter<'w> {
     add_future: DeliveryFutureManagerAddFuture<'w, GooglePubSubSinkDeliveryFuture>,
 }
 
-pub type GooglePubSubSinkDeliveryFuture =
-    impl TryFuture<Ok = (), Error = SinkError> + Unpin + 'static;
-
 impl GooglePubSubSinkWriter {
     pub async fn new(
         config: GooglePubSubConfig,

src/connector/src/sink/iceberg/prometheus/monitored_partition_writer.rs

@@ -27,7 +27,6 @@ pub struct MonitoredFanoutPartitionedWriterBuilder<B: IcebergWriterBuilder> {
 }
 
 impl<B: IcebergWriterBuilder> MonitoredFanoutPartitionedWriterBuilder<B> {
-    #[expect(dead_code)]
     pub fn new(
         inner: FanoutPartitionedWriterBuilder<B>,
         partition_num: LabelGuardedIntGauge<2>,

src/connector/src/sink/iceberg/prometheus/monitored_write_writer.rs

@@ -28,7 +28,6 @@ pub struct MonitoredWriteWriterBuilder<B: IcebergWriterBuilder> {
 
 impl<B: IcebergWriterBuilder> MonitoredWriteWriterBuilder<B> {
     /// Create writer context.
-    #[expect(dead_code)]
     pub fn new(
         inner: B,
         write_qps: LabelGuardedIntCounter<2>,

src/connector/src/sink/redis.rs

@@ -288,7 +288,7 @@ impl RedisSinkPayloadWriter {
                 return Ok(());
             }
         }
-        self.pipe.query(self.conn.as_mut().unwrap()).await?;
+        self.pipe.query::<()>(self.conn.as_mut().unwrap()).await?;
         self.pipe.clear();
         Ok(())
     }

src/connector/src/sink/remote.rs

@@ -23,7 +23,6 @@ use async_trait::async_trait;
 use await_tree::InstrumentAwait;
 use futures::future::select;
 use futures::TryStreamExt;
-use itertools::Itertools;
 use jni::JavaVM;
 use prost::Message;
 use risingwave_common::array::StreamChunk;
@@ -174,7 +173,7 @@ async fn validate_remote_sink(param: &SinkParam, sink_name: &str) -> ConnectorRe
         bail!("Es sink only supports single pk or pk with delimiter option");
     }
     // FIXME: support struct and array in stream sink
-    param.columns.iter().map(|col| {
+    param.columns.iter().try_for_each(|col| {
         match &col.data_type {
             DataType::Int16
                     | DataType::Int32
@@ -217,7 +216,7 @@ async fn validate_remote_sink(param: &SinkParam, sink_name: &str) -> ConnectorRe
                             "remote sink supports Int16, Int32, Int64, Float32, Float64, Boolean, Decimal, Time, Date, Interval, Jsonb, Timestamp, Timestamptz, Bytea, List and Varchar, (Es sink support Struct) got {:?}: {:?}",
                             col.name,
                             col.data_type,
-                        )))}}).try_collect()?;
+                        )))}})?;
 
     let jvm = JVM.get_or_init()?;
     let sink_param = param.to_proto();

src/dml/src/lib.rs

@@ -14,7 +14,6 @@
 
 #![allow(clippy::derive_partial_eq_without_eq)]
 #![feature(trait_alias)]
-#![feature(lint_reasons)]
 #![feature(coroutines)]
 #![feature(hash_extract_if)]
 #![feature(type_alias_impl_trait)]

src/error/src/lib.rs

@@ -21,7 +21,6 @@
 //! access if `risingwave_common` is already a dependency.
 
 #![feature(error_generic_member_access)]
-#![feature(lint_reasons)]
 #![feature(register_tool)]
 #![register_tool(rw)]
 #![feature(trait_alias)]

src/expr/core/src/lib.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 #![feature(let_chains)]
-#![feature(lint_reasons)]
 #![feature(iterator_try_collect)]
 #![feature(coroutines)]
 #![feature(never_type)]

src/expr/impl/src/lib.rs

@@ -23,7 +23,6 @@
 #![allow(non_snake_case)] // for `ctor` generated code
 #![feature(let_chains)]
 #![feature(assert_matches)]
-#![feature(lint_reasons)]
 #![feature(iterator_try_collect)]
 #![feature(coroutines)]
 #![feature(test)]

src/expr/macro/src/lib.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![feature(lint_reasons)]
 #![feature(let_chains)]
 
 use std::vec;