Merge branch 'main' into sts/optimizer_outer_join_simplify

e2e_test/v2/batch/order_by.slt

@@ -23,6 +23,30 @@ select * from t order by v1 desc
 2 3 3
 1 4 2
 
+query III
+select * from t order by 1;
+----
+1 4 2
+2 3 3
+3 4 4
+4 3 5
+
+query III
+select v2 from t order by v1;
+----
+4
+3
+4
+3
+
+query III
+select * from t order by v1 + v2;
+----
+1 4 2
+2 3 3
+3 4 4
+4 3 5
+
 query III
 select * from t order by v1 desc limit 1
 ----

grafana/risingwave-dashboard.json

@@ -59,7 +59,7 @@
 	"fiscalYearStartMonth": 0,
 	"graphTooltip": 0,
 	"id": 2,
-	"iteration": 1650019185521,
+	"iteration": 1651822117209,
 	"links": [],
 	"liveNow": true,
 	"panels": [
@@ -518,8 +518,8 @@
 						"mode": "palette-classic"
 					},
 					"custom": {
-						"axisLabel": "",
-						"axisPlacement": "auto",
+						"axisLabel": "KB",
+						"axisPlacement": "left",
 						"barAlignment": 0,
 						"drawStyle": "line",
 						"fillOpacity": 0,
@@ -568,7 +568,7 @@
 				"x": 12,
 				"y": 18
 			},
-			"id": 168,
+			"id": 180,
 			"options": {
 				"legend": {
 					"calcs": [],
@@ -586,13 +586,22 @@
 						"uid": "PEDE6B306CC9C0CD0"
 					},
 					"exemplar": true,
-					"expr": "storage_level_compact_read_curr",
+					"expr": "version_size / 1024.0",
+					"instant": false,
 					"interval": "",
-					"legendFormat": "",
+					"legendFormat": "version size",
 					"refId": "A"
+				},
+				{
+					"datasource": {
+						"type": "prometheus",
+						"uid": "PEDE6B306CC9C0CD0"
+					},
+					"hide": false,
+					"refId": "B"
 				}
 			],
-			"title": "GBs read from current level",
+			"title": "Hummock Version Size",
 			"type": "timeseries"
 		},
 		{
@@ -738,7 +747,7 @@
 				"x": 12,
 				"y": 26
 			},
-			"id": 172,
+			"id": 168,
 			"options": {
 				"legend": {
 					"calcs": [],
@@ -756,13 +765,13 @@
 						"uid": "PEDE6B306CC9C0CD0"
 					},
 					"exemplar": true,
-					"expr": "storage_level_compact_write",
+					"expr": "storage_level_compact_read_curr",
 					"interval": "",
 					"legendFormat": "",
 					"refId": "A"
 				}
 			],
-			"title": "GBs written to next level",
+			"title": "GBs read from current level",
 			"type": "timeseries"
 		},
 		{
@@ -908,7 +917,7 @@
 				"x": 12,
 				"y": 34
 			},
-			"id": 176,
+			"id": 172,
 			"options": {
 				"legend": {
 					"calcs": [],
@@ -926,13 +935,13 @@
 						"uid": "PEDE6B306CC9C0CD0"
 					},
 					"exemplar": true,
-					"expr": "storage_level_compact_read_sstn_next",
+					"expr": "storage_level_compact_write",
 					"interval": "",
 					"legendFormat": "",
 					"refId": "A"
 				}
 			],
-			"title": "count of sst read from next level",
+			"title": "GBs written to next level",
 			"type": "timeseries"
 		},
 		{
@@ -1020,6 +1029,91 @@
 			"title": "count of sst written to next level",
 			"type": "timeseries"
 		},
