fix: fix index create privilege check and ensure consistency between …

…the owner and its associated table (#19252)

src/frontend/src/handler/create_index.rs

@@ -21,11 +21,9 @@ use either::Either;
 use fixedbitset::FixedBitSet;
 use itertools::Itertools;
 use pgwire::pg_response::{PgResponse, StatementType};
-use risingwave_common::acl::AclMode;
 use risingwave_common::catalog::{IndexId, TableDesc, TableId};
 use risingwave_common::util::sort_util::{ColumnOrder, OrderType};
 use risingwave_pb::catalog::{PbIndex, PbIndexColumnProperties, PbStreamJobStatus, PbTable};
-use risingwave_pb::user::grant_privilege::Object;
 use risingwave_sqlparser::ast;
 use risingwave_sqlparser::ast::{Ident, ObjectName, OrderByExpr};
 
@@ -34,7 +32,6 @@ use crate::binder::Binder;
 use crate::catalog::root_catalog::SchemaPath;
 use crate::error::{ErrorCode, Result};
 use crate::expr::{Expr, ExprImpl, ExprRewriter, InputRef};
-use crate::handler::privilege::ObjectCheckItem;
 use crate::handler::HandlerArgs;
 use crate::optimizer::plan_expr_rewriter::ConstEvalRewriter;
 use crate::optimizer::plan_node::{Explain, LogicalProject, LogicalScan, StreamMaterialize};
@@ -83,11 +80,11 @@ pub(crate) fn gen_create_index_plan(
         );
     }
 
-    session.check_privileges(&[ObjectCheckItem::new(
-        table.owner,
-        AclMode::Select,
-        Object::TableId(table.id.table_id),
-    )])?;
+    if !session.is_super_user() && session.user_id() != table.owner {
+        return Err(
+            ErrorCode::PermissionDenied(format!("must be owner of table {}", table.name)).into(),
+        );
+    }
 
     let mut binder = Binder::new_for_stream(session);
     binder.bind_table(Some(&schema_name), &table_name, None)?;
@@ -202,7 +199,7 @@ pub(crate) fn gen_create_index_plan(
         &index_columns_ordered_expr,
         &include_columns_expr,
         // We use the first index column as distributed key by default if users
-        // haven't specify the distributed by columns.
+        // haven't specified the distributed by columns.
         if distributed_columns_expr.is_empty() {
             1
         } else {
@@ -221,7 +218,7 @@ pub(crate) fn gen_create_index_plan(
         index_table_prost.retention_seconds = table.retention_seconds;
     }
 
-    index_table_prost.owner = session.user_id();
+    index_table_prost.owner = table.owner;
     index_table_prost.dependent_relations = vec![table.id.table_id];
 
     let index_columns_len = index_columns_ordered_expr.len() as u32;