Merge branch 'main' of https://github.com/risingwavelabs/risingwave i…

…nto li0k/object_store_fix_retry

.github/workflows/auto-update-helm-and-operator-version-by-release.yml

@@ -0,0 +1,71 @@
+name: Update Helm Charts and Risingwave Operator on New Release
+
+on:
+  release:
+    types: [published]
+  workflow_dispatch:
+    inputs:
+      version:
+        description: 'release version'
+        required: true
+
+env:
+  NEW_APP_VERSION: ${{ github.event.inputs.version || github.event.release.tag_name }}
+
+jobs:
+  update-helm-charts:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout Helm Charts Repository
+        uses: actions/checkout@v3
+        with:
+          repository: 'risingwavelabs/helm-charts'
+          token: ${{ secrets.GITHUB_TOKEN }}
+          path: 'helm-charts'
+
+      - name: Update values.yaml
+        run: |
+          sed -i "s/^  tag:.*/  tag: \"${{ env.NEW_APP_VERSION }}\"/" helm-charts/charts/risingwave/values.yaml
+
+      - name: Update Chart.yaml
+        run: |
+          cd helm-charts/charts/risingwave
+          CURRENT_VERSION=$(grep 'version:' Chart.yaml | awk '{print $2}' | head -n 1)
+          NEW_VERSION=$(echo $CURRENT_VERSION | awk -F. -v OFS='.' '{$NF++; print}')
+          sed -i "/type: application/,/version:/!b; /version:/s/version: .*/version: $NEW_VERSION/" Chart.yaml
+          sed -i "s/^appVersion: .*/appVersion: \"${{ env.NEW_APP_VERSION }}\"/" Chart.yaml
+          echo "NEW_CHART_VERSION=$NEW_VERSION" >> $GITHUB_ENV
+
+      - name: Create Pull Request
+        uses: peter-evans/create-pull-request@v4
+        with:
+          token: ${{ secrets.GITHUB_TOKEN }}
+          commit-message: 'chore: bump risingwave to ${{ env.NEW_APP_VERSION }}, release chart ${{ env.NEW_CHART_VERSION }}'
+          title: 'chore: bump risingwave to ${{ env.NEW_APP_VERSION }}, release chart ${{ env.NEW_CHART_VERSION }}'
+          body: 'This is an automated pull request to update the chart versions'
+          branch: 'auto-update-${{ env.NEW_APP_VERSION }}'
+
+  update-risingwave-operator:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout Risingwave Operator Repository
+        uses: actions/checkout@v3
+        with:
+          repository: 'risingwavelabs/risingwave-operator'
+          token: ${{ secrets.GITHUB_TOKEN }}
+          path: 'risingwave-operator'
+
+      - name: Update risingwave-operator image tags
+        run: |
+          cd risingwave-operator
+          PREV_VERSION=$(grep -roh "risingwavelabs/risingwave:v[0-9\.]*" * | head -n 1 | cut -d':' -f2)
+          grep -rl "risingwavelabs/risingwave:$PREV_VERSION" . | xargs sed -i "s|risingwavelabs/risingwave:$PREV_VERSION|risingwavelabs/risingwave:${{ env.NEW_APP_VERSION }}|g"
+
+      - name: Create Pull Request for risingwave-operator
+        uses: peter-evans/create-pull-request@v4
+        with:
+          token: ${{ secrets.GITHUB_TOKEN }}
+          commit-message: 'chore: bump risingwave image tags to ${{ env.NEW_APP_VERSION }}'
+          title: 'chore: bump risingwave image tags to ${{ env.NEW_APP_VERSION }}'
+          body: 'This is an automated pull request to update the risingwave image tags'
+          branch: 'auto-update-${{ env.NEW_APP_VERSION }}'

e2e_test/sink/sink_into_table/specify_column.slt