+		{
+			"description": "",
+			"fieldConfig": {
+				"defaults": {
+					"color": {
+						"mode": "palette-classic"
+					},
+					"custom": {
+						"axisLabel": "",
+						"axisPlacement": "auto",
+						"barAlignment": 0,
+						"drawStyle": "line",
+						"fillOpacity": 0,
+						"gradientMode": "none",
+						"hideFrom": {
+							"legend": false,
+							"tooltip": false,
+							"viz": false
+						},
+						"lineInterpolation": "linear",
+						"lineWidth": 1,
+						"pointSize": 5,
+						"scaleDistribution": {
+							"type": "linear"
+						},
+						"showPoints": "auto",
+						"spanNulls": false,
+						"stacking": {
+							"group": "A",
+							"mode": "none"
+						},
+						"thresholdsStyle": {
+							"mode": "off"
+						}
+					},
+					"mappings": [],
+					"thresholds": {
+						"mode": "absolute",
+						"steps": [
+							{
+								"color": "green",
+								"value": null
+							},
+							{
+								"color": "red",
+								"value": 80
+							}
+						]
+					}
+				},
+				"overrides": []
+			},
+			"gridPos": {
+				"h": 8,
+				"w": 12,
+				"x": 12,
+				"y": 42
+			},
+			"id": 176,
+			"options": {
+				"legend": {
+					"calcs": [],
+					"displayMode": "list",
+					"placement": "bottom"
+				},
+				"tooltip": {
+					"mode": "single"
+				}
+			},
+			"targets": [
+				{
+					"datasource": {
+						"type": "prometheus",
+						"uid": "PEDE6B306CC9C0CD0"
+					},
+					"exemplar": true,
+					"expr": "storage_level_compact_read_sstn_next",
+					"interval": "",
+					"legendFormat": "",
+					"refId": "A"
+				}
+			],
+			"title": "count of sst read from next level",
+			"type": "timeseries"
+		},
 		{
 			"collapsed": false,
 			"gridPos": {
@@ -7647,6 +7741,6 @@
 	"timezone": "",
 	"title": "risingwave_dashboard",
 	"uid": "Ecy3uV1nz",
-	"version": 32,
+	"version": 33,
 	"weekStart": ""
 }

src/frontend/src/binder/insert.rs

@@ -87,6 +87,7 @@ impl Binder {
                         order: vec![],
                         limit: None,
                         offset: None,
+                        extra_order_exprs: vec![],
                     },
                     vec![],
                 )

src/frontend/src/binder/query.rs

@@ -17,9 +17,10 @@ use std::collections::HashMap;
 use risingwave_common::catalog::Schema;
 use risingwave_common::error::{ErrorCode, Result};
 use risingwave_common::types::DataType;
-use risingwave_sqlparser::ast::{Expr, OrderByExpr, Query};
+use risingwave_sqlparser::ast::{Expr, OrderByExpr, Query, Value};
 
 use crate::binder::{Binder, BoundSetExpr};
+use crate::expr::ExprImpl;
 use crate::optimizer::property::{Direction, FieldOrder};
 
 /// A validated sql query, including order and union.
@@ -30,6 +31,7 @@ pub struct BoundQuery {
     pub order: Vec<FieldOrder>,
     pub limit: Option<usize>,
     pub offset: Option<usize>,
+    pub extra_order_exprs: Vec<ExprImpl>,
 }
 
 impl BoundQuery {
@@ -76,39 +78,57 @@ impl Binder {
             }),
             BoundSetExpr::Values(_) => {}
         };
+        let mut extra_order_exprs = vec![];
+        let visible_output_num = body.schema().len();
         let order = query
             .order_by
             .into_iter()
