[jira] [Commented] (DRILL-8543) Add Support for Materialized Views

[ASF GitHub Bot (Jira)]

    [ 
https://issues.apache.org/jira/browse/DRILL-8543?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18090647#comment-18090647
 ] 

ASF GitHub Bot commented on DRILL-8543:
---------------------------------------

rymarm commented on code in PR #3036:
URL: https://github.com/apache/drill/pull/3036#discussion_r3442457008


##########
exec/java-exec/src/main/java/org/apache/drill/exec/dotdrill/MaterializedView.java:
##########
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ */
+package org.apache.drill.exec.dotdrill;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonInclude.Include;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+
+/**
+ * Represents a materialized view definition stored as a JSON file with
+ * .materialized_view.drill extension. The actual data is stored separately
+ * in Parquet format in the workspace directory.
+ */
+@JsonTypeName("materialized_view")
+public class MaterializedView {
+
+  /**
+   * Represents the refresh status of the materialized view.
+   */
+  public enum RefreshStatus {
+    /** The materialized view data is complete and up-to-date with its 
definition */
+    COMPLETE,
+    /** The materialized view data needs to be refreshed */
+    INCOMPLETE
+  }
+
+  private final String name;
+  private String sql;
+  private List<View.Field> fields;
+
+  /** Current schema when materialized view is created (not the schema to 
which view belongs to) */
+  private List<String> workspaceSchemaPath;
+
+  /** The relative path where the materialized data is stored (typically the 
view name) */
+  @JsonInclude(Include.NON_NULL)
+  private String dataStoragePath;

Review Comment:
   In 
`exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillMaterializedViewTable.java`
 you specified, that the MV data is stored within `{name}_mv_data/`:
   ```
    * A materialized view stores:
    * <ul>
    *   <li>Definition file (.materialized_view.drill) - JSON with name, SQL, 
schema info</li>
    *   <li>Data directory ({name}_mv_data/) - Parquet files with pre-computed 
results</li>
    * </ul>
   ```
   
   In the unit tests, you used the MV name for the data storage, but at the 
same time, you used `{name}_mv_data` in the following places:
   - 
`exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillMaterializedViewTable.java`
   - 
`exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/MaterializedViewRewriter.java`
   - 
`exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/WorkspaceSchemaFactory.java`
   
   We need to agree on what name pattern to use: `{name}_mv_data` or simply 
`{name}`. 



##########
exec/java-exec/src/main/java/org/apache/drill/exec/dotdrill/MaterializedView.java:
##########
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ */
+package org.apache.drill.exec.dotdrill;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonInclude.Include;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+
+/**
+ * Represents a materialized view definition stored as a JSON file with
+ * .materialized_view.drill extension. The actual data is stored separately
+ * in Parquet format in the workspace directory.
+ */
+@JsonTypeName("materialized_view")
+public class MaterializedView {
+
+  /**
+   * Represents the refresh status of the materialized view.
+   */
+  public enum RefreshStatus {
+    /** The materialized view data is complete and up-to-date with its 
definition */
+    COMPLETE,
+    /** The materialized view data needs to be refreshed */
+    INCOMPLETE
+  }
+
+  private final String name;
+  private String sql;
+  private List<View.Field> fields;
+
+  /** Current schema when materialized view is created (not the schema to 
which view belongs to) */
+  private List<String> workspaceSchemaPath;
+
+  /** The relative path where the materialized data is stored (typically the 
view name) */
+  @JsonInclude(Include.NON_NULL)
+  private String dataStoragePath;
+
+  /** Timestamp of the last successful refresh in milliseconds since epoch */
+  @JsonInclude(Include.NON_NULL)
+  private Long lastRefreshTime;
+
+  /** Current refresh status of the materialized view */
+  @JsonInclude(Include.NON_NULL)
+  private RefreshStatus refreshStatus;
+
+  public MaterializedView(String name, String sql, RelDataType rowType, 
List<String> workspaceSchemaPath) {
+    this(name,
+        sql,
+        rowType.getFieldList().stream()
+            .map(f -> new View.Field(f.getName(), f.getType()))
+            .collect(Collectors.toList()),
+        workspaceSchemaPath,
+        name,  // data storage path defaults to view name

Review Comment:
   We need to agree on what name pattern to use by default: `{name}_mv_data` or 
simply `{name}`.



##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/MaterializedViewRewriter.java:
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ */
+package org.apache.drill.exec.planner.logical;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.calcite.plan.RelOptMaterialization;
+import org.apache.calcite.plan.RelOptMaterializations;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelRoot;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.util.Pair;
+import org.apache.drill.exec.dotdrill.MaterializedView;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.planner.sql.conversion.SqlConverter;
+import org.apache.drill.exec.store.AbstractSchema;
+import org.apache.drill.exec.store.StoragePlugin;
+import org.apache.drill.exec.store.StoragePluginRegistry;
+import org.apache.drill.exec.store.StoragePluginRegistry.PluginException;
+import org.apache.drill.exec.store.dfs.FileSystemPlugin;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Utility class for materialized view query rewriting.
+ * <p>
+ * When enabled via planner.enable_materialized_view_rewrite, this class 
attempts
+ * to rewrite queries to use materialized views when beneficial.
+ * <p>
+ * Uses Calcite's {@link RelOptMaterializations#useMaterializedViews} API which
+ * normalizes both the query and MV definitions before performing structural
+ * matching via {@link org.apache.calcite.plan.SubstitutionVisitor}.
+ * <p>
+ * Materialized views are discovered by iterating over enabled file-based
+ * storage plugins (the only plugin type that supports MVs) and force-loading
+ * their schemas to find .materialized_view.drill files.
+ */
+public class MaterializedViewRewriter {
+  private static final Logger logger = 
LoggerFactory.getLogger(MaterializedViewRewriter.class);
+
+  private final QueryContext context;
+  private final SchemaPlus rootSchema;
+  private final SqlConverter sqlConverter;
+
+  public MaterializedViewRewriter(QueryContext context, SchemaPlus rootSchema, 
SqlConverter sqlConverter) {
+    this.context = context;
+    this.rootSchema = rootSchema;
+    this.sqlConverter = sqlConverter;
+  }
+
+  /**
+   * Attempts to rewrite the given RelNode to use a materialized view.
+   *
+   * @param queryRel the query plan to potentially rewrite
+   * @return the rewritten plan using an MV, or the original plan if no 
rewrite is possible
+   */
+  public RelNode rewrite(RelNode queryRel) {
+    if (!context.getPlannerSettings().isMaterializedViewRewriteEnabled()) {
+      return queryRel;
+    }
+
+    // Find all available materialized views that have been refreshed

Review Comment:
   Actually, `findCandidateMaterializedViews()` returns both refreshed and 
unrefreshed MVs.



##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/MaterializedViewRewriter.java:
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ */
+package org.apache.drill.exec.planner.logical;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.calcite.plan.RelOptMaterialization;
+import org.apache.calcite.plan.RelOptMaterializations;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelRoot;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.util.Pair;
+import org.apache.drill.exec.dotdrill.MaterializedView;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.planner.sql.conversion.SqlConverter;
+import org.apache.drill.exec.store.AbstractSchema;
+import org.apache.drill.exec.store.StoragePlugin;
+import org.apache.drill.exec.store.StoragePluginRegistry;
+import org.apache.drill.exec.store.StoragePluginRegistry.PluginException;
+import org.apache.drill.exec.store.dfs.FileSystemPlugin;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Utility class for materialized view query rewriting.
+ * <p>
+ * When enabled via planner.enable_materialized_view_rewrite, this class 
attempts
+ * to rewrite queries to use materialized views when beneficial.
+ * <p>
+ * Uses Calcite's {@link RelOptMaterializations#useMaterializedViews} API which
+ * normalizes both the query and MV definitions before performing structural
+ * matching via {@link org.apache.calcite.plan.SubstitutionVisitor}.
+ * <p>
+ * Materialized views are discovered by iterating over enabled file-based
+ * storage plugins (the only plugin type that supports MVs) and force-loading
+ * their schemas to find .materialized_view.drill files.
+ */
+public class MaterializedViewRewriter {
+  private static final Logger logger = 
LoggerFactory.getLogger(MaterializedViewRewriter.class);
+
+  private final QueryContext context;
+  private final SchemaPlus rootSchema;
+  private final SqlConverter sqlConverter;
+
+  public MaterializedViewRewriter(QueryContext context, SchemaPlus rootSchema, 
SqlConverter sqlConverter) {
+    this.context = context;
+    this.rootSchema = rootSchema;
+    this.sqlConverter = sqlConverter;
+  }
+
+  /**
+   * Attempts to rewrite the given RelNode to use a materialized view.
+   *
+   * @param queryRel the query plan to potentially rewrite
+   * @return the rewritten plan using an MV, or the original plan if no 
rewrite is possible
+   */
+  public RelNode rewrite(RelNode queryRel) {
+    if (!context.getPlannerSettings().isMaterializedViewRewriteEnabled()) {
+      return queryRel;
+    }
+
+    // Find all available materialized views that have been refreshed
+    List<MaterializedViewCandidate> candidates = 
findCandidateMaterializedViews();
+
+    if (candidates.isEmpty()) {
+      logger.debug("No refreshed materialized views available for rewriting");
+      return queryRel;
+    }
+
+    logger.debug("Found {} materialized view candidates for potential 
rewriting", candidates.size());
+
+    // Build Calcite RelOptMaterialization objects for each refreshed candidate
+    List<RelOptMaterialization> materializations = new ArrayList<>();
+    for (MaterializedViewCandidate candidate : candidates) {
+      if (!candidate.isRefreshed()) {
+        logger.debug("Skipping MV {} - not refreshed", candidate.getName());
+        continue;
+      }
+
+      try {
+        RelOptMaterialization mat = buildMaterialization(candidate);
+        if (mat != null) {
+          materializations.add(mat);
+        }
+      } catch (Exception e) {
+        logger.debug("Failed to build materialization for MV {}: {}", 
candidate.getName(), e.getMessage());
+      }
+    }
+
+    if (materializations.isEmpty()) {
+      logger.debug("No valid materializations could be built");
+      return queryRel;
+    }
+
+    // Use Calcite's materialized view matching API which normalizes both the
+    // query and MV definitions (trimming unused fields, converting 
Filter/Project
+    // to Calc, merging, etc.) before performing structural matching.
+    try {
+      List<Pair<RelNode, List<RelOptMaterialization>>> results =
+          RelOptMaterializations.useMaterializedViews(queryRel, 
materializations);
+
+      if (!results.isEmpty()) {
+        RelNode rewritten = results.get(0).left;
+        if (logger.isInfoEnabled()) {
+          List<RelOptMaterialization> usedMVs = results.get(0).right;
+          logger.info("Query rewritten to use materialized view(s): {}",
+              !usedMVs.isEmpty() ? usedMVs.get(0).qualifiedTableName : 
"unknown");
+        }
+        return rewritten;
+      }
+    } catch (Exception e) {
+      logger.debug("Materialized view rewriting failed: {}", e.getMessage());
+    }
+
+    logger.debug("No materialized view matched the query");
+    return queryRel;
+  }
+
+  /**
+   * Builds a Calcite {@link RelOptMaterialization} for a candidate MV.
+   */
+  private RelOptMaterialization buildMaterialization(MaterializedViewCandidate 
candidate) {
+    RelNode mvQueryRel = parseMvSql(candidate);
+    if (mvQueryRel == null) {
+      return null;
+    }
+
+    RelNode mvTableRel = buildMvScanRel(candidate);
+    if (mvTableRel == null) {
+      return null;
+    }
+
+    List<String> qualifiedTableName = java.util.Arrays.asList(

Review Comment:
   Is the fully qualified name used intentionally?



##########
exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/WorkspaceSchemaFactory.java:
##########
@@ -351,6 +354,278 @@ public void dropView(String viewName) throws IOException {
       getFS().delete(getViewPath(viewName), false);
     }
 
+    private Path getMaterializedViewPath(String name) {
+      return DotDrillType.MATERIALIZED_VIEW.getPath(config.getLocation(), 
name);
+    }
+
+    private Path getMaterializedViewDataPath(String name) {
+      // Use _mv_data suffix to distinguish data directory from MV definition 
lookup
+      return new Path(config.getLocation(), name + "_mv_data");
+    }
+
+    @Override
+    public boolean createMaterializedView(MaterializedView materializedView) 
throws IOException {
+      String viewName = materializedView.getName();
+      Path viewPath = getMaterializedViewPath(viewName);
+      Path dataPath = getMaterializedViewDataPath(viewName);

Review Comment:
   Use `materializedView.getDataStoragePath()` instead.



##########
exec/java-exec/src/main/java/org/apache/drill/exec/store/ischema/RecordCollector.java:
##########
@@ -181,6 +191,33 @@ public List<Records.View> views(String schemaPath, 
SchemaPlus schema) {
         .collect(Collectors.toList());
     }
 
+    @Override
+    public List<Records.MaterializedView> materializedViews(String schemaPath, 
SchemaPlus schema) {
+      AbstractSchema drillSchema = schema.unwrap(AbstractSchema.class);
+
+      return drillSchema.getTablesByNames(schema.getTableNames()).stream()
+        .filter(pair -> pair.getValue().getJdbcTableType() == 
Schema.TableType.MATERIALIZED_VIEW)
+        .filter(pair -> filterEvaluator.shouldVisitTable(schemaPath, 
pair.getKey(), pair.getValue().getJdbcTableType()))
+        .map(pair -> {
+          Table table = pair.getValue();
+          String viewSql = table instanceof DrillViewInfoProvider
+              ? ((DrillViewInfoProvider) table).getViewSql() : "";
+          String refreshStatus = null;
+          Long lastRefreshTime = null;
+          String dataLocation = null;
+          if (table instanceof DrillMaterializedViewTable) {
+            DrillMaterializedViewTable mvTable = (DrillMaterializedViewTable) 
table;

Review Comment:
   Use pattern matching feature: 
https://docs.oracle.com/en/java/javase/17/language/pattern-matching-instanceof.html.
 Drill use Java 17+



##########
exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/WorkspaceSchemaFactory.java:
##########
@@ -351,6 +354,278 @@ public void dropView(String viewName) throws IOException {
       getFS().delete(getViewPath(viewName), false);
     }
 
+    private Path getMaterializedViewPath(String name) {
+      return DotDrillType.MATERIALIZED_VIEW.getPath(config.getLocation(), 
name);
+    }
+
+    private Path getMaterializedViewDataPath(String name) {
+      // Use _mv_data suffix to distinguish data directory from MV definition 
lookup
+      return new Path(config.getLocation(), name + "_mv_data");
+    }
+
+    @Override
+    public boolean createMaterializedView(MaterializedView materializedView) 
throws IOException {
+      String viewName = materializedView.getName();
+      Path viewPath = getMaterializedViewPath(viewName);
+      Path dataPath = getMaterializedViewDataPath(viewName);
+
+      boolean replaced = getFS().exists(viewPath);
+
+      // If replacing, first drop the old data
+      if (replaced) {
+        if (getFS().exists(dataPath)) {
+          getFS().delete(dataPath, true);
+        }
+      }
+
+      // Create the data directory for the materialized view
+      final FsPermission dirPerms = new FsPermission(
+          
schemaConfig.getOption(ExecConstants.NEW_VIEW_DEFAULT_PERMS_KEY).string_val);
+      getFS().mkdirs(dataPath, dirPerms);
+
+      // Set the data storage path in the materialized view
+      materializedView.setDataStoragePath(viewName);
+
+      // Write the materialized view definition file
+      final FsPermission viewPerms = new FsPermission(
+          
schemaConfig.getOption(ExecConstants.NEW_VIEW_DEFAULT_PERMS_KEY).string_val);
+      try (OutputStream stream = DrillFileSystem.create(getFS(), viewPath, 
viewPerms)) {
+        mapper.writeValue(stream, materializedView);
+      }
+
+      // Sync to metastore if enabled
+      syncMaterializedViewToMetastore(materializedView);
+
+      // Mark as complete (data will be populated by the handler via CTAS-like 
operation)
+      return replaced;
+    }
+
+    @Override
+    public void dropMaterializedView(String viewName) throws IOException {
+      Path viewPath = getMaterializedViewPath(viewName);
+      Path dataPath = getMaterializedViewDataPath(viewName);
+
+      // Delete the definition file
+      if (getFS().exists(viewPath)) {
+        getFS().delete(viewPath, false);
+      }
+
+      // Delete the data directory
+      if (getFS().exists(dataPath)) {
+        getFS().delete(dataPath, true);
+      }
+
+      // Remove from metastore if enabled
+      removeMaterializedViewFromMetastore(viewName);
+    }
+
+    @Override
+    public void refreshMaterializedView(String viewName) throws IOException {
+      // Read the existing materialized view definition
+      MaterializedView mv = getMaterializedView(viewName);
+      if (mv == null) {
+        throw UserException.validationError()
+            .message("Materialized view [%s] not found in schema [%s]", 
viewName, getFullSchemaName())
+            .build(logger);
+      }
+
+      Path dataPath = getMaterializedViewDataPath(viewName);

Review Comment:
   Use `materializedView.getDataStoragePath()` instead.



##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillMaterializedViewTable.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ */
+package org.apache.drill.exec.planner.logical;
+
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.calcite.config.CalciteConnectionConfig;
+import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.plan.RelOptTable.ToRelContext;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.schema.Schema.TableType;
+import org.apache.calcite.schema.Statistic;
+import org.apache.calcite.schema.Statistics;
+import org.apache.calcite.schema.TranslatableTable;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.drill.exec.dotdrill.MaterializedView;
+import org.apache.drill.exec.ops.ViewExpansionContext;
+import org.apache.drill.exec.planner.sql.conversion.DrillViewExpander;
+
+/**
+ * Represents a materialized view in the Drill query planning.
+ * <p>
+ * A materialized view stores:
+ * <ul>
+ *   <li>Definition file (.materialized_view.drill) - JSON with name, SQL, 
schema info</li>
+ *   <li>Data directory ({name}_mv_data/) - Parquet files with pre-computed 
results</li>
+ * </ul>
+ * <p>
+ * <b>Behavior:</b>
+ * <ul>
+ *   <li>Before REFRESH: queries expand the SQL definition (like a view)</li>
+ *   <li>After REFRESH: queries scan from pre-computed Parquet data</li>
+ * </ul>
+ *
+ * @see org.apache.drill.exec.dotdrill.MaterializedView
+ */
+public class DrillMaterializedViewTable implements TranslatableTable, 
DrillViewInfoProvider {
+
+  private final MaterializedView materializedView;
+  private final String viewOwner;
+  private final ViewExpansionContext viewExpansionContext;
+  private final String workspaceLocation;
+
+  public DrillMaterializedViewTable(MaterializedView materializedView, String 
viewOwner,
+                                    ViewExpansionContext viewExpansionContext, 
String workspaceLocation) {
+    this.materializedView = materializedView;
+    this.viewOwner = viewOwner;
+    this.viewExpansionContext = viewExpansionContext;
+    this.workspaceLocation = workspaceLocation;
+  }
+
+  @Override
+  public RelDataType getRowType(RelDataTypeFactory typeFactory) {
+    return materializedView.getRowType(typeFactory);
+  }
+
+  @Override
+  public Statistic getStatistic() {
+    return Statistics.UNKNOWN;
+  }
+
+  /**
+   * Converts this materialized view to a RelNode for query planning.
+   * <p>
+   * If the MV has been refreshed (data exists), scans from the pre-computed 
Parquet data.
+   * Otherwise, expands the SQL definition like a regular view.
+   */
+  @Override
+  public RelNode toRel(ToRelContext context, RelOptTable relOptTable) {
+    DrillViewExpander viewExpander = viewExpansionContext.getViewExpander();
+    ViewExpansionContext.ViewExpansionToken token = null;
+    try {
+      RelDataType rowType = relOptTable.getRowType();
+      RelNode rel;
+
+      // Check if materialized data exists (REFRESH has been called)
+      boolean hasData = materializedView.getRefreshStatus() == 
MaterializedView.RefreshStatus.COMPLETE;
+
+      // Build the SQL to execute - either scan data or expand definition
+      String sqlToExpand;
+      if (hasData) {
+        // Scan from the pre-computed data directory
+        sqlToExpand = buildDataScanSql();
+      } else {
+        // No data yet - expand the SQL definition like a view
+        sqlToExpand = materializedView.getSql();
+      }
+
+      // Always use the workspace schema path for context - needed for table 
resolution
+      List<String> schemaPath = materializedView.getWorkspaceSchemaPath();
+
+      if (viewExpansionContext.isImpersonationEnabled()) {
+        token = viewExpansionContext.reserveViewExpansionToken(viewOwner);
+        rel = viewExpander.expandView(sqlToExpand, token.getSchemaTree(), 
schemaPath).rel;
+      } else {
+        // When scanning data, pass null for rowType to let Parquet schema be 
inferred
+        // When expanding SQL definition, use the MV's row type
+        RelDataType typeHint = hasData ? null : rowType;
+        rel = viewExpander.expandView(typeHint, sqlToExpand, schemaPath, 
Collections.emptyList()).rel;
+      }
+
+      return rel;
+    } finally {
+      if (token != null) {
+        token.release();
+      }
+    }
+  }
+
+  /**
+   * Builds SQL to scan the materialized data directory.
+   * The data is stored in {workspace}/{mvName}_mv_data/ directory.
+   * We explicitly select the MV's columns to ensure proper schema matching.
+   */
+  private String buildDataScanSql() {
+    String dataTableName = materializedView.getName() + "_mv_data";
+
+    // Build explicit column list from the MV's field definitions
+    List<String> fieldNames = materializedView.getFields().stream()
+        .map(f -> f.getName())
+        .collect(java.util.stream.Collectors.toList());
+    if (fieldNames.isEmpty()) {
+      // Fallback to SELECT * if no fields defined (shouldn't happen for 
non-dynamic MVs)
+      return "SELECT * FROM `" + dataTableName + "`";

Review Comment:
   Apache Drill allows to configure the identifier quote character. Won't it 
fail if `planner.parser.quoting_identifiers` is set to double quotes?
   https://drill.apache.org/docs/lexical-structure#identifier-quotes
   
   And doesn't it should include either the workspace name?



##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillMaterializedViewTable.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ */
+package org.apache.drill.exec.planner.logical;
+
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.calcite.config.CalciteConnectionConfig;
+import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.plan.RelOptTable.ToRelContext;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.schema.Schema.TableType;
+import org.apache.calcite.schema.Statistic;
+import org.apache.calcite.schema.Statistics;
+import org.apache.calcite.schema.TranslatableTable;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.drill.exec.dotdrill.MaterializedView;
+import org.apache.drill.exec.ops.ViewExpansionContext;
+import org.apache.drill.exec.planner.sql.conversion.DrillViewExpander;
+
+/**
+ * Represents a materialized view in the Drill query planning.
+ * <p>
+ * A materialized view stores:
+ * <ul>
+ *   <li>Definition file (.materialized_view.drill) - JSON with name, SQL, 
schema info</li>
+ *   <li>Data directory ({name}_mv_data/) - Parquet files with pre-computed 
results</li>
+ * </ul>
+ * <p>
+ * <b>Behavior:</b>
+ * <ul>
+ *   <li>Before REFRESH: queries expand the SQL definition (like a view)</li>
+ *   <li>After REFRESH: queries scan from pre-computed Parquet data</li>
+ * </ul>
+ *
+ * @see org.apache.drill.exec.dotdrill.MaterializedView
+ */
+public class DrillMaterializedViewTable implements TranslatableTable, 
DrillViewInfoProvider {
+
+  private final MaterializedView materializedView;
+  private final String viewOwner;
+  private final ViewExpansionContext viewExpansionContext;
+  private final String workspaceLocation;
+
+  public DrillMaterializedViewTable(MaterializedView materializedView, String 
viewOwner,
+                                    ViewExpansionContext viewExpansionContext, 
String workspaceLocation) {
+    this.materializedView = materializedView;
+    this.viewOwner = viewOwner;
+    this.viewExpansionContext = viewExpansionContext;
+    this.workspaceLocation = workspaceLocation;
+  }
+
+  @Override
+  public RelDataType getRowType(RelDataTypeFactory typeFactory) {
+    return materializedView.getRowType(typeFactory);
+  }
+
+  @Override
+  public Statistic getStatistic() {
+    return Statistics.UNKNOWN;
+  }
+
+  /**
+   * Converts this materialized view to a RelNode for query planning.
+   * <p>
+   * If the MV has been refreshed (data exists), scans from the pre-computed 
Parquet data.
+   * Otherwise, expands the SQL definition like a regular view.
+   */
+  @Override
+  public RelNode toRel(ToRelContext context, RelOptTable relOptTable) {
+    DrillViewExpander viewExpander = viewExpansionContext.getViewExpander();
+    ViewExpansionContext.ViewExpansionToken token = null;
+    try {
+      RelDataType rowType = relOptTable.getRowType();
+      RelNode rel;
+
+      // Check if materialized data exists (REFRESH has been called)
+      boolean hasData = materializedView.getRefreshStatus() == 
MaterializedView.RefreshStatus.COMPLETE;
+
+      // Build the SQL to execute - either scan data or expand definition
+      String sqlToExpand;
+      if (hasData) {
+        // Scan from the pre-computed data directory
+        sqlToExpand = buildDataScanSql();
+      } else {
+        // No data yet - expand the SQL definition like a view
+        sqlToExpand = materializedView.getSql();
+      }
+
+      // Always use the workspace schema path for context - needed for table 
resolution
+      List<String> schemaPath = materializedView.getWorkspaceSchemaPath();
+
+      if (viewExpansionContext.isImpersonationEnabled()) {
+        token = viewExpansionContext.reserveViewExpansionToken(viewOwner);
+        rel = viewExpander.expandView(sqlToExpand, token.getSchemaTree(), 
schemaPath).rel;
+      } else {
+        // When scanning data, pass null for rowType to let Parquet schema be 
inferred
+        // When expanding SQL definition, use the MV's row type
+        RelDataType typeHint = hasData ? null : rowType;
+        rel = viewExpander.expandView(typeHint, sqlToExpand, schemaPath, 
Collections.emptyList()).rel;
+      }
+
+      return rel;
+    } finally {
+      if (token != null) {
+        token.release();
+      }
+    }
+  }
+
+  /**
+   * Builds SQL to scan the materialized data directory.
+   * The data is stored in {workspace}/{mvName}_mv_data/ directory.
+   * We explicitly select the MV's columns to ensure proper schema matching.
+   */
+  private String buildDataScanSql() {
+    String dataTableName = materializedView.getName() + "_mv_data";
+
+    // Build explicit column list from the MV's field definitions
+    List<String> fieldNames = materializedView.getFields().stream()
+        .map(f -> f.getName())
+        .collect(java.util.stream.Collectors.toList());
+    if (fieldNames.isEmpty()) {
+      // Fallback to SELECT * if no fields defined (shouldn't happen for 
non-dynamic MVs)
+      return "SELECT * FROM `" + dataTableName + "`";
+    }
+
+    StringBuilder sql = new StringBuilder("SELECT ");
+    for (int i = 0; i < fieldNames.size(); i++) {
+      if (i > 0) {
+        sql.append(", ");
+      }
+      sql.append("`").append(fieldNames.get(i)).append("`");
+    }
+    sql.append(" FROM `").append(dataTableName).append("`");

Review Comment:
   The same:
   > Won't it fail if planner.parser.quoting_identifiers is set to double 
quotes?
   https://drill.apache.org/docs/lexical-structure#identifier-quotes



##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/MaterializedViewRewriter.java:
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ */
+package org.apache.drill.exec.planner.logical;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.calcite.plan.RelOptMaterialization;
+import org.apache.calcite.plan.RelOptMaterializations;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelRoot;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.util.Pair;
+import org.apache.drill.exec.dotdrill.MaterializedView;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.planner.sql.conversion.SqlConverter;
+import org.apache.drill.exec.store.AbstractSchema;
+import org.apache.drill.exec.store.StoragePlugin;
+import org.apache.drill.exec.store.StoragePluginRegistry;
+import org.apache.drill.exec.store.StoragePluginRegistry.PluginException;
+import org.apache.drill.exec.store.dfs.FileSystemPlugin;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Utility class for materialized view query rewriting.
+ * <p>
+ * When enabled via planner.enable_materialized_view_rewrite, this class 
attempts
+ * to rewrite queries to use materialized views when beneficial.
+ * <p>
+ * Uses Calcite's {@link RelOptMaterializations#useMaterializedViews} API which
+ * normalizes both the query and MV definitions before performing structural
+ * matching via {@link org.apache.calcite.plan.SubstitutionVisitor}.
+ * <p>
+ * Materialized views are discovered by iterating over enabled file-based
+ * storage plugins (the only plugin type that supports MVs) and force-loading
+ * their schemas to find .materialized_view.drill files.
+ */
+public class MaterializedViewRewriter {
+  private static final Logger logger = 
LoggerFactory.getLogger(MaterializedViewRewriter.class);
+
+  private final QueryContext context;
+  private final SchemaPlus rootSchema;
+  private final SqlConverter sqlConverter;
+
+  public MaterializedViewRewriter(QueryContext context, SchemaPlus rootSchema, 
SqlConverter sqlConverter) {
+    this.context = context;
+    this.rootSchema = rootSchema;
+    this.sqlConverter = sqlConverter;
+  }
+
+  /**
+   * Attempts to rewrite the given RelNode to use a materialized view.
+   *
+   * @param queryRel the query plan to potentially rewrite
+   * @return the rewritten plan using an MV, or the original plan if no 
rewrite is possible
+   */
+  public RelNode rewrite(RelNode queryRel) {
+    if (!context.getPlannerSettings().isMaterializedViewRewriteEnabled()) {
+      return queryRel;
+    }
+
+    // Find all available materialized views that have been refreshed
+    List<MaterializedViewCandidate> candidates = 
findCandidateMaterializedViews();
+
+    if (candidates.isEmpty()) {
+      logger.debug("No refreshed materialized views available for rewriting");
+      return queryRel;
+    }
+
+    logger.debug("Found {} materialized view candidates for potential 
rewriting", candidates.size());
+
+    // Build Calcite RelOptMaterialization objects for each refreshed candidate
+    List<RelOptMaterialization> materializations = new ArrayList<>();
+    for (MaterializedViewCandidate candidate : candidates) {
+      if (!candidate.isRefreshed()) {
+        logger.debug("Skipping MV {} - not refreshed", candidate.getName());
+        continue;
+      }
+
+      try {
+        RelOptMaterialization mat = buildMaterialization(candidate);
+        if (mat != null) {
+          materializations.add(mat);
+        }
+      } catch (Exception e) {
+        logger.debug("Failed to build materialization for MV {}: {}", 
candidate.getName(), e.getMessage());
+      }
+    }
+
+    if (materializations.isEmpty()) {
+      logger.debug("No valid materializations could be built");
+      return queryRel;
+    }
+
+    // Use Calcite's materialized view matching API which normalizes both the
+    // query and MV definitions (trimming unused fields, converting 
Filter/Project
+    // to Calc, merging, etc.) before performing structural matching.
+    try {
+      List<Pair<RelNode, List<RelOptMaterialization>>> results =
+          RelOptMaterializations.useMaterializedViews(queryRel, 
materializations);
+
+      if (!results.isEmpty()) {
+        RelNode rewritten = results.get(0).left;
+        if (logger.isInfoEnabled()) {
+          List<RelOptMaterialization> usedMVs = results.get(0).right;
+          logger.info("Query rewritten to use materialized view(s): {}",
+              !usedMVs.isEmpty() ? usedMVs.get(0).qualifiedTableName : 
"unknown");
+        }
+        return rewritten;
+      }
+    } catch (Exception e) {
+      logger.debug("Materialized view rewriting failed: {}", e.getMessage());
+    }
+
+    logger.debug("No materialized view matched the query");
+    return queryRel;
+  }
+
+  /**
+   * Builds a Calcite {@link RelOptMaterialization} for a candidate MV.
+   */
+  private RelOptMaterialization buildMaterialization(MaterializedViewCandidate 
candidate) {
+    RelNode mvQueryRel = parseMvSql(candidate);
+    if (mvQueryRel == null) {
+      return null;
+    }
+
+    RelNode mvTableRel = buildMvScanRel(candidate);
+    if (mvTableRel == null) {
+      return null;
+    }
+
+    List<String> qualifiedTableName = java.util.Arrays.asList(
+        candidate.getSchemaPath().split("\\."));
+
+    return new RelOptMaterialization(
+        mvTableRel,
+        mvQueryRel,
+        null,
+        qualifiedTableName
+    );
+  }
+
+  /**
+   * Parses the MV's SQL definition into a RelNode.
+   */
+  private RelNode parseMvSql(MaterializedViewCandidate candidate) {
+    try {
+      String mvSql = candidate.getSql();
+      org.apache.calcite.sql.SqlNode parsedNode = sqlConverter.parse(mvSql);
+      org.apache.calcite.sql.SqlNode validatedNode = 
sqlConverter.validate(parsedNode);

Review Comment:
   Is the fully qualified name used intentionally?



##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillMaterializedViewTable.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ */
+package org.apache.drill.exec.planner.logical;
+
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.calcite.config.CalciteConnectionConfig;
+import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.plan.RelOptTable.ToRelContext;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.schema.Schema.TableType;
+import org.apache.calcite.schema.Statistic;
+import org.apache.calcite.schema.Statistics;
+import org.apache.calcite.schema.TranslatableTable;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.drill.exec.dotdrill.MaterializedView;
+import org.apache.drill.exec.ops.ViewExpansionContext;
+import org.apache.drill.exec.planner.sql.conversion.DrillViewExpander;
+
+/**
+ * Represents a materialized view in the Drill query planning.
+ * <p>
+ * A materialized view stores:
+ * <ul>
+ *   <li>Definition file (.materialized_view.drill) - JSON with name, SQL, 
schema info</li>
+ *   <li>Data directory ({name}_mv_data/) - Parquet files with pre-computed 
results</li>
+ * </ul>
+ * <p>
+ * <b>Behavior:</b>
+ * <ul>
+ *   <li>Before REFRESH: queries expand the SQL definition (like a view)</li>
+ *   <li>After REFRESH: queries scan from pre-computed Parquet data</li>
+ * </ul>
+ *
+ * @see org.apache.drill.exec.dotdrill.MaterializedView
+ */
+public class DrillMaterializedViewTable implements TranslatableTable, 
DrillViewInfoProvider {
+
+  private final MaterializedView materializedView;
+  private final String viewOwner;
+  private final ViewExpansionContext viewExpansionContext;
+  private final String workspaceLocation;
+
+  public DrillMaterializedViewTable(MaterializedView materializedView, String 
viewOwner,
+                                    ViewExpansionContext viewExpansionContext, 
String workspaceLocation) {
+    this.materializedView = materializedView;
+    this.viewOwner = viewOwner;
+    this.viewExpansionContext = viewExpansionContext;
+    this.workspaceLocation = workspaceLocation;
+  }
+
+  @Override
+  public RelDataType getRowType(RelDataTypeFactory typeFactory) {
+    return materializedView.getRowType(typeFactory);
+  }
+
+  @Override
+  public Statistic getStatistic() {
+    return Statistics.UNKNOWN;
+  }
+
+  /**
+   * Converts this materialized view to a RelNode for query planning.
+   * <p>
+   * If the MV has been refreshed (data exists), scans from the pre-computed 
Parquet data.
+   * Otherwise, expands the SQL definition like a regular view.
+   */
+  @Override
+  public RelNode toRel(ToRelContext context, RelOptTable relOptTable) {
+    DrillViewExpander viewExpander = viewExpansionContext.getViewExpander();
+    ViewExpansionContext.ViewExpansionToken token = null;
+    try {
+      RelDataType rowType = relOptTable.getRowType();
+      RelNode rel;
+
+      // Check if materialized data exists (REFRESH has been called)
+      boolean hasData = materializedView.getRefreshStatus() == 
MaterializedView.RefreshStatus.COMPLETE;
+
+      // Build the SQL to execute - either scan data or expand definition
+      String sqlToExpand;
+      if (hasData) {
+        // Scan from the pre-computed data directory
+        sqlToExpand = buildDataScanSql();
+      } else {
+        // No data yet - expand the SQL definition like a view
+        sqlToExpand = materializedView.getSql();
+      }
+
+      // Always use the workspace schema path for context - needed for table 
resolution
+      List<String> schemaPath = materializedView.getWorkspaceSchemaPath();
+
+      if (viewExpansionContext.isImpersonationEnabled()) {
+        token = viewExpansionContext.reserveViewExpansionToken(viewOwner);
+        rel = viewExpander.expandView(sqlToExpand, token.getSchemaTree(), 
schemaPath).rel;
+      } else {
+        // When scanning data, pass null for rowType to let Parquet schema be 
inferred
+        // When expanding SQL definition, use the MV's row type
+        RelDataType typeHint = hasData ? null : rowType;
+        rel = viewExpander.expandView(typeHint, sqlToExpand, schemaPath, 
Collections.emptyList()).rel;
+      }
+
+      return rel;
+    } finally {
+      if (token != null) {
+        token.release();
+      }
+    }
+  }
+
+  /**
+   * Builds SQL to scan the materialized data directory.
+   * The data is stored in {workspace}/{mvName}_mv_data/ directory.
+   * We explicitly select the MV's columns to ensure proper schema matching.
+   */
+  private String buildDataScanSql() {
+    String dataTableName = materializedView.getName() + "_mv_data";

Review Comment:
   I believe `materializedView.getDataStoragePath()` should be used there.



##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/MaterializedViewRewriter.java:
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ */
+package org.apache.drill.exec.planner.logical;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.calcite.plan.RelOptMaterialization;
+import org.apache.calcite.plan.RelOptMaterializations;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelRoot;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.util.Pair;
+import org.apache.drill.exec.dotdrill.MaterializedView;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.planner.sql.conversion.SqlConverter;
+import org.apache.drill.exec.store.AbstractSchema;
+import org.apache.drill.exec.store.StoragePlugin;
+import org.apache.drill.exec.store.StoragePluginRegistry;
+import org.apache.drill.exec.store.StoragePluginRegistry.PluginException;
+import org.apache.drill.exec.store.dfs.FileSystemPlugin;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Utility class for materialized view query rewriting.
+ * <p>
+ * When enabled via planner.enable_materialized_view_rewrite, this class 
attempts
+ * to rewrite queries to use materialized views when beneficial.
+ * <p>
+ * Uses Calcite's {@link RelOptMaterializations#useMaterializedViews} API which
+ * normalizes both the query and MV definitions before performing structural
+ * matching via {@link org.apache.calcite.plan.SubstitutionVisitor}.
+ * <p>
+ * Materialized views are discovered by iterating over enabled file-based
+ * storage plugins (the only plugin type that supports MVs) and force-loading
+ * their schemas to find .materialized_view.drill files.
+ */
+public class MaterializedViewRewriter {
+  private static final Logger logger = 
LoggerFactory.getLogger(MaterializedViewRewriter.class);
+
+  private final QueryContext context;
+  private final SchemaPlus rootSchema;
+  private final SqlConverter sqlConverter;
+
+  public MaterializedViewRewriter(QueryContext context, SchemaPlus rootSchema, 
SqlConverter sqlConverter) {
+    this.context = context;
+    this.rootSchema = rootSchema;
+    this.sqlConverter = sqlConverter;
+  }
+
+  /**
+   * Attempts to rewrite the given RelNode to use a materialized view.
+   *
+   * @param queryRel the query plan to potentially rewrite
+   * @return the rewritten plan using an MV, or the original plan if no 
rewrite is possible
+   */
+  public RelNode rewrite(RelNode queryRel) {
+    if (!context.getPlannerSettings().isMaterializedViewRewriteEnabled()) {
+      return queryRel;
+    }
+
+    // Find all available materialized views that have been refreshed
+    List<MaterializedViewCandidate> candidates = 
findCandidateMaterializedViews();
+
+    if (candidates.isEmpty()) {
+      logger.debug("No refreshed materialized views available for rewriting");
+      return queryRel;
+    }
+
+    logger.debug("Found {} materialized view candidates for potential 
rewriting", candidates.size());
+
+    // Build Calcite RelOptMaterialization objects for each refreshed candidate
+    List<RelOptMaterialization> materializations = new ArrayList<>();
+    for (MaterializedViewCandidate candidate : candidates) {
+      if (!candidate.isRefreshed()) {
+        logger.debug("Skipping MV {} - not refreshed", candidate.getName());
+        continue;
+      }
+
+      try {
+        RelOptMaterialization mat = buildMaterialization(candidate);
+        if (mat != null) {
+          materializations.add(mat);
+        }
+      } catch (Exception e) {
+        logger.debug("Failed to build materialization for MV {}: {}", 
candidate.getName(), e.getMessage());
+      }
+    }
+
+    if (materializations.isEmpty()) {
+      logger.debug("No valid materializations could be built");
+      return queryRel;
+    }
+
+    // Use Calcite's materialized view matching API which normalizes both the
+    // query and MV definitions (trimming unused fields, converting 
Filter/Project
+    // to Calc, merging, etc.) before performing structural matching.
+    try {
+      List<Pair<RelNode, List<RelOptMaterialization>>> results =
+          RelOptMaterializations.useMaterializedViews(queryRel, 
materializations);
+
+      if (!results.isEmpty()) {
+        RelNode rewritten = results.get(0).left;
+        if (logger.isInfoEnabled()) {
+          List<RelOptMaterialization> usedMVs = results.get(0).right;
+          logger.info("Query rewritten to use materialized view(s): {}",
+              !usedMVs.isEmpty() ? usedMVs.get(0).qualifiedTableName : 
"unknown");
+        }
+        return rewritten;
+      }
+    } catch (Exception e) {
+      logger.debug("Materialized view rewriting failed: {}", e.getMessage());
+    }
+
+    logger.debug("No materialized view matched the query");
+    return queryRel;
+  }
+
+  /**
+   * Builds a Calcite {@link RelOptMaterialization} for a candidate MV.
+   */
+  private RelOptMaterialization buildMaterialization(MaterializedViewCandidate 
candidate) {
+    RelNode mvQueryRel = parseMvSql(candidate);
+    if (mvQueryRel == null) {
+      return null;
+    }
+
+    RelNode mvTableRel = buildMvScanRel(candidate);
+    if (mvTableRel == null) {
+      return null;
+    }
+
+    List<String> qualifiedTableName = java.util.Arrays.asList(
+        candidate.getSchemaPath().split("\\."));
+
+    return new RelOptMaterialization(
+        mvTableRel,
+        mvQueryRel,
+        null,
+        qualifiedTableName
+    );
+  }
+
+  /**
+   * Parses the MV's SQL definition into a RelNode.
+   */
+  private RelNode parseMvSql(MaterializedViewCandidate candidate) {
+    try {
+      String mvSql = candidate.getSql();
+      org.apache.calcite.sql.SqlNode parsedNode = sqlConverter.parse(mvSql);
+      org.apache.calcite.sql.SqlNode validatedNode = 
sqlConverter.validate(parsedNode);
+      RelRoot relRoot = sqlConverter.toRel(validatedNode);
+      return relRoot.rel;
+    } catch (Exception e) {
+      logger.debug("Failed to parse MV SQL for {}: {}", candidate.getName(), 
e.getMessage());
+      return null;
+    }
+  }
+
+  /**
+   * Builds a RelNode that scans the MV's pre-computed data.
+   */
+  private RelNode buildMvScanRel(MaterializedViewCandidate candidate) {
+    try {
+      String mvDataTable = candidate.getSchemaPath() + ".`" + 
candidate.getName() + "_mv_data`";

Review Comment:
   I believe `materializedView.getDataStoragePath()` should be used there. 
   
   And also:
   > Won't it fail if `planner.parser.quoting_identifiers` is set to double 
quotes?
       https://drill.apache.org/docs/lexical-structure#identifier-quotes
   
   



##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/MaterializedViewHandler.java:
##########
@@ -0,0 +1,327 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ */
+package org.apache.drill.exec.planner.sql.handlers;
+
+import java.io.IOException;
+
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.schema.Schema;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.schema.Table;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.tools.RelConversionException;
+import org.apache.calcite.tools.ValidationException;
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.util.DrillStringUtils;
+import org.apache.drill.exec.dotdrill.MaterializedView;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.PhysicalPlan;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.planner.logical.CreateTableEntry;
+import org.apache.drill.exec.planner.logical.DrillRel;
+import org.apache.drill.exec.planner.logical.DrillScreenRel;
+import org.apache.drill.exec.planner.logical.DrillWriterRel;
+import org.apache.drill.exec.planner.physical.Prel;
+import org.apache.drill.exec.planner.sql.DirectPlan;
+import org.apache.drill.exec.planner.sql.SchemaUtilities;
+import org.apache.drill.exec.planner.sql.parser.SqlCreateMaterializedView;
+import org.apache.drill.exec.planner.sql.parser.SqlCreateType;
+import org.apache.drill.exec.planner.sql.parser.SqlDropMaterializedView;
+import org.apache.drill.exec.planner.sql.parser.SqlRefreshMaterializedView;
+import org.apache.drill.exec.store.AbstractSchema;
+import org.apache.drill.exec.work.foreman.ForemanSetupException;
+import org.apache.drill.exec.work.foreman.SqlUnsupportedException;
+
+/**
+ * Handlers for materialized view DDL commands: CREATE, DROP, and REFRESH 
MATERIALIZED VIEW.
+ * <p>
+ * CREATE and DROP return DirectPlan with ok/summary output.
+ * REFRESH executes the MV query and writes data to Parquet, returning write 
statistics.
+ */
+public abstract class MaterializedViewHandler extends DefaultSqlHandler {
+  private static final org.slf4j.Logger logger = 
org.slf4j.LoggerFactory.getLogger(MaterializedViewHandler.class);
+
+  protected QueryContext context;
+
+  public MaterializedViewHandler(SqlHandlerConfig config) {
+    super(config);
+    this.context = config.getContext();
+  }
+
+  /**
+   * Handler for CREATE MATERIALIZED VIEW DDL command.
+   * <p>
+   * Creates the MV definition file. The data will be materialized on first 
query
+   * or can be explicitly populated via REFRESH MATERIALIZED VIEW.
+   */
+  public static class CreateMaterializedView extends MaterializedViewHandler {
+
+    public CreateMaterializedView(SqlHandlerConfig config) {
+      super(config);
+    }
+
+    @Override
+    public PhysicalPlan getPlan(SqlNode sqlNode) throws ValidationException, 
RelConversionException,
+        IOException, ForemanSetupException {
+      SqlCreateMaterializedView createMV = unwrap(sqlNode, 
SqlCreateMaterializedView.class);
+
+      final String newViewName = 
DrillStringUtils.removeLeadingSlash(createMV.getName());
+
+      // Disallow temporary tables usage in materialized view definition
+      config.getConverter().disallowTemporaryTables();
+
+      // Store the SQL as the view definition
+      final String viewSql = createMV.getQuery().toSqlString(null, 
true).getSql();
+      final ConvertedRelNode convertedRelNode = 
validateAndConvert(createMV.getQuery());
+      final RelDataType validatedRowType = 
convertedRelNode.getValidatedRowType();
+      final RelNode queryRelNode = convertedRelNode.getConvertedNode();
+
+      final RelNode newViewRelNode = SqlHandlerUtil.resolveNewTableRel(true, 
createMV.getFieldNames(),
+          validatedRowType, queryRelNode);
+
+      final SchemaPlus defaultSchema = context.getNewDefaultSchema();
+      final AbstractSchema drillSchema = 
SchemaUtilities.resolveToMutableDrillSchema(defaultSchema,
+          createMV.getSchemaPath());
+
+      final String schemaPath = drillSchema.getFullSchemaName();
+
+      // Check view creation possibility
+      if (!checkMaterializedViewCreationPossibility(drillSchema, createMV, 
context)) {
+        return DirectPlan.createDirectPlan(context, false,
+            String.format("A table or view with given name [%s] already exists 
in schema [%s]",
+                newViewName, schemaPath));
+      }
+
+      // Create the materialized view definition
+      // Use the actual schema path where the MV is created (not the session's 
default schema)
+      final MaterializedView materializedView = new 
MaterializedView(newViewName, viewSql,
+          newViewRelNode.getRowType(), drillSchema.getSchemaPath());
+
+      // Create the materialized view definition file
+      final boolean replaced = 
drillSchema.createMaterializedView(materializedView);
+
+      String message = replaced
+          ? String.format("Materialized view '%s' replaced successfully in 
'%s' schema", newViewName, schemaPath)
+          : String.format("Materialized view '%s' created successfully in '%s' 
schema", newViewName, schemaPath);
+
+      logger.info("Created materialized view [{}] in schema [{}]", 
newViewName, schemaPath);
+      return DirectPlan.createDirectPlan(context, true, message);
+    }
+
+    /**
+     * Validates if materialized view can be created in indicated schema.
+     */
+    private boolean checkMaterializedViewCreationPossibility(AbstractSchema 
drillSchema,
+                                                              
SqlCreateMaterializedView createMV,
+                                                              QueryContext 
context) {
+      final String schemaPath = drillSchema.getFullSchemaName();
+      final String viewName = createMV.getName();
+      final Table table = SqlHandlerUtil.getTableFromSchema(drillSchema, 
viewName);
+
+      // Check if it's a materialized view
+      final boolean isMaterializedView = table != null &&
+          table.getJdbcTableType() == Schema.TableType.MATERIALIZED_VIEW;
+      final boolean isView = (table != null && table.getJdbcTableType() == 
Schema.TableType.VIEW);
+      // Regular table check excludes views and materialized views
+      final boolean isTable = (table != null
+          && table.getJdbcTableType() != Schema.TableType.VIEW
+          && table.getJdbcTableType() != Schema.TableType.MATERIALIZED_VIEW)
+          || context.getSession().isTemporaryTable(drillSchema, 
context.getConfig(), viewName);

Review Comment:
   Since we already verified whether the table is materialized or a simple 
view, we can rewrite it with:
   ```java
   final boolean isTable = (!isView && !isMaterializedView) 
       || context.getSession().isTemporaryTable(drillSchema, 
context.getConfig(), viewName);
   ```



##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/MaterializedViewRewriter.java:
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ */
+package org.apache.drill.exec.planner.logical;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.calcite.plan.RelOptMaterialization;
+import org.apache.calcite.plan.RelOptMaterializations;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelRoot;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.util.Pair;
+import org.apache.drill.exec.dotdrill.MaterializedView;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.planner.sql.conversion.SqlConverter;
+import org.apache.drill.exec.store.AbstractSchema;
+import org.apache.drill.exec.store.StoragePlugin;
+import org.apache.drill.exec.store.StoragePluginRegistry;
+import org.apache.drill.exec.store.StoragePluginRegistry.PluginException;
+import org.apache.drill.exec.store.dfs.FileSystemPlugin;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Utility class for materialized view query rewriting.
+ * <p>
+ * When enabled via planner.enable_materialized_view_rewrite, this class 
attempts
+ * to rewrite queries to use materialized views when beneficial.
+ * <p>
+ * Uses Calcite's {@link RelOptMaterializations#useMaterializedViews} API which
+ * normalizes both the query and MV definitions before performing structural
+ * matching via {@link org.apache.calcite.plan.SubstitutionVisitor}.
+ * <p>
+ * Materialized views are discovered by iterating over enabled file-based
+ * storage plugins (the only plugin type that supports MVs) and force-loading
+ * their schemas to find .materialized_view.drill files.
+ */
+public class MaterializedViewRewriter {
+  private static final Logger logger = 
LoggerFactory.getLogger(MaterializedViewRewriter.class);
+
+  private final QueryContext context;
+  private final SchemaPlus rootSchema;
+  private final SqlConverter sqlConverter;
+
+  public MaterializedViewRewriter(QueryContext context, SchemaPlus rootSchema, 
SqlConverter sqlConverter) {
+    this.context = context;
+    this.rootSchema = rootSchema;
+    this.sqlConverter = sqlConverter;
+  }
+
+  /**
+   * Attempts to rewrite the given RelNode to use a materialized view.
+   *
+   * @param queryRel the query plan to potentially rewrite
+   * @return the rewritten plan using an MV, or the original plan if no 
rewrite is possible
+   */
+  public RelNode rewrite(RelNode queryRel) {
+    if (!context.getPlannerSettings().isMaterializedViewRewriteEnabled()) {
+      return queryRel;
+    }
+
+    // Find all available materialized views that have been refreshed
+    List<MaterializedViewCandidate> candidates = 
findCandidateMaterializedViews();
+
+    if (candidates.isEmpty()) {
+      logger.debug("No refreshed materialized views available for rewriting");
+      return queryRel;
+    }

Review Comment:
   Since `findCandidateMaterializedViews()` returns both refreshed and 
unrefreshed MVs, the log message is misleading. 



##########
exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/WorkspaceSchemaFactory.java:
##########
@@ -351,6 +354,278 @@ public void dropView(String viewName) throws IOException {
       getFS().delete(getViewPath(viewName), false);
     }
 
+    private Path getMaterializedViewPath(String name) {
+      return DotDrillType.MATERIALIZED_VIEW.getPath(config.getLocation(), 
name);
+    }
+
+    private Path getMaterializedViewDataPath(String name) {
+      // Use _mv_data suffix to distinguish data directory from MV definition 
lookup
+      return new Path(config.getLocation(), name + "_mv_data");
+    }
+
+    @Override
+    public boolean createMaterializedView(MaterializedView materializedView) 
throws IOException {
+      String viewName = materializedView.getName();
+      Path viewPath = getMaterializedViewPath(viewName);
+      Path dataPath = getMaterializedViewDataPath(viewName);
+
+      boolean replaced = getFS().exists(viewPath);
+
+      // If replacing, first drop the old data
+      if (replaced) {
+        if (getFS().exists(dataPath)) {
+          getFS().delete(dataPath, true);
+        }
+      }
+
+      // Create the data directory for the materialized view
+      final FsPermission dirPerms = new FsPermission(
+          
schemaConfig.getOption(ExecConstants.NEW_VIEW_DEFAULT_PERMS_KEY).string_val);
+      getFS().mkdirs(dataPath, dirPerms);
+
+      // Set the data storage path in the materialized view
+      materializedView.setDataStoragePath(viewName);
+
+      // Write the materialized view definition file
+      final FsPermission viewPerms = new FsPermission(
+          
schemaConfig.getOption(ExecConstants.NEW_VIEW_DEFAULT_PERMS_KEY).string_val);
+      try (OutputStream stream = DrillFileSystem.create(getFS(), viewPath, 
viewPerms)) {
+        mapper.writeValue(stream, materializedView);
+      }
+
+      // Sync to metastore if enabled
+      syncMaterializedViewToMetastore(materializedView);
+
+      // Mark as complete (data will be populated by the handler via CTAS-like 
operation)
+      return replaced;
+    }
+
+    @Override
+    public void dropMaterializedView(String viewName) throws IOException {
+      Path viewPath = getMaterializedViewPath(viewName);
+      Path dataPath = getMaterializedViewDataPath(viewName);
+
+      // Delete the definition file
+      if (getFS().exists(viewPath)) {
+        getFS().delete(viewPath, false);
+      }
+
+      // Delete the data directory
+      if (getFS().exists(dataPath)) {
+        getFS().delete(dataPath, true);
+      }
+
+      // Remove from metastore if enabled
+      removeMaterializedViewFromMetastore(viewName);
+    }
+
+    @Override
+    public void refreshMaterializedView(String viewName) throws IOException {
+      // Read the existing materialized view definition
+      MaterializedView mv = getMaterializedView(viewName);
+      if (mv == null) {
+        throw UserException.validationError()
+            .message("Materialized view [%s] not found in schema [%s]", 
viewName, getFullSchemaName())
+            .build(logger);
+      }
+
+      Path dataPath = getMaterializedViewDataPath(viewName);
+
+      // Delete existing data
+      if (getFS().exists(dataPath)) {
+        getFS().delete(dataPath, true);
+      }
+
+      // Recreate the data directory
+      final FsPermission dirPerms = new FsPermission(
+          
schemaConfig.getOption(ExecConstants.NEW_VIEW_DEFAULT_PERMS_KEY).string_val);
+      getFS().mkdirs(dataPath, dirPerms);
+
+      // Mark as INCOMPLETE while data is being refreshed.
+      // completeMaterializedViewRefresh() should be called after data is 
fully written.
+      MaterializedView updatedMV = mv.withRefreshInfo(
+          mv.getLastRefreshTime(),
+          MaterializedView.RefreshStatus.INCOMPLETE);
+
+      // Write the updated definition file
+      Path viewPath = getMaterializedViewPath(viewName);
+      final FsPermission viewPerms = new FsPermission(
+          
schemaConfig.getOption(ExecConstants.NEW_VIEW_DEFAULT_PERMS_KEY).string_val);
+      try (OutputStream stream = DrillFileSystem.create(getFS(), viewPath, 
viewPerms)) {
+        mapper.writeValue(stream, updatedMV);
+      }
+
+      // Sync updated metadata to metastore if enabled
+      syncMaterializedViewToMetastore(updatedMV);
+    }
+
+    @Override
+    public void completeMaterializedViewRefresh(String viewName) throws 
IOException {
+      MaterializedView mv = getMaterializedView(viewName);
+      if (mv == null) {
+        throw UserException.validationError()
+            .message("Materialized view [%s] not found in schema [%s]", 
viewName, getFullSchemaName())
+            .build(logger);
+      }
+
+      // Mark as COMPLETE with current timestamp now that data is fully written
+      MaterializedView updatedMV = mv.withRefreshInfo(
+          System.currentTimeMillis(),
+          MaterializedView.RefreshStatus.COMPLETE);
+
+      Path viewPath = getMaterializedViewPath(viewName);
+      final FsPermission viewPerms = new FsPermission(
+          
schemaConfig.getOption(ExecConstants.NEW_VIEW_DEFAULT_PERMS_KEY).string_val);
+      try (OutputStream stream = DrillFileSystem.create(getFS(), viewPath, 
viewPerms)) {
+        mapper.writeValue(stream, updatedMV);
+      }
+
+      syncMaterializedViewToMetastore(updatedMV);
+    }
+
+    @Override
+    public CreateTableEntry createMaterializedViewDataWriter(String viewName) {
+      // Use Parquet format for storing materialized view data
+      FormatPlugin formatPlugin = plugin.getFormatPlugin("parquet");
+      if (formatPlugin == null) {
+        throw UserException.unsupportedError()
+            .message("Parquet format plugin not available for materialized 
view storage")
+            .build(logger);
+      }
+
+      // Store data in a directory with _mv_data suffix to avoid name collision
+      // with the materialized view lookup (which uses the same base name)
+      String dataLocation = config.getLocation() + Path.SEPARATOR + viewName + 
"_mv_data";
+      return new FileSystemCreateTableEntry(
+          (FileSystemConfig) plugin.getConfig(),
+          formatPlugin,
+          dataLocation,
+          Collections.emptyList(),  // No partition columns for MVs
+          StorageStrategy.DEFAULT);
+    }

Review Comment:
   I think, the method should have `MaterializedView` parameter and call 
`materializedView.getDataStoragePath()` to retrieve the `dataLocation`. Or at 
least this method shouldn't try to guess the data path and get the date path as 
argument.



##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/MaterializedViewRewriter.java:
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ */
+package org.apache.drill.exec.planner.logical;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.calcite.plan.RelOptMaterialization;
+import org.apache.calcite.plan.RelOptMaterializations;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelRoot;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.util.Pair;
+import org.apache.drill.exec.dotdrill.MaterializedView;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.planner.sql.conversion.SqlConverter;
+import org.apache.drill.exec.store.AbstractSchema;
+import org.apache.drill.exec.store.StoragePlugin;
+import org.apache.drill.exec.store.StoragePluginRegistry;
+import org.apache.drill.exec.store.StoragePluginRegistry.PluginException;
+import org.apache.drill.exec.store.dfs.FileSystemPlugin;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Utility class for materialized view query rewriting.
+ * <p>
+ * When enabled via planner.enable_materialized_view_rewrite, this class 
attempts
+ * to rewrite queries to use materialized views when beneficial.
+ * <p>
+ * Uses Calcite's {@link RelOptMaterializations#useMaterializedViews} API which
+ * normalizes both the query and MV definitions before performing structural
+ * matching via {@link org.apache.calcite.plan.SubstitutionVisitor}.
+ * <p>
+ * Materialized views are discovered by iterating over enabled file-based
+ * storage plugins (the only plugin type that supports MVs) and force-loading
+ * their schemas to find .materialized_view.drill files.
+ */
+public class MaterializedViewRewriter {
+  private static final Logger logger = 
LoggerFactory.getLogger(MaterializedViewRewriter.class);
+
+  private final QueryContext context;
+  private final SchemaPlus rootSchema;
+  private final SqlConverter sqlConverter;
+
+  public MaterializedViewRewriter(QueryContext context, SchemaPlus rootSchema, 
SqlConverter sqlConverter) {
+    this.context = context;
+    this.rootSchema = rootSchema;
+    this.sqlConverter = sqlConverter;
+  }
+
+  /**
+   * Attempts to rewrite the given RelNode to use a materialized view.
+   *
+   * @param queryRel the query plan to potentially rewrite
+   * @return the rewritten plan using an MV, or the original plan if no 
rewrite is possible
+   */
+  public RelNode rewrite(RelNode queryRel) {
+    if (!context.getPlannerSettings().isMaterializedViewRewriteEnabled()) {
+      return queryRel;
+    }
+
+    // Find all available materialized views that have been refreshed
+    List<MaterializedViewCandidate> candidates = 
findCandidateMaterializedViews();
+
+    if (candidates.isEmpty()) {
+      logger.debug("No refreshed materialized views available for rewriting");
+      return queryRel;
+    }
+
+    logger.debug("Found {} materialized view candidates for potential 
rewriting", candidates.size());
+
+    // Build Calcite RelOptMaterialization objects for each refreshed candidate
+    List<RelOptMaterialization> materializations = new ArrayList<>();
+    for (MaterializedViewCandidate candidate : candidates) {
+      if (!candidate.isRefreshed()) {
+        logger.debug("Skipping MV {} - not refreshed", candidate.getName());
+        continue;
+      }
+
+      try {
+        RelOptMaterialization mat = buildMaterialization(candidate);
+        if (mat != null) {
+          materializations.add(mat);
+        }
+      } catch (Exception e) {
+        logger.debug("Failed to build materialization for MV {}: {}", 
candidate.getName(), e.getMessage());
+      }
+    }
+
+    if (materializations.isEmpty()) {
+      logger.debug("No valid materializations could be built");
+      return queryRel;
+    }
+
+    // Use Calcite's materialized view matching API which normalizes both the
+    // query and MV definitions (trimming unused fields, converting 
Filter/Project
+    // to Calc, merging, etc.) before performing structural matching.
+    try {
+      List<Pair<RelNode, List<RelOptMaterialization>>> results =
+          RelOptMaterializations.useMaterializedViews(queryRel, 
materializations);
+
+      if (!results.isEmpty()) {
+        RelNode rewritten = results.get(0).left;
+        if (logger.isInfoEnabled()) {
+          List<RelOptMaterialization> usedMVs = results.get(0).right;
+          logger.info("Query rewritten to use materialized view(s): {}",
+              !usedMVs.isEmpty() ? usedMVs.get(0).qualifiedTableName : 
"unknown");
+        }
+        return rewritten;
+      }
+    } catch (Exception e) {
+      logger.debug("Materialized view rewriting failed: {}", e.getMessage());
+    }
+
+    logger.debug("No materialized view matched the query");
+    return queryRel;
+  }
+
+  /**
+   * Builds a Calcite {@link RelOptMaterialization} for a candidate MV.
+   */
+  private RelOptMaterialization buildMaterialization(MaterializedViewCandidate 
candidate) {
+    RelNode mvQueryRel = parseMvSql(candidate);
+    if (mvQueryRel == null) {
+      return null;
+    }
+
+    RelNode mvTableRel = buildMvScanRel(candidate);
+    if (mvTableRel == null) {
+      return null;
+    }
+
+    List<String> qualifiedTableName = java.util.Arrays.asList(
+        candidate.getSchemaPath().split("\\."));
+
+    return new RelOptMaterialization(
+        mvTableRel,
+        mvQueryRel,
+        null,
+        qualifiedTableName
+    );
+  }
+
+  /**
+   * Parses the MV's SQL definition into a RelNode.
+   */
+  private RelNode parseMvSql(MaterializedViewCandidate candidate) {
+    try {
+      String mvSql = candidate.getSql();
+      org.apache.calcite.sql.SqlNode parsedNode = sqlConverter.parse(mvSql);
+      org.apache.calcite.sql.SqlNode validatedNode = 
sqlConverter.validate(parsedNode);
+      RelRoot relRoot = sqlConverter.toRel(validatedNode);
+      return relRoot.rel;
+    } catch (Exception e) {
+      logger.debug("Failed to parse MV SQL for {}: {}", candidate.getName(), 
e.getMessage());
+      return null;
+    }
+  }
+
+  /**
+   * Builds a RelNode that scans the MV's pre-computed data.
+   */
+  private RelNode buildMvScanRel(MaterializedViewCandidate candidate) {
+    try {
+      String mvDataTable = candidate.getSchemaPath() + ".`" + 
candidate.getName() + "_mv_data`";
+
+      // Build explicit column list from the MV's field definitions to avoid
+      // DYNAMIC_STAR type issues with SELECT *
+      List<org.apache.drill.exec.dotdrill.View.Field> fields = 
candidate.getMaterializedView().getFields();
+      StringBuilder scanSql = new StringBuilder("SELECT ");
+      if (fields != null && !fields.isEmpty()) {
+        for (int i = 0; i < fields.size(); i++) {
+          if (i > 0) {
+            scanSql.append(", ");
+          }
+          scanSql.append("`").append(fields.get(i).getName()).append("`");
+        }
+      } else {
+        scanSql.append("*");
+      }
+      scanSql.append(" FROM ").append(mvDataTable);
+
+      org.apache.calcite.sql.SqlNode parsedNode = 
sqlConverter.parse(scanSql.toString());
+      org.apache.calcite.sql.SqlNode validatedNode = 
sqlConverter.validate(parsedNode);

Review Comment:
   Is the fully qualified name used intentionally?



##########
exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/WorkspaceSchemaFactory.java:
##########
@@ -351,6 +354,278 @@ public void dropView(String viewName) throws IOException {
       getFS().delete(getViewPath(viewName), false);
     }
 
+    private Path getMaterializedViewPath(String name) {
+      return DotDrillType.MATERIALIZED_VIEW.getPath(config.getLocation(), 
name);
+    }
+
+    private Path getMaterializedViewDataPath(String name) {
+      // Use _mv_data suffix to distinguish data directory from MV definition 
lookup
+      return new Path(config.getLocation(), name + "_mv_data");
+    }
+
+    @Override
+    public boolean createMaterializedView(MaterializedView materializedView) 
throws IOException {
+      String viewName = materializedView.getName();
+      Path viewPath = getMaterializedViewPath(viewName);
+      Path dataPath = getMaterializedViewDataPath(viewName);
+
+      boolean replaced = getFS().exists(viewPath);
+
+      // If replacing, first drop the old data
+      if (replaced) {
+        if (getFS().exists(dataPath)) {
+          getFS().delete(dataPath, true);
+        }
+      }
+
+      // Create the data directory for the materialized view
+      final FsPermission dirPerms = new FsPermission(
+          
schemaConfig.getOption(ExecConstants.NEW_VIEW_DEFAULT_PERMS_KEY).string_val);
+      getFS().mkdirs(dataPath, dirPerms);
+
+      // Set the data storage path in the materialized view
+      materializedView.setDataStoragePath(viewName);
+
+      // Write the materialized view definition file
+      final FsPermission viewPerms = new FsPermission(
+          
schemaConfig.getOption(ExecConstants.NEW_VIEW_DEFAULT_PERMS_KEY).string_val);
+      try (OutputStream stream = DrillFileSystem.create(getFS(), viewPath, 
viewPerms)) {
+        mapper.writeValue(stream, materializedView);
+      }
+
+      // Sync to metastore if enabled
+      syncMaterializedViewToMetastore(materializedView);
+
+      // Mark as complete (data will be populated by the handler via CTAS-like 
operation)
+      return replaced;
+    }
+
+    @Override
+    public void dropMaterializedView(String viewName) throws IOException {
+      Path viewPath = getMaterializedViewPath(viewName);
+      Path dataPath = getMaterializedViewDataPath(viewName);

Review Comment:
   Use `materializedView.getDataStoragePath()` instead.



##########
exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/WorkspaceSchemaFactory.java:
##########
@@ -351,6 +354,278 @@ public void dropView(String viewName) throws IOException {
       getFS().delete(getViewPath(viewName), false);
     }
 
+    private Path getMaterializedViewPath(String name) {
+      return DotDrillType.MATERIALIZED_VIEW.getPath(config.getLocation(), 
name);
+    }
+
+    private Path getMaterializedViewDataPath(String name) {
+      // Use _mv_data suffix to distinguish data directory from MV definition 
lookup
+      return new Path(config.getLocation(), name + "_mv_data");
+    }

Review Comment:
   Drop it. Use `materializedView.getDataStoragePath()` instead.



##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/MaterializedViewRewriter.java:
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ */
+package org.apache.drill.exec.planner.logical;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.calcite.plan.RelOptMaterialization;
+import org.apache.calcite.plan.RelOptMaterializations;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelRoot;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.util.Pair;
+import org.apache.drill.exec.dotdrill.MaterializedView;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.planner.sql.conversion.SqlConverter;
+import org.apache.drill.exec.store.AbstractSchema;
+import org.apache.drill.exec.store.StoragePlugin;
+import org.apache.drill.exec.store.StoragePluginRegistry;
+import org.apache.drill.exec.store.StoragePluginRegistry.PluginException;
+import org.apache.drill.exec.store.dfs.FileSystemPlugin;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Utility class for materialized view query rewriting.
+ * <p>
+ * When enabled via planner.enable_materialized_view_rewrite, this class 
attempts
+ * to rewrite queries to use materialized views when beneficial.
+ * <p>
+ * Uses Calcite's {@link RelOptMaterializations#useMaterializedViews} API which
+ * normalizes both the query and MV definitions before performing structural
+ * matching via {@link org.apache.calcite.plan.SubstitutionVisitor}.
+ * <p>
+ * Materialized views are discovered by iterating over enabled file-based
+ * storage plugins (the only plugin type that supports MVs) and force-loading
+ * their schemas to find .materialized_view.drill files.
+ */
+public class MaterializedViewRewriter {
+  private static final Logger logger = 
LoggerFactory.getLogger(MaterializedViewRewriter.class);
+
+  private final QueryContext context;
+  private final SchemaPlus rootSchema;
+  private final SqlConverter sqlConverter;
+
+  public MaterializedViewRewriter(QueryContext context, SchemaPlus rootSchema, 
SqlConverter sqlConverter) {
+    this.context = context;
+    this.rootSchema = rootSchema;
+    this.sqlConverter = sqlConverter;
+  }
+
+  /**
+   * Attempts to rewrite the given RelNode to use a materialized view.
+   *
+   * @param queryRel the query plan to potentially rewrite
+   * @return the rewritten plan using an MV, or the original plan if no 
rewrite is possible
+   */
+  public RelNode rewrite(RelNode queryRel) {
+    if (!context.getPlannerSettings().isMaterializedViewRewriteEnabled()) {
+      return queryRel;
+    }
+
+    // Find all available materialized views that have been refreshed
+    List<MaterializedViewCandidate> candidates = 
findCandidateMaterializedViews();
+
+    if (candidates.isEmpty()) {
+      logger.debug("No refreshed materialized views available for rewriting");
+      return queryRel;
+    }
+
+    logger.debug("Found {} materialized view candidates for potential 
rewriting", candidates.size());
+
+    // Build Calcite RelOptMaterialization objects for each refreshed candidate
+    List<RelOptMaterialization> materializations = new ArrayList<>();
+    for (MaterializedViewCandidate candidate : candidates) {
+      if (!candidate.isRefreshed()) {
+        logger.debug("Skipping MV {} - not refreshed", candidate.getName());
+        continue;
+      }
+
+      try {
+        RelOptMaterialization mat = buildMaterialization(candidate);
+        if (mat != null) {
+          materializations.add(mat);
+        }
+      } catch (Exception e) {
+        logger.debug("Failed to build materialization for MV {}: {}", 
candidate.getName(), e.getMessage());
+      }
+    }
+
+    if (materializations.isEmpty()) {
+      logger.debug("No valid materializations could be built");
+      return queryRel;
+    }
+
+    // Use Calcite's materialized view matching API which normalizes both the
+    // query and MV definitions (trimming unused fields, converting 
Filter/Project
+    // to Calc, merging, etc.) before performing structural matching.
+    try {
+      List<Pair<RelNode, List<RelOptMaterialization>>> results =
+          RelOptMaterializations.useMaterializedViews(queryRel, 
materializations);
+
+      if (!results.isEmpty()) {
+        RelNode rewritten = results.get(0).left;
+        if (logger.isInfoEnabled()) {
+          List<RelOptMaterialization> usedMVs = results.get(0).right;
+          logger.info("Query rewritten to use materialized view(s): {}",
+              !usedMVs.isEmpty() ? usedMVs.get(0).qualifiedTableName : 
"unknown");
+        }
+        return rewritten;
+      }
+    } catch (Exception e) {
+      logger.debug("Materialized view rewriting failed: {}", e.getMessage());
+    }
+
+    logger.debug("No materialized view matched the query");
+    return queryRel;
+  }
+
+  /**
+   * Builds a Calcite {@link RelOptMaterialization} for a candidate MV.
+   */
+  private RelOptMaterialization buildMaterialization(MaterializedViewCandidate 
candidate) {
+    RelNode mvQueryRel = parseMvSql(candidate);
+    if (mvQueryRel == null) {
+      return null;
+    }
+
+    RelNode mvTableRel = buildMvScanRel(candidate);
+    if (mvTableRel == null) {
+      return null;
+    }
+
+    List<String> qualifiedTableName = java.util.Arrays.asList(
+        candidate.getSchemaPath().split("\\."));
+
+    return new RelOptMaterialization(
+        mvTableRel,
+        mvQueryRel,
+        null,
+        qualifiedTableName
+    );
+  }
+
+  /**
+   * Parses the MV's SQL definition into a RelNode.
+   */
+  private RelNode parseMvSql(MaterializedViewCandidate candidate) {
+    try {
+      String mvSql = candidate.getSql();
+      org.apache.calcite.sql.SqlNode parsedNode = sqlConverter.parse(mvSql);
+      org.apache.calcite.sql.SqlNode validatedNode = 
sqlConverter.validate(parsedNode);
+      RelRoot relRoot = sqlConverter.toRel(validatedNode);
+      return relRoot.rel;
+    } catch (Exception e) {
+      logger.debug("Failed to parse MV SQL for {}: {}", candidate.getName(), 
e.getMessage());
+      return null;
+    }
+  }
+
+  /**
+   * Builds a RelNode that scans the MV's pre-computed data.
+   */
+  private RelNode buildMvScanRel(MaterializedViewCandidate candidate) {
+    try {
+      String mvDataTable = candidate.getSchemaPath() + ".`" + 
candidate.getName() + "_mv_data`";
+
+      // Build explicit column list from the MV's field definitions to avoid
+      // DYNAMIC_STAR type issues with SELECT *
+      List<org.apache.drill.exec.dotdrill.View.Field> fields = 
candidate.getMaterializedView().getFields();
+      StringBuilder scanSql = new StringBuilder("SELECT ");
+      if (fields != null && !fields.isEmpty()) {
+        for (int i = 0; i < fields.size(); i++) {
+          if (i > 0) {
+            scanSql.append(", ");
+          }
+          scanSql.append("`").append(fields.get(i).getName()).append("`");
+        }
+      } else {
+        scanSql.append("*");
+      }
+      scanSql.append(" FROM ").append(mvDataTable);
+
+      org.apache.calcite.sql.SqlNode parsedNode = 
sqlConverter.parse(scanSql.toString());
+      org.apache.calcite.sql.SqlNode validatedNode = 
sqlConverter.validate(parsedNode);
+      RelRoot relRoot = sqlConverter.toRel(validatedNode);
+      return relRoot.rel;
+    } catch (Exception e) {
+      logger.debug("Failed to build MV scan for {}: {}", candidate.getName(), 
e.getMessage());
+      return null;
+    }
+  }
+
+  /**
+   * Finds all materialized views by iterating over enabled file-based storage
+   * plugins. Only file-based plugins (FileSystemPlugin) support materialized 
views.
+   * <p>
+   * Because Drill's schema tree is lazily loaded, we cannot simply traverse
+   * already-loaded schemas. Instead, we use the StoragePluginRegistry to
+   * discover file-based plugins and force-load their workspace schemas.
+   */
+  private List<MaterializedViewCandidate> findCandidateMaterializedViews() {
+    List<MaterializedViewCandidate> candidates = new ArrayList<>();
+    StoragePluginRegistry pluginRegistry = context.getStorage();
+
+    // Get all enabled storage plugin names
+    Map<String, org.apache.drill.common.logical.StoragePluginConfig> 
enabledPlugins =
+        
pluginRegistry.storedConfigs(StoragePluginRegistry.PluginFilter.ENABLED);
+
+    for (String pluginName : enabledPlugins.keySet()) {
+      try {
+        StoragePlugin plugin = pluginRegistry.getPlugin(pluginName);
+        if (!(plugin instanceof FileSystemPlugin)) {
+          continue;
+        }
+
+        // Force-load this plugin's schema into the root schema tree.
+        // DynamicRootSchema lazily loads schemas only on getSubSchema() calls.
+        SchemaPlus pluginSchema = rootSchema.getSubSchema(pluginName);
+        if (pluginSchema == null) {
+          continue;
+        }
+
+        // Iterate over workspaces (sub-schemas of the plugin)
+        for (String workspaceName : pluginSchema.getSubSchemaNames()) {
+          SchemaPlus workspaceSchema = 
pluginSchema.getSubSchema(workspaceName);
+          if (workspaceSchema == null) {
+            continue;
+          }
+
+          AbstractSchema abstractSchema = 
workspaceSchema.unwrap(AbstractSchema.class);
+          if (abstractSchema == null) {
+            continue;
+          }
+
+          collectMaterializedViewsFromSchema(abstractSchema, candidates);
+        }
+      } catch (PluginException e) {
+        logger.debug("Error accessing plugin {}: {}", pluginName, 
e.getMessage());
+      }
+    }
+
+    return candidates;
+  }
+
+  /**
+   * Collects MVs from a specific workspace schema.
+   */
+  private void collectMaterializedViewsFromSchema(AbstractSchema schema,
+                                                   
List<MaterializedViewCandidate> candidates) {
+    Set<String> tableNames = schema.getTableNames();
+    for (String tableName : tableNames) {
+      try {
+        MaterializedView mv = schema.getMaterializedView(tableName);
+        if (mv != null) {
+          boolean isRefreshed = mv.getRefreshStatus() == 
MaterializedView.RefreshStatus.COMPLETE;
+          candidates.add(new MaterializedViewCandidate(
+              mv.getName(),
+              schema.getFullSchemaName(),
+              mv,
+              isRefreshed));
+        }
+      } catch (IOException e) {
+        logger.debug("Error reading MV {}: {}", tableName, e.getMessage());
+      }
+    }
+  }
+
+  /**
+   * Represents a candidate materialized view for query rewriting.
+   */
+  public static class MaterializedViewCandidate {
+    private final String name;
+    private final String schemaPath;
+    private final MaterializedView materializedView;
+    private final boolean refreshed;

Review Comment:
   Did you intentionally leave `MaterializedView.RefreshStatus` as ENUM with 2 
possible values instead of a boolean? Because, logically, it can be easily 
replaced with `boolean refreshed`, as you did there.



##########
exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/WorkspaceSchemaFactory.java:
##########
@@ -351,6 +354,278 @@ public void dropView(String viewName) throws IOException {
       getFS().delete(getViewPath(viewName), false);
     }
 
+    private Path getMaterializedViewPath(String name) {
+      return DotDrillType.MATERIALIZED_VIEW.getPath(config.getLocation(), 
name);
+    }
+
+    private Path getMaterializedViewDataPath(String name) {
+      // Use _mv_data suffix to distinguish data directory from MV definition 
lookup
+      return new Path(config.getLocation(), name + "_mv_data");
+    }
+
+    @Override
+    public boolean createMaterializedView(MaterializedView materializedView) 
throws IOException {
+      String viewName = materializedView.getName();
+      Path viewPath = getMaterializedViewPath(viewName);
+      Path dataPath = getMaterializedViewDataPath(viewName);
+
+      boolean replaced = getFS().exists(viewPath);
+
+      // If replacing, first drop the old data
+      if (replaced) {
+        if (getFS().exists(dataPath)) {
+          getFS().delete(dataPath, true);
+        }
+      }
+
+      // Create the data directory for the materialized view
+      final FsPermission dirPerms = new FsPermission(
+          
schemaConfig.getOption(ExecConstants.NEW_VIEW_DEFAULT_PERMS_KEY).string_val);
+      getFS().mkdirs(dataPath, dirPerms);
+
+      // Set the data storage path in the materialized view
+      materializedView.setDataStoragePath(viewName);
+
+      // Write the materialized view definition file
+      final FsPermission viewPerms = new FsPermission(
+          
schemaConfig.getOption(ExecConstants.NEW_VIEW_DEFAULT_PERMS_KEY).string_val);
+      try (OutputStream stream = DrillFileSystem.create(getFS(), viewPath, 
viewPerms)) {
+        mapper.writeValue(stream, materializedView);
+      }
+
+      // Sync to metastore if enabled
+      syncMaterializedViewToMetastore(materializedView);
+
+      // Mark as complete (data will be populated by the handler via CTAS-like 
operation)
+      return replaced;
+    }
+
+    @Override
+    public void dropMaterializedView(String viewName) throws IOException {
+      Path viewPath = getMaterializedViewPath(viewName);
+      Path dataPath = getMaterializedViewDataPath(viewName);
+
+      // Delete the definition file
+      if (getFS().exists(viewPath)) {
+        getFS().delete(viewPath, false);
+      }
+
+      // Delete the data directory
+      if (getFS().exists(dataPath)) {
+        getFS().delete(dataPath, true);
+      }
+
+      // Remove from metastore if enabled
+      removeMaterializedViewFromMetastore(viewName);
+    }
+
+    @Override
+    public void refreshMaterializedView(String viewName) throws IOException {
+      // Read the existing materialized view definition
+      MaterializedView mv = getMaterializedView(viewName);
+      if (mv == null) {
+        throw UserException.validationError()
+            .message("Materialized view [%s] not found in schema [%s]", 
viewName, getFullSchemaName())
+            .build(logger);
+      }
+
+      Path dataPath = getMaterializedViewDataPath(viewName);
+
+      // Delete existing data
+      if (getFS().exists(dataPath)) {
+        getFS().delete(dataPath, true);
+      }
+
+      // Recreate the data directory
+      final FsPermission dirPerms = new FsPermission(
+          
schemaConfig.getOption(ExecConstants.NEW_VIEW_DEFAULT_PERMS_KEY).string_val);
+      getFS().mkdirs(dataPath, dirPerms);
+
+      // Mark as INCOMPLETE while data is being refreshed.
+      // completeMaterializedViewRefresh() should be called after data is 
fully written.
+      MaterializedView updatedMV = mv.withRefreshInfo(
+          mv.getLastRefreshTime(),
+          MaterializedView.RefreshStatus.INCOMPLETE);
+
+      // Write the updated definition file
+      Path viewPath = getMaterializedViewPath(viewName);
+      final FsPermission viewPerms = new FsPermission(
+          
schemaConfig.getOption(ExecConstants.NEW_VIEW_DEFAULT_PERMS_KEY).string_val);
+      try (OutputStream stream = DrillFileSystem.create(getFS(), viewPath, 
viewPerms)) {
+        mapper.writeValue(stream, updatedMV);
+      }
+
+      // Sync updated metadata to metastore if enabled
+      syncMaterializedViewToMetastore(updatedMV);
+    }
+
+    @Override
+    public void completeMaterializedViewRefresh(String viewName) throws 
IOException {
+      MaterializedView mv = getMaterializedView(viewName);
+      if (mv == null) {
+        throw UserException.validationError()
+            .message("Materialized view [%s] not found in schema [%s]", 
viewName, getFullSchemaName())
+            .build(logger);
+      }
+
+      // Mark as COMPLETE with current timestamp now that data is fully written
+      MaterializedView updatedMV = mv.withRefreshInfo(
+          System.currentTimeMillis(),
+          MaterializedView.RefreshStatus.COMPLETE);
+
+      Path viewPath = getMaterializedViewPath(viewName);
+      final FsPermission viewPerms = new FsPermission(
+          
schemaConfig.getOption(ExecConstants.NEW_VIEW_DEFAULT_PERMS_KEY).string_val);
+      try (OutputStream stream = DrillFileSystem.create(getFS(), viewPath, 
viewPerms)) {
+        mapper.writeValue(stream, updatedMV);
+      }
+
+      syncMaterializedViewToMetastore(updatedMV);
+    }
+
+    @Override
+    public CreateTableEntry createMaterializedViewDataWriter(String viewName) {
+      // Use Parquet format for storing materialized view data
+      FormatPlugin formatPlugin = plugin.getFormatPlugin("parquet");
+      if (formatPlugin == null) {
+        throw UserException.unsupportedError()
+            .message("Parquet format plugin not available for materialized 
view storage")
+            .build(logger);
+      }
+
+      // Store data in a directory with _mv_data suffix to avoid name collision
+      // with the materialized view lookup (which uses the same base name)
+      String dataLocation = config.getLocation() + Path.SEPARATOR + viewName + 
"_mv_data";
+      return new FileSystemCreateTableEntry(
+          (FileSystemConfig) plugin.getConfig(),
+          formatPlugin,
+          dataLocation,
+          Collections.emptyList(),  // No partition columns for MVs
+          StorageStrategy.DEFAULT);
+    }
+
+    @Override
+    public MaterializedView getMaterializedView(String viewName) throws 
IOException {
+      List<DotDrillFile> files = Collections.emptyList();
+      try {
+        files = DotDrillUtil.getDotDrills(getFS(), new 
Path(config.getLocation()),
+            DrillStringUtils.removeLeadingSlash(viewName), 
DotDrillType.MATERIALIZED_VIEW);
+      } catch (UnsupportedOperationException e) {
+        logger.debug("The filesystem for this workspace does not support this 
operation.", e);
+        return null;
+      } catch (IOException e) {
+        logger.warn("Failure while trying to list materialized view in 
workspace [{}]", getFullSchemaName(), e);
+        return null;
+      }
+
+      for (DotDrillFile f : files) {
+        if (f.getType() == DotDrillType.MATERIALIZED_VIEW) {
+          return f.getMaterializedView(mapper);
+        }
+      }

Review Comment:
   Don't you want to replace it with a more declarative syntax to get first 
`MATERIALIZED_VIEW`:
   ```java
   return files.stream()
       .filter(f -> f.getType() == DotDrillType.MATERIALIZED_VIEW)
       .findFirst()
       .map(f -> f.getMaterializedView(mapper))
       .orElse(null);
   ```





> Add Support for Materialized Views
> ----------------------------------
>
>                 Key: DRILL-8543
>                 URL: https://issues.apache.org/jira/browse/DRILL-8543
>             Project: Apache Drill
>          Issue Type: New Feature
>          Components: Metadata, Query Planning &amp; Optimization
>    Affects Versions: 1.22.0
>            Reporter: Charles Givre
>            Assignee: Charles Givre
>            Priority: Major
>             Fix For: 1.23.0
>
>




--
This message was sent by Atlassian Jira
(v8.20.10#820010)