fix: changelog does not support fully qualified path (#18785)

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

e2e_test/streaming/changelog.slt

@@ -10,8 +10,9 @@ create table t2 (v1 int, v2 int);
 statement ok
 create table t3 (v1 int primary key, v2 int);
 
+# test public.t1 due to https://github.com/risingwavelabs/risingwave/issues/18747
 statement ok
-create materialized view mv1 as with sub as changelog from t1 select * from sub;
+create materialized view mv1 as with sub as changelog from public.t1 select * from sub;
 
 statement ok
 create materialized view mv2 as with sub as changelog from t2 select * from sub;

src/frontend/src/binder/query.rs

@@ -20,7 +20,7 @@ use risingwave_common::catalog::Schema;
 use risingwave_common::types::DataType;
 use risingwave_common::util::sort_util::{ColumnOrder, OrderType};
 use risingwave_sqlparser::ast::{
-    Cte, CteInner, Expr, Fetch, ObjectName, OrderByExpr, Query, SetExpr, SetOperator, Value, With,
+    Cte, CteInner, Expr, Fetch, OrderByExpr, Query, SetExpr, SetOperator, Value, With,
 };
 use thiserror_ext::AsReport;
 
@@ -350,11 +350,8 @@ impl Binder {
                     }
                     CteInner::ChangeLog(from_table_name) => {
                         self.push_context();
-                        let from_table_relation = self.bind_relation_by_name(
-                            ObjectName::from(vec![from_table_name]),
-                            None,
-                            None,
-                        )?;
+                        let from_table_relation =
+                            self.bind_relation_by_name(from_table_name.clone(), None, None)?;
                         self.pop_context()?;
                         self.context.cte_to_relation.insert(
                             table_name,

src/meta/src/controller/rename.rs

@@ -156,7 +156,7 @@ impl QueryRewriter<'_> {
                 match &mut cte_table.cte_inner {
                     risingwave_sqlparser::ast::CteInner::Query(query) => self.visit_query(query),
                     risingwave_sqlparser::ast::CteInner::ChangeLog(from) => {
-                        *from = Ident::new_unchecked(self.to)
+                        *from = ObjectName(vec![Ident::new_unchecked(self.to)])
                     }
                 }
             }

src/sqlparser/src/ast/query.rs

@@ -335,9 +335,12 @@ impl fmt::Display for Cte {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match &self.cte_inner {
             CteInner::Query(query) => write!(f, "{} AS ({})", self.alias, query)?,
-            CteInner::ChangeLog(ident) => {
-                write!(f, "{} AS changelog from {}", self.alias, ident.value)?
-            }
+            CteInner::ChangeLog(obj_name) => write!(
+                f,
+                "{} AS changelog from {}",
+                self.alias,
+                obj_name.real_value()
+            )?,
         }
         Ok(())
     }
@@ -347,7 +350,7 @@ impl fmt::Display for Cte {
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 pub enum CteInner {
     Query(Query),
-    ChangeLog(Ident),
+    ChangeLog(ObjectName),
 }
 
 /// One item of the comma-separated list following `SELECT`

src/sqlparser/src/parser.rs

@@ -4141,7 +4141,7 @@ impl Parser<'_> {
                 parser_err!("Expected 'changelog' but found '{}'", changelog);
             }
             self.expect_keyword(Keyword::FROM)?;
-            Ok(CteInner::ChangeLog(self.parse_identifier()?))
+            Ok(CteInner::ChangeLog(self.parse_object_name()?))
         }
     }