-            .map(|order_by_expr| self.bind_order_by_expr(order_by_expr, &name_to_index))
+            .map(|order_by_expr| {
+                self.bind_order_by_expr(
+                    order_by_expr,
+                    &name_to_index,
+                    &mut extra_order_exprs,
+                    visible_output_num,
+                )
+            })
             .collect::<Result<_>>()?;
         Ok(BoundQuery {
             body,
             order,
             limit,
             offset,
+            extra_order_exprs,
         })
     }
 
     fn bind_order_by_expr(
         &mut self,
         order_by_expr: OrderByExpr,
         name_to_index: &HashMap<String, usize>,
+        extra_order_exprs: &mut Vec<ExprImpl>,
+        visible_output_num: usize,
     ) -> Result<FieldOrder> {
         let direct = match order_by_expr.asc {
             None | Some(true) => Direction::Asc,
             Some(false) => Direction::Desc,
         };
-        let name = match order_by_expr.expr {
-            Expr::Identifier(name) => name.value,
+        let index = match order_by_expr.expr {
+            Expr::Identifier(name) if let Some(index) = name_to_index.get(&name.value) => *index,
+            Expr::Value(Value::Number(number, _)) => match number.parse::<usize>() {
+                Ok(index) if 1 <= index && index <= visible_output_num => index - 1,
+                _ => {
+                    return Err(ErrorCode::InvalidInputSyntax(format!(
+                        "Invalid value in ORDER BY: {}",
+                        number
+                    ))
+                    .into())
+                }
+            },
             expr => {
-                return Err(
-                    ErrorCode::NotImplemented(format!("ORDER BY {:?}", expr), 1635.into()).into(),
-                )
+                extra_order_exprs.push(self.bind_expr(expr)?);
+                visible_output_num + extra_order_exprs.len() - 1
             }
         };
-        let index = *name_to_index
-            .get(&name)
-            .ok_or_else(|| ErrorCode::ItemNotFound(format!("output column \"{}\"", name)))?;
         Ok(FieldOrder { index, direct })
     }
 }

src/frontend/src/lib.rs

@@ -31,6 +31,7 @@
 #![feature(let_else)]
 #![feature(trait_alias)]
 #![feature(drain_filter)]
+#![feature(if_let_guard)]
 #[macro_use]
 pub mod catalog;
 pub mod binder;

src/frontend/src/optimizer/mod.rs

@@ -29,7 +29,7 @@ use risingwave_common::catalog::Schema;
 use risingwave_common::error::Result;
 
 use self::heuristic::{ApplyOrder, HeuristicOptimizer};
-use self::plan_node::{Convention, LogicalProject, StreamMaterialize};
+use self::plan_node::{BatchProject, Convention, LogicalProject, StreamMaterialize};
 use self::rule::*;
 use crate::catalog::TableId;
 use crate::expr::InputRef;
@@ -141,7 +141,14 @@ impl PlanRoot {
         };
 
         // Prune Columns
-        plan = plan.prune_col(&self.out_fields);
+        //
+        // Currently, the expressions in ORDER BY will be merged into the expressions in SELECT and
+        // they shouldn't be a part of output columns, so we use `out_fields` to control the
+        // visibility of these expressions. To avoid these expressions being pruned, we can't
+        // use `self.out_fields` as `required_cols` here.
+        let mut required_cols = FixedBitSet::with_capacity(self.plan.schema().len());
+        required_cols.insert_range(..);
+        plan = plan.prune_col(&required_cols);
 
         plan = {
             let rules = vec![
@@ -167,7 +174,25 @@ impl PlanRoot {
         plan = plan.to_batch_with_order_required(&self.required_order);
 
         // Convert to distributed plan
-        plan.to_distributed_with_required(&self.required_order, &self.required_dist)
+        plan = plan.to_distributed_with_required(&self.required_order, &self.required_dist);
+
+        // Add Project if the any position of `self.out_fields` is set to zero.
+        if self.out_fields.count_ones(..) != self.out_fields.len() {
+            let (exprs, expr_aliases) = self
+                .out_fields
+                .ones()
+                .zip_eq(self.schema.fields.clone())
+                .map(|(index, field)| {
+                    (
+                        InputRef::new(index, field.data_type).into(),
+                        Some(field.name),
+                    )
+                })
+                .unzip();
+            plan = BatchProject::new(LogicalProject::new(plan, exprs, expr_aliases)).into();
+        }
+
+        plan
     }
 
     /// Generate create index or create materialize view plan.