feat(optimizer): recommend index for backfill to users

src/frontend/src/optimizer/optimizer_context.rs

@@ -206,6 +206,10 @@ impl OptimizerContext {
         self.explain_options.backfill
     }
 
+    pub fn is_explain_advisor(&self) -> bool {
+        self.explain_options.advisor
+    }
+
     pub fn explain_type(&self) -> ExplainType {
         self.explain_options.explain_type.clone()
     }

src/frontend/src/optimizer/plan_node/convert.rs

@@ -56,6 +56,9 @@ pub trait ToStream {
         required_dist.streaming_enforce_if_not_satisfies(ret)
     }
 
+    /// Try to generate a plan with better locality for the given columns.
+    /// This is used to improve the performance of aggregation, over window, join and so on
+    /// by colocating the (group by, partition or join) keys.
     fn try_better_locality(&self, _columns: &[usize]) -> Option<LogicalPlanRef> {
         None
     }

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

@@ -20,7 +20,9 @@ use pretty_xmlish::{Pretty, XmlNode};
 use risingwave_common::catalog::{ColumnDesc, Schema};
 use risingwave_common::util::sort_util::{ColumnOrder, OrderType};
 use risingwave_pb::stream_plan::StreamScanType;
-use risingwave_sqlparser::ast::AsOf;
+use risingwave_sqlparser::ast::{
+    AsOf, Expr, Ident, ObjectName, OrderByExpr, Statement, WithProperties,
+};
 
 use super::generic::{GenericPlanNode, GenericPlanRef};
 use super::utils::{Distill, childless_record};
@@ -565,6 +567,63 @@ impl LogicalScan {
 
         None
     }
+
+    /// Notify user about the recommended index
+    fn notice_recommended_index(&self, columns: &[usize]) {
+        let table_column_indices = columns
+            .iter()
+            .map(|&col| self.output_col_idx()[col])
+            .collect_vec();
+
+        let primary_key_columns: Vec<usize> = self
+            .primary_key()
+            .iter()
+            .map(|col_order| col_order.column_index)
+            .collect();
+
+        if primary_key_columns == table_column_indices {
+            // Don't recommend an index that's identical to the primary key
+            return;
+        }
+
+        let column_names = table_column_indices
+            .iter()
+            .map(|&col| self.table().columns[col].name.clone())
+            .collect_vec();
+
+        // Construct CREATE INDEX statement using AST for proper identifier quoting
+        let columns = column_names
+            .iter()
+            .map(|col| OrderByExpr {
+                expr: Expr::Identifier(Ident::with_quote_unchecked('"', col)),
+                asc: None, // defaults to ASC
+                nulls_first: None,
+            })
+            .collect();
+
+        let index_name = format!(
+            "__recommended_idx_of_{}_{}",
+            self.table_name(),
+            column_names.join("_")
+        );
+
+        let create_index_stmt = Statement::CreateIndex {
+            name: ObjectName::from(vec![Ident::with_quote_unchecked('"', &index_name)]),
+            table_name: ObjectName::from(vec![Ident::with_quote_unchecked('"', self.table_name())]),
+            columns,
+            method: None,
+            include: vec![],
+            distributed_by: vec![],
+            unique: false,
+            if_not_exists: false,
+            with_properties: WithProperties(vec![]),
+        };
+
+        self.core.ctx().warn_to_user(format!(
+            "To speed up the backfilling, consider creating an index: {}",
+            create_index_stmt
+        ));
+    }
 }
 
 impl ToBatch for LogicalScan {
@@ -692,9 +751,14 @@ impl ToStream for LogicalScan {
         if columns.is_empty() {
             return None;
         }
+
         if self.table_indexes().is_empty() {
+            if self.ctx().is_explain_advisor() {
+                self.notice_recommended_index(columns);
+            }
             return None;
         }
+
         let orders = if columns.len() <= 3 {
             OrderType::all()
         } else {
@@ -722,6 +786,10 @@ impl ToStream for LogicalScan {
                 }
             }
         }
+
+        if self.ctx().is_explain_advisor() {
+            self.notice_recommended_index(columns);
+        }
         None
     }
 }

src/sqlparser/src/ast/mod.rs

@@ -1204,6 +1204,8 @@ pub struct ExplainOptions {
     pub trace: bool,
     // Display backfill order
     pub backfill: bool,
+    // Show advisor recommendations
+    pub advisor: bool,
     // explain's plan type
     pub explain_type: ExplainType,
     // explain's plan format
@@ -1216,6 +1218,7 @@ impl Default for ExplainOptions {
             verbose: false,
             trace: false,
             backfill: false,
+            advisor: false,
             explain_type: ExplainType::Physical,
             explain_format: ExplainFormat::Text,
         }

src/sqlparser/src/keywords.rs

@@ -73,6 +73,7 @@ define_keywords!(
     ACTION,
     ADAPTIVE,
     ADD,
+    ADVISOR,
     AGGREGATE,
     ALL,
     ALLOCATE,

src/sqlparser/src/parser.rs

@@ -4364,6 +4364,7 @@ impl Parser<'_> {
         let mut analyze_duration = None;
 
         let explain_key_words = [
+            Keyword::ADVISOR,
             Keyword::BACKFILL,
             Keyword::VERBOSE,
             Keyword::TRACE,
@@ -4378,6 +4379,7 @@ impl Parser<'_> {
         let parse_explain_option = |parser: &mut Parser<'_>| -> ModalResult<()> {
             let keyword = parser.expect_one_of_keywords(&explain_key_words)?;
             match keyword {
+                Keyword::ADVISOR => options.advisor = parser.parse_optional_boolean(true),
                 Keyword::VERBOSE => options.verbose = parser.parse_optional_boolean(true),
                 Keyword::TRACE => options.trace = parser.parse_optional_boolean(true),
                 Keyword::BACKFILL => options.backfill = parser.parse_optional_boolean(true),