feat(query): support indexes on materialized views

src/query/service/src/interpreters/access/privilege_access.rs

@@ -590,6 +590,14 @@ impl PrivilegeAccess {
             UserPrivilegeType::Alter,
         )?;
 
+        let table = self
+            .ctx
+            .get_table(catalog_name, db_name, table_name)
+            .await?;
+        if is_materialized_view_engine(table.engine()) {
+            return self.validate_mv_source_access(table.as_ref()).await;
+        }
+
         self.validate_table_index_alter_or_super_access(catalog_name, db_name, table_name)
             .await
     }
@@ -607,6 +615,14 @@ impl PrivilegeAccess {
             UserPrivilegeType::Drop,
         )?;
 
+        let table = self
+            .ctx
+            .get_table(catalog_name, db_name, table_name)
+            .await?;
+        if is_materialized_view_engine(table.engine()) {
+            return self.validate_mv_source_access(table.as_ref()).await;
+        }
+
         match self
             .validate_table_index_alter_or_super_access(catalog_name, db_name, table_name)
             .await

src/query/service/src/interpreters/common/materialized_view.rs

@@ -16,9 +16,22 @@ use databend_common_catalog::table::Table;
 use databend_common_catalog::table::TableExt;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
+use databend_common_license::license::Feature;
+use databend_common_license::license_manager::LicenseManagerSwitch;
 use databend_common_meta_app::schema::is_materialized_view_engine;
 use databend_common_sql::plans::MaintenanceTarget;
 