@@ -0,0 +1,54 @@
+statement ok
+SET RW_IMPLICIT_FLUSH TO true;
+
+statement ok
+create table s (a int, b int, c int) append only;
+
+statement ok
+create table t (a int, b int default 900, c int default 9000);
+
+statement error
+create sink ss into t(aaa) as select a from s with(type = 'append-only');
+
+statement error
+create sink ss into t(a) as select a, b from s with(type = 'append-only');
+
+statement error
+create sink ss into t(a, b) as select b from s with(type = 'append-only');
+
+statement error
+create sink ss into t(a, b, c, a) as select a, b from s with(type = 'append-only');
+
+statement ok
+create sink s1 into t(a,B,c) as select c, b, a from s with(type = 'append-only');
+
+statement ok
+create sink s2 into t(a,B) as select 2*c, 2*b from s with(type = 'append-only');
+
+statement ok
+create sink s3 into t(c) as select 3*a from s with(type = 'append-only');
+
+statement ok
+insert into s values(10, 100, 1000);
+
+query III rowsort
+select * from t order by a;
+----
+1000	100	10
+2000	200	9000
+NULL	900	30
+
+statement ok
+drop sink s1;
+
+statement ok
+drop sink s2;
+
+statement ok
+drop sink s3;
+
+statement ok
+drop table s;
+
+statement ok
+drop table t;

src/frontend/src/catalog/table_catalog.rs

@@ -450,6 +450,20 @@ impl TableCatalog {
             .map(|(i, _)| i)
     }
 
+    pub fn default_column_expr(&self, col_idx: usize) -> ExprImpl {
+        if let Some(GeneratedOrDefaultColumn::DefaultColumn(DefaultColumnDesc { expr, .. })) = self
+            .columns[col_idx]
+            .column_desc
+            .generated_or_default_column
+            .as_ref()
+        {
+            ExprImpl::from_expr_proto(expr.as_ref().unwrap())
+                .expect("expr in default columns corrupted")
+        } else {
+            ExprImpl::literal_null(self.columns[col_idx].data_type().clone())
+        }
+    }
+
     pub fn default_columns(&self) -> impl Iterator<Item = (usize, ExprImpl)> + '_ {
         self.columns.iter().enumerate().filter_map(|(i, c)| {
             if let Some(GeneratedOrDefaultColumn::DefaultColumn(DefaultColumnDesc {

src/frontend/src/handler/create_mv.rs

@@ -37,21 +37,24 @@ use crate::scheduler::streaming_manager::CreatingStreamingJobInfo;
 use crate::session::SessionImpl;
 use crate::stream_fragmenter::build_graph;
 
+pub(super) fn parse_column_names(columns: &[Ident]) -> Option<Vec<String>> {
+    if columns.is_empty() {
+        None
+    } else {
+        Some(columns.iter().map(|v| v.real_value()).collect())
+    }
+}
+
+/// If columns is empty, it means that the user did not specify the column names.
+/// In this case, we extract the column names from the query.
+/// If columns is not empty, it means that user specify the column names and the user
+/// should guarantee that the column names number are consistent with the query.
 pub(super) fn get_column_names(
     bound: &BoundQuery,
     session: &SessionImpl,
     columns: Vec<Ident>,
 ) -> Result<Option<Vec<String>>> {
-    // If columns is empty, it means that the user did not specify the column names.
-    // In this case, we extract the column names from the query.
-    // If columns is not empty, it means that user specify the column names and the user
-    // should guarantee that the column names number are consistent with the query.
-    let col_names: Option<Vec<String>> = if columns.is_empty() {
-        None
-    } else {
-        Some(columns.iter().map(|v| v.real_value()).collect())
-    };
-
+    let col_names = parse_column_names(&columns);
     if let BoundSetExpr::Select(select) = &bound.body {
         // `InputRef`'s alias will be implicitly assigned in `bind_project`.
         // If user provide columns name (col_names.is_some()), we don't need alias.