chore(cherry-pick): cherry-pick #7562 #7539 #7482 (#7582)

* fix(frontend): remove alias check (#7562)

In the old version, we guarantee every column with a alias name, these behaviour caused the bug #7549.

### And why we have this guarantee:
At that time, function didn't have a alias, to guarantee every function column have a column name, we enforce the user must specify column alias.(A rude solution) #6798

But now, we can derive function alias so that I think we can remove the alias check.

After remove it, our behaviour will be compatiable with postgres:

```
test_db=> create table t1 as select 1;
SELECT 1
test_db=> select * from t1;
?column?
----------
1
(1 row)

test_db=> create table t as select 1,2,3;
ERROR:  column "?column?" specified more than once

test_db=> create table t as select 1 as a , 2 as b;
SELECT 1
test_db=> select * from t;
a | b
---+---
1 | 2
(1 row)

```

Approved-By: st1page

* fix: move install mysql-client to ci image (#7539)

fix main

aca7030

Approved-By: jon-chuang

* build(*): bump toolchain to 2023-01-18 (#7482)

* build(toolchain): bump toolchain to 2023-01-18

Signed-off-by: TennyZhuang <[email protected]>

* revert deps

Signed-off-by: TennyZhuang <[email protected]>

* update futures-async-stream

Signed-off-by: TennyZhuang <[email protected]>

Signed-off-by: TennyZhuang <[email protected]>

---------

Signed-off-by: TennyZhuang <[email protected]>
Co-authored-by: ZENOTME <[email protected]>
Co-authored-by: Bohan Zhang <[email protected]>
Co-authored-by: TennyZhuang <[email protected]>

Author: 4 people

Cargo.lock

@@ -6,7 +6,7 @@ ARG RUST_TOOLCHAIN
 
 RUN apt-get update -yy && \
     DEBIAN_FRONTEND=noninteractive apt-get -y install make build-essential cmake protobuf-compiler curl parallel \
-    openssl libssl-dev libsasl2-dev libcurl4-openssl-dev pkg-config bash openjdk-11-jdk wget unzip git tmux lld postgresql-client kafkacat \
+    openssl libssl-dev libsasl2-dev libcurl4-openssl-dev pkg-config bash openjdk-11-jdk wget unzip git tmux lld postgresql-client kafkacat netcat mysql-client \
     maven -yy \
     && rm -rf /var/lib/{apt,dpkg,cache,log}/
 

ci/Dockerfile

@@ -14,7 +14,7 @@ export RUST_TOOLCHAIN=$(cat ../rust-toolchain)
 # !!! CHANGE THIS WHEN YOU WANT TO BUMP CI IMAGE !!! #
 #          AND ALSO docker-compose.yml               #
 ######################################################
-export BUILD_ENV_VERSION=v20230116
+export BUILD_ENV_VERSION=v20230126
 
 export BUILD_TAG="public.ecr.aws/x5u3w5h6/rw-build-env:${BUILD_ENV_VERSION}"
 

ci/build-ci-image.sh

@@ -30,28 +30,28 @@ services:
       retries: 5
 
   source-test-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230116
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230126
     depends_on:
       - mysql
       - db
     volumes:
       - ..:/risingwave
 
   sink-test-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230116
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230126
     depends_on:
       - mysql
       - db
     volumes:
       - ..:/risingwave
 
   rw-build-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230116
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230126
     volumes:
       - ..:/risingwave
 
   regress-test-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230116
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230126
     depends_on:
       db:
         condition: service_healthy

ci/docker-compose.yml

@@ -43,7 +43,6 @@ cargo make pre-start-dev
 cargo make link-all-in-one-binaries
 
 # prepare environment mysql sink
-apt-get -y install mysql-client
 mysql --host=mysql --port=3306 -u root -p123456 -e "CREATE DATABASE IF NOT EXISTS test;"
 # grant access to `test` for ci test user
 mysql --host=mysql --port=3306 -u root -p123456 -e "GRANT ALL PRIVILEGES ON test.* TO 'mysqluser'@'%';"

ci/scripts/e2e-sink-test.sh

@@ -51,8 +51,7 @@ cargo make pre-start-dev
 cargo make link-all-in-one-binaries
 
 echo "--- e2e, ci-1cn-1fe, mysql & postgres cdc"
-# install mysql client
-apt-get -y install mysql-client
+
 # import data to mysql
 mysql --host=mysql --port=3306 -u root -p123456 < ./e2e_test/source/cdc/mysql_cdc.sql
 

ci/scripts/e2e-source-test.sh

@@ -126,13 +126,27 @@ statement error
 create table t (v1 varchar collate "en_US");
 
 # Test create-table-as 
-statement error alias must be specified
+statement ok
 create table t as select 1;
 
-# FIXME: this should be supported
-statement error alias must be specified
+statement ok
+drop table t;
+
+statement error
+create table t as select 1,2;
+
+statement ok
+create table t as select 1 as a, 2 as b;
+
+statement ok
+drop table t;
+
+statement ok 
 create table t(v1) as select 1;
 
+statement ok
+drop table t;
+
 statement ok
 create table t (v1 int,v2 int);
 

e2e_test/ddl/table.slt

@@ -1 +1 @@
-nightly-2022-12-12
+nightly-2023-01-18

rust-toolchain

@@ -553,7 +553,7 @@ mod tests {
             lookup_prefix_len: 1,
             chunk_builder: DataChunkBuilder::new(original_schema.data_types(), CHUNK_SIZE),
             schema: original_schema.clone(),
-            output_indices: (0..original_schema.len()).into_iter().collect(),
+            output_indices: (0..original_schema.len()).collect(),
             chunk_size: CHUNK_SIZE,
             identity: "TestLookupJoinExecutor".to_string(),
         }

src/batch/src/executor/join/local_lookup_join.rs

@@ -33,9 +33,10 @@ use risingwave_common::types::{DataType, DatumRef};
 use risingwave_pb::plan_common::JoinType as JoinTypeProst;
 
 use crate::error::BatchError;
-use crate::executor::join::JoinType::Inner;
-#[derive(Copy, Clone, Debug, PartialEq)]
+
+#[derive(Copy, Clone, Debug, Default, PartialEq)]
 pub enum JoinType {
+    #[default]
     Inner,
     LeftOuter,
     /// Semi join when probe side should output when matched
@@ -103,12 +104,6 @@ impl JoinType {
     }
 }
 
-impl Default for JoinType {
-    fn default() -> Self {
-        Inner
-    }
-}
-
 /// The layout be like:
 ///
 /// [ `left` chunk     |  `right` chunk     ]

src/batch/src/executor/join/mod.rs

@@ -166,7 +166,6 @@ mod tests {
     ) {
         let col = create_column(
             (0..row_num)
-                .into_iter()
                 .map(|x| Some(x as i32))
                 .collect_vec()
                 .as_slice(),
@@ -281,7 +280,6 @@ mod tests {
         assert_eq!(visible.len(), row_num);
         let col0 = create_column(
             (0..row_num)
-                .into_iter()
                 .map(|x| Some(x as i32))
                 .collect_vec()
                 .as_slice(),

src/batch/src/executor/limit.rs

@@ -225,7 +225,6 @@ async fn multi_part_upload(
     };
     let chunks = obj
         .chunks(part_size.as_u64() as usize)
-        .into_iter()
         .map(|chunk| chunk.to_owned())
         .collect_vec();
 

src/bench/s3_bench/main.rs

@@ -344,11 +344,7 @@ impl DataChunk {
     /// `[c, b, a]`. If `column_mapping` is [2, 0], then the output will be `[c, a]`
     /// If the input mapping is identity mapping, no reorder will be performed.
     pub fn reorder_columns(self, column_mapping: &[usize]) -> Self {
-        if column_mapping
-            .iter()
-            .copied()
-            .eq((0..self.columns().len()).into_iter())
-        {
+        if column_mapping.iter().copied().eq(0..self.columns().len()) {
             return self;
         }
         let mut new_columns = Vec::with_capacity(column_mapping.len());

src/common/src/array/data_chunk.rs

@@ -254,7 +254,7 @@ impl StreamChunk {
         if column_mapping
             .iter()
             .copied()
-            .eq((0..self.data.columns().len()).into_iter())
+            .eq(0..self.data.columns().len())
         {
             // no reorder is needed
             self

src/common/src/array/stream_chunk.rs

@@ -61,17 +61,13 @@ pub trait NumericFieldSequenceGenerator {
 }
 
 /// the way that datagen create the field data. such as 'sequence' or 'random'.
+#[derive(Default)]
 pub enum FieldKind {
     Sequence,
+    #[default]
     Random,
 }
 
-impl Default for FieldKind {
-    fn default() -> Self {
-        FieldKind::Random
-    }
-}
-
 pub enum FieldGeneratorImpl {
     I16Sequence(I16SequenceField),
     I32Sequence(I32SequenceField),

src/common/src/field_generator/mod.rs

@@ -802,7 +802,6 @@ macro_rules! scalar_impl_hash {
             }
         }
 
-        #[expect(clippy::derive_hash_xor_eq)]
         impl Hash for ScalarImpl {
             fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
                 match self {

src/common/src/types/mod.rs

@@ -84,7 +84,7 @@ enum ResponseStreamImpl {
 
 impl ResponseStreamImpl {
     pub async fn next(&mut self) -> Result<SinkResponse> {
-        return match self {
+        match self {
             ResponseStreamImpl::Grpc(ref mut response) => response
                 .next()
                 .await
@@ -95,7 +95,7 @@ impl ResponseStreamImpl {
                     SinkError::Remote("response stream closed unexpectedly".to_string())
                 })
             }
-        };
+        }
     }
 }
 

src/connector/src/sink/remote.rs

@@ -92,7 +92,6 @@ impl SplitEnumerator for PulsarSplitEnumerator {
         let splits = if topic_metadata.partitions > 0 {
             // partitioned topic
             (0..topic_metadata.partitions as i32)
-                .into_iter()
                 .map(|p| PulsarSplit {
                     topic: self.topic.sub_topic(p).unwrap(),
                     start_offset: offset.clone(),

src/connector/src/source/pulsar/enumerator/client.rs

@@ -65,7 +65,7 @@ pub fn compile_pattern_to_chrono(tmpl: &str) -> ChronoPattern {
     });
     ChronoPatternBuilder {
         tmpl: chrono_tmpl,
-        items_builder: |tmpl| StrftimeItems::new(tmpl).into_iter().collect::<Vec<_>>(),
+        items_builder: |tmpl| StrftimeItems::new(tmpl).collect::<Vec<_>>(),
     }
     .build()
 }

src/expr/src/vector_op/to_char.rs

@@ -222,11 +222,11 @@ fn infer_struct_cast_target_type(
         }
         (l, r @ NestedType::Struct(_)) | (l @ NestedType::Struct(_), r) => {
             // If only one side is nested type, these two types can never be casted.
-            return Err(ErrorCode::BindError(format!(
+            Err(ErrorCode::BindError(format!(
                 "cannot infer type because unmatched types: left={:?} right={:?}",
                 l, r
             ))
-            .into());
+            .into())
         }
         (NestedType::Type(l), NestedType::Type(r)) => {
             // If both sides are concrete types, try cast in either direction.

src/frontend/src/expr/type_inference/func.rs

@@ -295,7 +295,6 @@ fn assemble_materialize(
         )),
         Order::new(
             (0..index_columns.len())
-                .into_iter()
                 .map(FieldOrder::ascending)
                 .collect(),
         ),

src/frontend/src/handler/create_index.rs

@@ -19,7 +19,7 @@ use risingwave_pb::stream_plan::stream_fragment_graph::Parallelism;
 use risingwave_sqlparser::ast::{ColumnDef, ObjectName, Query, Statement};
 
 use super::{HandlerArgs, RwPgResponse};
-use crate::binder::{BoundSetExpr, BoundStatement};
+use crate::binder::BoundStatement;
 use crate::handler::create_table::{gen_create_table_plan_without_bind, ColumnIdGenerator};
 use crate::handler::query::handle_query;
 use crate::{build_graph, Binder, OptimizerContext};
@@ -55,16 +55,6 @@ pub async fn handle_create_as(
         let mut binder = Binder::new(&session);
         let bound = binder.bind(Statement::Query(query.clone()))?;
         if let BoundStatement::Query(query) = bound {
-            // Check if all expressions have an alias
-            if let BoundSetExpr::Select(select) = &query.body {
-                if select.aliases.iter().any(Option::is_none) {
-                    return Err(ErrorCode::BindError(
-                        "An alias must be specified for an expression".to_string(),
-                    )
-                    .into());
-                }
-            }
-
             let mut col_id_gen = ColumnIdGenerator::new_initial();
 
             // Create ColumnCatelog by Field
@@ -89,6 +79,7 @@ pub async fn handle_create_as(
         .into());
     }
 
+    // Override column name if it specified in creaet statement.
     columns.iter().enumerate().for_each(|(idx, column)| {
         column_descs[idx].name = column.name.real_value();
     });

src/frontend/src/handler/create_table_as.rs

@@ -66,7 +66,7 @@ impl<PlanRef: GenericPlanRef> GenericPlanNode for Agg<PlanRef> {
     }
 
     fn logical_pk(&self) -> Option<Vec<usize>> {
-        Some((0..self.group_key.len()).into_iter().collect_vec())
+        Some((0..self.group_key.len()).collect_vec())
     }
 
     fn ctx(&self) -> OptimizerContextRef {

src/frontend/src/optimizer/plan_node/generic/agg.rs

@@ -948,7 +948,7 @@ impl ToStream for LogicalAgg {
         // LogicalAgg.
         // Please note that the index of group key need not be changed.
 
-        let mut output_indices = (0..self.schema().len()).into_iter().collect_vec();
+        let mut output_indices = (0..self.schema().len()).collect_vec();
         output_indices
             .iter_mut()
             .skip(self.group_key().len())

src/frontend/src/optimizer/plan_node/logical_agg.rs

@@ -48,8 +48,8 @@ impl LogicalHopWindow {
         output_indices: Option<Vec<usize>>,
     ) -> Self {
         // if output_indices is not specified, use default output_indices
-        let output_indices = output_indices
-            .unwrap_or_else(|| (0..input.schema().len() + 2).into_iter().collect_vec());
+        let output_indices =
+            output_indices.unwrap_or_else(|| (0..input.schema().len() + 2).collect_vec());
         let output_type = DataType::window_of(&time_col.data_type).unwrap();
         let original_schema: Schema = input
             .schema()

src/frontend/src/optimizer/plan_node/logical_hop_window.rs

@@ -536,9 +536,7 @@ impl LogicalJoin {
                 .map(|x| x.as_input_ref().unwrap().index)
                 .collect_vec()
         } else {
-            (0..logical_scan.output_col_idx().len())
-                .into_iter()
-                .collect_vec()
+            (0..logical_scan.output_col_idx().len()).collect_vec()
         };
         let left_schema_len = logical_join.left().schema().len();
 

src/frontend/src/optimizer/plan_node/logical_join.rs

@@ -360,7 +360,6 @@ impl LogicalMultiJoin {
                 .cloned()
                 .flat_map(|input_idx| {
                     (0..self.inputs[input_idx].schema().len())
-                        .into_iter()
                         .map(move |col_idx| self.inner_i2o_mappings[input_idx].map(col_idx))
                 })
                 .enumerate()

src/frontend/src/optimizer/plan_node/logical_multi_join.rs

@@ -116,7 +116,7 @@ impl LogicalScan {
         Self::new(
             table_name,
             is_sys_table,
-            (0..table_desc.columns.len()).into_iter().collect(),
+            (0..table_desc.columns.len()).collect(),
             table_desc,
             indexes,
             ctx,

src/frontend/src/optimizer/plan_node/logical_scan.rs

@@ -297,7 +297,7 @@ impl FunctionalDependencySet {
         assert!(
             self.is_key_inner(&key),
             "{:?} is not a key!",
-            key.ones().into_iter().collect_vec()
+            key.ones().collect_vec()
         );
         let mut new_key = key.clone();
         for i in key.ones() {