+use crate::sessions::QueryContext;
+use crate::sessions::TableContextLicense;
+
+pub fn check_materialized_view_license(ctx: &QueryContext, engine: &str) -> Result<()> {
+    if is_materialized_view_engine(engine) {
+        LicenseManagerSwitch::instance()
+            .check_enterprise_enabled(ctx.get_license_key(), Feature::MaterializedView)?;
+    }
+    Ok(())
+}
+
 pub fn check_maintenance_target(table: &dyn Table, target: &MaintenanceTarget) -> Result<()> {
     match target {
         MaintenanceTarget::Table => table.check_mutable(),

src/query/service/src/interpreters/common/mod.rs

@@ -42,6 +42,7 @@ pub(crate) use lineage_log::serialize_delete_edge;
 pub(crate) use lineage_log::serialize_upsert_edge;
 pub use log::*;
 pub use materialized_view::check_maintenance_target;
+pub use materialized_view::check_materialized_view_license;
 pub use notification::get_notification_client_config;
 pub use query_log::InterpreterQueryLog;
 pub use stream::dml_build_update_stream_req;

src/query/service/src/interpreters/interpreter_table_index_create.rs

@@ -22,6 +22,7 @@ use databend_common_meta_app::schema::TableIndexType;
 use databend_common_sql::plans::CreateTableIndexPlan;
 
 use crate::interpreters::Interpreter;
+use crate::interpreters::common::check_materialized_view_license;
 use crate::pipelines::PipelineBuildResult;
 use crate::sessions::QueryContext;
 use crate::sessions::TableContextTableAccess;
@@ -55,6 +56,9 @@ impl Interpreter for CreateTableIndexInterpreter {
         let table_id = self.plan.table_id;
         let catalog = self.ctx.get_catalog(&self.plan.catalog).await?;
         let tenant = self.ctx.get_tenant();
+        if let Some(table_meta) = catalog.get_table_meta_by_id(table_id).await? {
+            check_materialized_view_license(&self.ctx, &table_meta.data.engine)?;
+        }
         let index_type = match self.plan.index_type {
             ast::TableIndexType::Aggregating => {
                 return Err(ErrorCode::InvalidArgument(

src/query/service/src/interpreters/interpreter_table_index_drop.rs

@@ -23,6 +23,7 @@ use databend_common_meta_app::schema::TableMeta;
 use databend_common_sql::plans::DropTableIndexPlan;
 
 use crate::interpreters::Interpreter;
+use crate::interpreters::common::check_materialized_view_license;
 use crate::interpreters::common::cluster_key_referenced_columns;
 use crate::pipelines::PipelineBuildResult;
 use crate::sessions::QueryContext;
@@ -66,10 +67,11 @@ impl Interpreter for DropTableIndexInterpreter {
             ast::TableIndexType::Spatial => TableIndexType::Spatial,
         };
 
-        if matches!(index_type, TableIndexType::Vector)
-            && let Some(table_meta) = catalog.get_table_meta_by_id(table_id).await?
-        {
-            validate_drop_vector_index(&table_meta.data, &index_name)?;
+        if let Some(table_meta) = catalog.get_table_meta_by_id(table_id).await? {
+            check_materialized_view_license(&self.ctx, &table_meta.data.engine)?;
+            if matches!(index_type, TableIndexType::Vector) {
+                validate_drop_vector_index(&table_meta.data, &index_name)?;
+            }
         }
 
         let drop_index_req = DropTableIndexReq {

src/query/sql/src/planner/binder/ddl/index.rs

@@ -47,8 +47,11 @@ use databend_common_meta_app::schema::GetIndexReq;
 use databend_common_meta_app::schema::IndexMeta;
 use databend_common_meta_app::schema::IndexNameIdent;
 use databend_common_meta_app::schema::ListIndexesByIdReq;
+use databend_common_meta_app::schema::MATERIALIZED_VIEW_SOURCE_ROW_ID_COLUMN;
 use databend_common_meta_app::schema::TableIndexType;
+use databend_common_meta_app::schema::is_materialized_view_engine;
 use databend_common_meta_app::tenant::Tenant;
+use databend_enterprise_materialized_view::get_materialized_view_handler;
 use databend_storages_common_table_meta::meta::Location;
 use derive_visitor::Drive;
 use derive_visitor::DriveMut;
@@ -66,6 +69,8 @@ use crate::binder::Binder;
 use crate::optimizer::OptimizerContext;
 use crate::optimizer::ir::SExpr;
 use crate::optimizer::optimize;
+use crate::parse_materialized_view_query;
+use crate::planner::semantic::MaterializedViewChecker;
 use crate::plans::CreateIndexPlan;
 use crate::plans::CreateTableIndexPlan;
 use crate::plans::DropIndexPlan;
@@ -451,8 +456,17 @@ impl Binder {
             self.normalize_object_identifier_triple(catalog, database, table);
 
         let table = self.ctx.get_table(&catalog, &database, &table).await?;
+        let is_materialized_view = is_materialized_view_engine(table.engine());
+        if is_materialized_view {
+            self.check_materialized_view_license()?;
+            if !*sync_creation {
+                return Err(ErrorCode::UnsupportedIndex(
+                    "ASYNC indexes on materialized views are not supported".to_string(),
+                ));
+            }
+        }
 
-        if table.is_read_only() {
+        if table.is_read_only() && !is_materialized_view {
             return Err(ErrorCode::UnsupportedIndex(format!(
                 "Table {} is read-only, creating index not allowed",
                 table.name()
@@ -471,6 +485,43 @@ impl Binder {
                 table.name()
             )));
         }
+        if is_materialized_view {
+            let catalog_impl = self.ctx.get_catalog(&catalog).await?;
+            let definition = get_materialized_view_handler()
+                .get_mv_definition(
+                    catalog_impl.as_ref(),
+                    &self.ctx.get_tenant(),
+                    table.get_id(),
+                )
+                .await?
+                .ok_or_else(|| {
+                    ErrorCode::InvalidMaterializedView(format!(
+                        "materialized view {} has no definition",
+                        table.name()
+                    ))
+                })?;
+            let query = parse_materialized_view_query(
+                &definition.data.original_query,
+                "invalid materialized view logical query",
+            )?;
+            if MaterializedViewChecker::check_query(&query).is_aggregating() {
+                return Err(ErrorCode::UnsupportedIndex(
+                    "Indexes on aggregating materialized views are not supported".to_string(),
+                ));
+            }
+        }
+        if is_materialized_view
+            && columns.iter().any(|column| {
+                column
+                    .name
+                    .eq_ignore_ascii_case(MATERIALIZED_VIEW_SOURCE_ROW_ID_COLUMN)
+            })
+        {
+            return Err(ErrorCode::UnsupportedIndex(format!(
+                "Materialized view internal column {} cannot be indexed",
+                MATERIALIZED_VIEW_SOURCE_ROW_ID_COLUMN
+            )));
+        }
         let table_schema = table.schema();
         let table_id = table.get_id();
         let index_name = self.normalize_object_identifier(index_name);
@@ -912,6 +963,9 @@ impl Binder {
             self.normalize_object_identifier_triple(catalog, database, table);
 
         let table = self.ctx.get_table(&catalog, &database, &table).await?;
+        if is_materialized_view_engine(table.engine()) {
+            self.check_materialized_view_license()?;
+        }
         if !table.support_index() {
             return Err(ErrorCode::UnsupportedIndex(format!(
                 "Table engine {} does not support create index",

src/query/sql/src/planner/binder/ddl/materialized_view.rs

@@ -308,7 +308,7 @@ impl Binder {
         Ok(())
     }
 
-    fn check_materialized_view_license(&self) -> Result<()> {
+    pub(in crate::planner::binder) fn check_materialized_view_license(&self) -> Result<()> {
         LicenseManagerSwitch::instance()
             .check_enterprise_enabled(self.ctx.get_license_key(), Feature::MaterializedView)
     }

src/query/storages/system/src/indexes_table.rs

@@ -115,11 +115,11 @@ impl AsyncSystemTable for IndexesTable {
             updated_on.push(index.updated_on.map(|u| u.timestamp_micros()));
         }
 
-        for table in table_index_tables {
+        for (database_name, table) in table_index_tables {
             for (name, index) in &table.meta.indexes {
                 names.push(name.clone());
                 types.push(index.index_type.to_string());
-                databases.push(table.database_name()?.to_string());
+                databases.push(database_name.clone());
                 tables.push(Some(table.name.to_string()));
                 originals.push("".to_string());
 
@@ -201,7 +201,7 @@ impl IndexesTable {
         ctx: Arc<dyn TableContext>,
         database_names: Option<&[String]>,
         table_names: Option<&[String]>,
-    ) -> Result<Vec<TableInfo>> {
+    ) -> Result<Vec<(String, TableInfo)>> {
         let tenant = ctx.get_tenant();
         let visibility_checker = ctx.get_visibility_checker(false, Object::All).await?;
         let catalog = ctx.get_catalog(CATALOG_DEFAULT).await?;
@@ -289,7 +289,7 @@ impl IndexesTable {
                     db_id,
                     table.get_id(),
                 ) {
-                    index_tables.push(table_info.clone());
+                    index_tables.push((db_name.to_string(), table_info.clone()));
                 }
             }
         }

tests/sqllogictests/suites/ee/07_ee_materialized_view/07_0003_materialized_view_index.test

@@ -0,0 +1,121 @@
+## Copyright 2023 Databend Cloud
+##
+## Licensed under the Elastic License, Version 2.0 (the "License");
+## you may not use this file except in compliance with the License.
+## You may obtain a copy of the License at
+##
+##     https://www.elastic.co/licensing/elastic-license
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+
+# Materialized views reuse Fuse table-index metadata and synchronous file construction. ASYNC
+# indexes and index read paths are intentionally out of scope.
+
+statement ok
+drop database if exists test_materialized_view_index
+
+statement ok
+create database test_materialized_view_index
+
+statement ok
+use test_materialized_view_index
+
+statement ok
+create table source (
+    id int,
+    content string,
+    tags string,
+    embedding vector(4)
+) row_per_block = 2
+
+statement ok
+insert into source values
+    (1, 'alpha beta',  'one',   [1.0, 0.0, 0.0, 0.0]),
+    (2, 'gamma',       'two',   [0.0, 1.0, 0.0, 0.0]),
+    (3, 'alpha delta', 'three', [0.9, 0.1, 0.0, 0.0]),
+    (4, 'epsilon',     'four',  [0.0, 0.0, 1.0, 0.0])
+
+statement ok
+create materialized view mv_idx row_per_block = 2 as
+select id, content, tags, embedding from source where 1 = 1
+
+# CREATE reuses the table-index metadata API. The first MV refresh builds index files while it
+# writes the initial MV blocks.
+statement ok
+create inverted index mv_inv on mv_idx(content)
+
+# ASYNC indexes require an explicit MV refresh-index lifecycle for backfill and repair.
+statement error 1601
+create async inverted index mv_tags on mv_idx(tags)
+
+statement ok
+create vector index mv_vec on mv_idx(embedding)
+    m = 10 ef_construct = 40 distance = 'cosine'
+
+query TT rowsort
+select name, type
+from system.indexes
+where database = 'test_materialized_view_index'
+  and name in ('mv_inv', 'mv_tags', 'mv_vec')
+----
+mv_inv INVERTED
+mv_vec VECTOR
+
+# Physical refresh columns are internal implementation details, not MV index targets.
+statement error 1601
+create inverted index hidden_idx on mv_idx(_mv_source_row_id)
+
+# MV refresh writes the initial blocks and builds synchronous index files in the write pipeline.
+statement ok
+refresh materialized view mv_idx
+
+query BB
+select count(*) > 0,
+       count_if(inverted_index_size > 0 and vector_index_size > 0) = count(*)
+from fuse_block('test_materialized_view_index', 'mv_idx')
+----
+1 1
+
+statement ok
+insert into source values (5, 'alpha zeta', 'fresh_tag', [0.8, 0.2, 0.0, 0.0])
+
+# Incremental MV refresh also builds both synchronous indexes for newly written blocks.
+statement ok
+refresh materialized view mv_idx
+
+query BB
+select count(*) > 0,
+       count_if(inverted_index_size > 0 and vector_index_size > 0) = count(*)
+from fuse_block('test_materialized_view_index', 'mv_idx')
+----
+1 1
+
+statement ok
+drop inverted index mv_inv on mv_idx
+
+statement ok
+drop vector index mv_vec on mv_idx
+
+query I
+select count(*)
+from system.indexes
+where database = 'test_materialized_view_index'
+  and name in ('mv_inv', 'mv_tags', 'mv_vec')
+----
+0
+
+# Aggregating MVs persist aggregate states instead of finalized values, so their table-index
+# lifecycle remains unsupported until those index semantics are defined.
+statement ok
+create materialized view mv_agg (content, row_count) as
+select content, count(*) from source where 1 = 1 group by content
+
+statement error 1601
+create inverted index aggregate_idx on mv_agg(content)
+
+statement ok
+drop database test_materialized_view_index

tests/suites/5_ee/10_rbac/10_0003_materialized_view_privilege.result

@@ -1,7 +1,9 @@
 === CREATE MV requires CREATE on the database and SELECT on the source ===
 Error: APIError: QueryFailed: [1063]Permission denied: privilege [Create] is required on 'default'.'default'.* for user 'mv-priv-user'@'%' with roles [mv_priv_role]. Note: Please ensure that your current role have the appropriate permissions to create a new Object
 Error: APIError: QueryFailed: [1063]Permission denied: privilege [Select] is required on 'default'.'default'.'mv_rbac_source_0020' for user 'mv-priv-user'@'%' with roles [mv_priv_role,public]
-=== SELECT, REFRESH, and SHOW CREATE MATERIALIZED VIEW require only SELECT on the source ===
+=== SELECT, REFRESH, SHOW CREATE, and INDEX DDL require only SELECT on the source ===
+Error: APIError: QueryFailed: [1063]Permission denied: privilege [Select] is required on 'default'.'default'.'mv_rbac_source_0020' for user 'mv-priv-user'@'%' with roles [mv_priv_role,public]
+Error: APIError: QueryFailed: [1063]Permission denied: privilege [Select] is required on 'default'.'default'.'mv_rbac_source_0020' for user 'mv-priv-user'@'%' with roles [mv_priv_role,public]
 Error: APIError: QueryFailed: [1063]Permission denied: privilege [Select] is required on 'default'.'default'.'mv_rbac_source_0020' for user 'mv-priv-user'@'%' with roles [mv_priv_role,public]
 Error: APIError: QueryFailed: [1063]Permission denied: privilege [Select] is required on 'default'.'default'.'mv_rbac_source_0020' for user 'mv-priv-user'@'%' with roles [mv_priv_role,public]
 Error: APIError: QueryFailed: [1063]Permission denied: privilege [Select] is required on 'default'.'default'.'mv_rbac_source_0020' for user 'mv-priv-user'@'%' with roles [mv_priv_role,public]