[GitHub] [carbondata] niuge01 opened a new pull request #3661: [WIP] Support materialized view

[GitBox]

CarbonDataQA1 commented on issue #3661: [WIP] Support materialized view
URL: https://github.com/apache/carbondata/pull/3661#issuecomment-602496537
 
 
   Build Failed  with Spark 2.3.4, Please check CI http://121.244.95.60:12545/job/ApacheCarbonPRBuilder2.3/2542/
   

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

Indhumathi27 commented on a change in pull request #3661: [WIP] Support materialized view
URL: https://github.com/apache/carbondata/pull/3661#discussion_r396306214
 
 

 ##########
 File path: core/src/main/java/org/apache/carbondata/core/datamap/status/MVSegmentStatusUtil.java
 ##########
 @@ -32,7 +32,7 @@
 /**
  * Utility class to get updated segment mapping for datamap table
  */
-public class DataMapSegmentStatusUtil {
+public class MVSegmentStatusUtil {
 
 Review comment:
   Can move `getSegmentMap` method from this class to `MaterializedViewManager` and delete this class

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

Indhumathi27 commented on a change in pull request #3661: [WIP] Support materialized view
URL: https://github.com/apache/carbondata/pull/3661#discussion_r396301531
 
 

 ##########
 File path: core/src/main/java/org/apache/carbondata/core/datamap/IndexUtil.java
 ##########
 @@ -185,38 +185,37 @@ public static void pruneSegments(List<Segment> segments, List<ExtendedBlocklet>
    Loads the datamaps in parallel by utilizing executor
    *
    @param carbonTable
-   @param dataMapExprWrapper
+   @param indexExprWrapper
    @param validSegments
    @param partitionsToPrune
    @throws IOException
    */
-  public static void loadDataMaps(CarbonTable carbonTable, DataMapExprWrapper dataMapExprWrapper,
+  public static void loadDataMaps(CarbonTable carbonTable, IndexExprWrapper indexExprWrapper,
       List<Segment> validSegments, List<PartitionSpec> partitionsToPrune) throws IOException {
     if (!CarbonProperties.getInstance()
         .isDistributedPruningEnabled(carbonTable.getDatabaseName(), carbonTable.getTableName())
-        && BlockletDataMapUtil.loadDataMapsParallel(carbonTable)) {
-      String clsName = "org.apache.spark.sql.secondaryindex.Jobs.SparkBlockletDataMapLoaderJob";
-      DataMapJob dataMapJob = (DataMapJob) createDataMapJob(clsName);
-      String className =
-          "org.apache.spark.sql.secondaryindex.Jobs.DistributableBlockletDataMapLoader";
+        && BlockletIndexUtil.loadDataMapsParallel(carbonTable)) {
+      String clsName = "org.apache.carbondata.spark.rdd.SparkBlockletDataMapLoaderJob";
 
 Review comment:
   Please change the className and classPath for `SparkBlockletDataMapLoaderJob` and `DistributableBlockletDataMapLoader`

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

Indhumathi27 commented on a change in pull request #3661: [WIP] Support materialized view
URL: https://github.com/apache/carbondata/pull/3661#discussion_r396303230
 
 

 ##########
 File path: core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapSyncStatus.java
 ##########
 @@ -36,7 +36,7 @@
 /**
  * Interface to check whether datamap can be enabled
  */
[e-mail]("DataMap")
[e-mail]("Index")
 
 Review comment:
   Rename className from `DataMapSyncStatus` to `MVSyncStatus`

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

Indhumathi27 commented on a change in pull request #3661: [WIP] Support materialized view
URL: https://github.com/apache/carbondata/pull/3661#discussion_r396333133
 
 

 ##########
 File path: core/src/main/java/org/apache/carbondata/core/view/MaterializedViewProvider.java
 ##########
 @@ -0,0 +1,573 @@
+/*
+ * 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.carbondata.core.view;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStreamWriter;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.carbondata.common.annotations.InterfaceAudience;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.constants.CarbonCommonConstants;
+import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
+import org.apache.carbondata.core.datastore.filesystem.CarbonFileFilter;
+import org.apache.carbondata.core.datastore.impl.FileFactory;
+import org.apache.carbondata.core.fileoperations.AtomicFileOperationFactory;
+import org.apache.carbondata.core.fileoperations.AtomicFileOperations;
+import org.apache.carbondata.core.fileoperations.FileWriteOperation;
+import org.apache.carbondata.core.locks.CarbonLockFactory;
+import org.apache.carbondata.core.locks.CarbonLockUtil;
+import org.apache.carbondata.core.locks.ICarbonLock;
+import org.apache.carbondata.core.locks.LockUsage;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.metadata.schema.table.RelationIdentifier;
+import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
+import org.apache.carbondata.core.statusmanager.SegmentStatus;
+import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
+import org.apache.carbondata.core.util.CarbonProperties;
+import org.apache.carbondata.core.util.CarbonUtil;
+import org.apache.carbondata.core.util.path.CarbonTablePath;
+
+import com.google.gson.Gson;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.log4j.Logger;
+
+/**
+ * Stores mv schema in disk as json format
+ */
[e-mail]
+public class MaterializedViewProvider {
+
+  private static final Logger LOG = LogServiceFactory.getLogService(
+      MaterializedViewProvider.class.getCanonicalName());
+
+  private static final String STATUS_FILE_NAME = "mv_status";
+
+  private final String storeLocation;
+
+  private final Map<String, SchemaProvider> schemaProviders = new ConcurrentHashMap<>();
+
+  private MaterializedViewProvider(String storeLocation) {
+    this.storeLocation = storeLocation;
+  }
+
+  public static MaterializedViewProvider get() {
+    String storeLocation =
+        CarbonProperties.getInstance().getProperty(CarbonCommonConstants.STORE_LOCATION);
+    if (storeLocation == null) {
+      throw new RuntimeException(
+          "Property [" + CarbonCommonConstants.STORE_LOCATION + "] is not set.");
+    }
+    return new MaterializedViewProvider(storeLocation);
+  }
+
+  private static String getSchemaPath(String schemaRoot, String viewName) {
+    return schemaRoot + CarbonCommonConstants.FILE_SEPARATOR + "mv_schema." + viewName;
+  }
+
+  private SchemaProvider getSchemaProvider(String databaseName) {
+    String databaseNameUpper = databaseName.toUpperCase();
+    SchemaProvider schemaProvider = this.schemaProviders.get(databaseNameUpper);
+    if (schemaProvider == null) {
+      synchronized (this.schemaProviders) {
+        schemaProvider = this.schemaProviders.get(databaseNameUpper);
+        if (schemaProvider == null) {
+          String databaseLocation;
+          if (databaseNameUpper.equalsIgnoreCase("default")) {
+            databaseLocation = CarbonUtil.checkAndAppendHDFSUrl(this.storeLocation);
+          } else {
+            databaseLocation = CarbonUtil.checkAndAppendHDFSUrl(this.storeLocation +
+                CarbonCommonConstants.FILE_SEPARATOR + databaseName + ".db");
+          }
+          if (!FileFactory.getCarbonFile(databaseLocation).exists()) {
+            return null;
+          }
+          schemaProvider = new SchemaProvider(databaseLocation);
+          this.schemaProviders.put(databaseNameUpper, schemaProvider);
+        }
+      }
+    }
+    return schemaProvider;
+  }
+
+  public MaterializedViewSchema getSchema(MaterializedViewManager viewManager,
+      String databaseName, String viewName) throws IOException {
+    SchemaProvider schemaProvider = this.getSchemaProvider(databaseName);
+    if (schemaProvider == null) {
+      return null;
+    }
+    return schemaProvider.retrieveSchema(viewManager, viewName);
+  }
+
+  List<MaterializedViewSchema> getSchemas(MaterializedViewManager viewManager,
+      String databaseName, CarbonTable carbonTable) throws IOException {
+    SchemaProvider schemaProvider = this.getSchemaProvider(databaseName);
+    if (schemaProvider == null) {
+      return Collections.emptyList();
+    } else {
+      return schemaProvider.retrieveSchemas(viewManager, carbonTable);
+    }
+  }
+
+  List<MaterializedViewSchema> getSchemas(MaterializedViewManager viewManager,
+      String databaseName) throws IOException {
+    SchemaProvider schemaProvider = this.getSchemaProvider(databaseName);
+    if (schemaProvider == null) {
+      return Collections.emptyList();
+    } else {
+      return schemaProvider.retrieveAllSchemas(viewManager);
+    }
+  }
+
+  void saveSchema(MaterializedViewManager viewManager, String databaseName,
+      MaterializedViewSchema viewSchema) throws IOException {
+    SchemaProvider schemaProvider = this.getSchemaProvider(databaseName);
+    if (schemaProvider == null) {
+      throw new IOException("Database [" + databaseName + "] is not found.");
+    }
+    schemaProvider.saveSchema(viewManager, viewSchema);
+  }
+
+  public void dropSchema(String databaseName, String viewName)
+      throws IOException {
+    SchemaProvider schemaProvider = this.getSchemaProvider(databaseName);
+    if (schemaProvider == null) {
+      throw new IOException("Materialized view with name " + databaseName + "." + viewName +
+          " does not exists in storage");
+    }
+    schemaProvider.dropSchema(viewName);
+  }
+
+  private String getStatusFileName(String databaseName) {
+    if (databaseName.equalsIgnoreCase("default")) {
+      return this.storeLocation +
+          CarbonCommonConstants.FILE_SEPARATOR + "_system" +
+          CarbonCommonConstants.FILE_SEPARATOR + STATUS_FILE_NAME;
+    } else {
+      return this.storeLocation +
+          CarbonCommonConstants.FILE_SEPARATOR + databaseName + ".db" +
+          CarbonCommonConstants.FILE_SEPARATOR + "_system" +
+          CarbonCommonConstants.FILE_SEPARATOR + STATUS_FILE_NAME;
+    }
+  }
+
+  public List<MaterializedViewStatusDetail> getStatusDetails(String databaseName)
+      throws IOException {
+    String statusPath = this.getStatusFileName(databaseName);
+    Gson gsonObjectToRead = new Gson();
+    DataInputStream dataInputStream = null;
+    BufferedReader buffReader = null;
+    InputStreamReader inStream = null;
+    MaterializedViewStatusDetail[] statusDetails;
+    try {
+      if (!FileFactory.isFileExist(statusPath)) {
+        return Collections.emptyList();
+      }
+      dataInputStream = FileFactory.getDataInputStream(statusPath);
+      inStream = new InputStreamReader(dataInputStream,
+          Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET));
+      buffReader = new BufferedReader(inStream);
+      statusDetails = gsonObjectToRead.fromJson(buffReader,
+          MaterializedViewStatusDetail[].class);
+    } catch (IOException e) {
+      LOG.error("Failed to read datamap status", e);
+      throw e;
+    } finally {
+      CarbonUtil.closeStreams(buffReader, inStream, dataInputStream);
+    }
+    // if status details is null, return empty array
+    if (null == statusDetails) {
+      return Collections.emptyList();
+    }
+    return Arrays.asList(statusDetails);
+  }
+
+  private static ICarbonLock getStatusLock(String databaseName) {
+    return CarbonLockFactory
+        .getSystemLevelCarbonLockObj(
+            CarbonProperties.getInstance().getSystemFolderLocation(databaseName),
+            LockUsage.MATERIALIZED_VIEW_STATUS_LOCK);
+  }
+
+  /**
+   * Update or add the status of passed mvs with the given mv status. If the mv status
+   * given is enabled/disabled then updates/adds the mv, in case of drop it just removes it
+   * from the file.
+   * This method always overwrites the old file.
+   * @param schemaList schemas of which are need to be updated in mv status
+   * @param status  status to be updated for the mv schemas
+   */
+  public void updateStatus(List<MaterializedViewSchema> schemaList,
+      MaterializedViewStatus status) throws IOException {
+    if (schemaList == null || schemaList.size() == 0) {
+      // There is nothing to update
+      return;
+    }
+    final Map<String, List<MaterializedViewSchema>> schemasMapByDatabase = new HashMap<>();
+    for (MaterializedViewSchema schema : schemaList) {
+      String databaseName = schema.getIdentifier().getDatabaseName().toLowerCase();
+      List<MaterializedViewSchema> schemas = schemasMapByDatabase.get(databaseName);
+      if (schemas == null) {
+        schemas = new ArrayList<>();
+        schemasMapByDatabase.put(databaseName, schemas);
+      }
+      schemas.add(schema);
+    }
+    for (Map.Entry<String, List<MaterializedViewSchema>> entry : schemasMapByDatabase.entrySet()) {
+      this.updateStatus(entry.getKey(), entry.getValue(), status);
+    }
+  }
+
+  private void updateStatus(String databaseName, List<MaterializedViewSchema> schemaList,
+      MaterializedViewStatus status) throws IOException {
+    ICarbonLock carbonTableStatusLock = getStatusLock(databaseName);
+    boolean locked = false;
+    try {
+      locked = carbonTableStatusLock.lockWithRetries();
+      if (locked) {
+        LOG.info("Materialized view status lock has been successfully acquired.");
+        if (status == MaterializedViewStatus.ENABLED) {
+          // Enable mv only if mv tables and main table are in sync
+          if (!isViewCanBeEnabled(schemaList.get(0))) {
+            return;
+          }
+        }
+        List<MaterializedViewStatusDetail> statusDetailList =
+            new ArrayList<>(getStatusDetails(databaseName));
+        List<MaterializedViewStatusDetail> changedStatusDetails = new ArrayList<>();
+        List<MaterializedViewStatusDetail> newStatusDetails = new ArrayList<>();
+        for (MaterializedViewSchema schema : schemaList) {
+          boolean exists = false;
+          for (MaterializedViewStatusDetail statusDetail : statusDetailList) {
+            if (statusDetail.getIdentifier().equals(schema.getIdentifier())) {
+              statusDetail.setStatus(status);
+              changedStatusDetails.add(statusDetail);
+              exists = true;
+            }
+          }
+          if (!exists) {
+            newStatusDetails
+                .add(new MaterializedViewStatusDetail(schema.getIdentifier(),
+                    status));
+          }
+        }
+        // Add the newly added datamaps to the list.
+        if (newStatusDetails.size() > 0 &&
+            status != MaterializedViewStatus.DROPPED) {
+          statusDetailList.addAll(newStatusDetails);
+        }
+        // In case of dropped datamap, just remove from the list.
+        if (status == MaterializedViewStatus.DROPPED) {
+          statusDetailList.removeAll(changedStatusDetails);
+        }
+        writeLoadDetailsIntoFile(
+            this.getStatusFileName(databaseName),
+            statusDetailList.toArray(
+                new MaterializedViewStatusDetail[statusDetailList.size()]));
+      } else {
+        String errorMsg = "Upadating datamapstatus is failed due to another process taken the lock"
+            + " for updating it";
+        LOG.error(errorMsg);
+        throw new IOException(errorMsg + " Please try after some time.");
+      }
+    } finally {
+      if (locked) {
+        CarbonLockUtil.fileUnlock(carbonTableStatusLock, LockUsage.DATAMAP_STATUS_LOCK);
+      }
+    }
+  }
+
+  /**
+   * writes mv status details
+   */
+  private static void writeLoadDetailsIntoFile(String location,
+      MaterializedViewStatusDetail[] statusDetails) throws IOException {
+    FileFactory.touchFile(FileFactory.getCarbonFile(location),
+        new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
+    AtomicFileOperations fileWrite = AtomicFileOperationFactory.getAtomicFileOperations(location);
+    BufferedWriter brWriter = null;
+    DataOutputStream dataOutputStream = null;
+    Gson gsonObjectToWrite = new Gson();
+    // write the updated data into the mv status file.
+    try {
+      dataOutputStream = fileWrite.openForWrite(FileWriteOperation.OVERWRITE);
+      brWriter = new BufferedWriter(new OutputStreamWriter(dataOutputStream,
+          Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)));
+
+      String metadataInstance = gsonObjectToWrite.toJson(statusDetails);
+      brWriter.write(metadataInstance);
+    } catch (IOException ioe) {
+      LOG.error("Error message: " + ioe.getLocalizedMessage());
+      fileWrite.setFailed();
+      throw ioe;
+    } finally {
+      if (null != brWriter) {
+        brWriter.flush();
+      }
+      CarbonUtil.closeStreams(brWriter);
+      fileWrite.close();
+    }
+
+  }
+
+  /**
+   * This method checks if main table and mv table are synchronised or not. If synchronised
+   * return true to enable the mv
+   *
+   * @param schema of mv to be disabled or enabled
+   * @return flag to enable or disable mv
+   * @throws IOException
+   */
+  private static boolean isViewCanBeEnabled(MaterializedViewSchema schema)
+      throws IOException {
+    if (!schema.isRefreshIncremental()) {
+      return true;
+    }
+    boolean isViewCanBeEnabled = true;
+    String viewMetadataPath =
+        CarbonTablePath.getMetadataPath(schema.getIdentifier().getTablePath());
+    LoadMetadataDetails[] viewLoadMetadataDetails =
+        SegmentStatusManager.readLoadMetadata(viewMetadataPath);
+    Map<String, List<String>> viewSegmentMap = new HashMap<>();
+    for (LoadMetadataDetails loadMetadataDetail : viewLoadMetadataDetails) {
+      if (loadMetadataDetail.getSegmentStatus() == SegmentStatus.SUCCESS) {
+        Map<String, List<String>> segmentMap =
+            new Gson().fromJson(loadMetadataDetail.getExtraInfo(), Map.class);
+        if (viewSegmentMap.isEmpty()) {
+          viewSegmentMap.putAll(segmentMap);
+        } else {
+          for (Map.Entry<String, List<String>> entry : segmentMap.entrySet()) {
+            if (null != viewSegmentMap.get(entry.getKey())) {
+              viewSegmentMap.get(entry.getKey()).addAll(entry.getValue());
+            }
+          }
+        }
+      }
+    }
+    List<RelationIdentifier> associatedTables = schema.getAssociatedTables();
+    for (RelationIdentifier associatedTable : associatedTables) {
+      List<String> associatedTableSegmentList =
+          SegmentStatusManager.getValidSegmentList(associatedTable);
+      if (!associatedTableSegmentList.isEmpty()) {
+        if (viewSegmentMap.isEmpty()) {
+          isViewCanBeEnabled = false;
+        } else {
+          isViewCanBeEnabled = viewSegmentMap.get(
+              associatedTable.getDatabaseName() + CarbonCommonConstants.POINT +
+                  associatedTable.getTableName()).containsAll(associatedTableSegmentList);
+        }
+      }
+    }
+    return isViewCanBeEnabled;
+  }
+
+  /**
+   * Data map schema provider of a database.
+   */
+  private static final class SchemaProvider {
+
+    private String systemDirectory;
+
+    private String schemaIndexFilePath;
 
 Review comment:
   Can change the variable name, as it indicates schema modified file path and can variable description

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

Indhumathi27 commented on a change in pull request #3661: [WIP] Support materialized view
URL: https://github.com/apache/carbondata/pull/3661#discussion_r396313702
 
 

 ##########
 File path: core/src/main/java/org/apache/carbondata/core/metadata/schema/datamap/DataMapClassProvider.java
 ##########
 @@ -18,26 +18,20 @@
 package org.apache.carbondata.core.metadata.schema.datamap;
 
 /**
- * type for create datamap
- * The syntax of datamap creation is as follows.
- * CREATE DATAMAP IF NOT EXISTS dataMapName ON TABLE tableName USING 'DataMapClassProvider'
- * DMPROPERTIES('KEY'='VALUE') AS SELECT COUNT(COL1) FROM tableName
- *
  * Please refer {{org.apache.spark.sql.parser.CarbonSpark2SqlParser}}
  */
-
 public enum DataMapClassProvider {
-  LUCENE("org.apache.carbondata.datamap.lucene.LuceneFineGrainDataMapFactory", "lucene"),
-  BLOOMFILTER("org.apache.carbondata.datamap.bloom.BloomCoarseGrainDataMapFactory", "bloomfilter"),
+  LUCENE("org.apache.carbondata.datamap.lucene.LuceneFineGrainIndexFactory", "lucene"),
+  BLOOMFILTER("org.apache.carbondata.datamap.bloom.BloomCoarseGrainIndexFactory", "bloomfilter"),
 
 Review comment:
   Please change the package for lucene and bloom classes

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

Indhumathi27 commented on a change in pull request #3661: [WIP] Support materialized view
URL: https://github.com/apache/carbondata/pull/3661#discussion_r396326086
 
 

 ##########
 File path: core/src/main/java/org/apache/carbondata/core/view/MaterializedViewProvider.java
 ##########
 @@ -0,0 +1,573 @@
+/*
+ * 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.carbondata.core.view;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStreamWriter;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.carbondata.common.annotations.InterfaceAudience;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.constants.CarbonCommonConstants;
+import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
+import org.apache.carbondata.core.datastore.filesystem.CarbonFileFilter;
+import org.apache.carbondata.core.datastore.impl.FileFactory;
+import org.apache.carbondata.core.fileoperations.AtomicFileOperationFactory;
+import org.apache.carbondata.core.fileoperations.AtomicFileOperations;
+import org.apache.carbondata.core.fileoperations.FileWriteOperation;
+import org.apache.carbondata.core.locks.CarbonLockFactory;
+import org.apache.carbondata.core.locks.CarbonLockUtil;
+import org.apache.carbondata.core.locks.ICarbonLock;
+import org.apache.carbondata.core.locks.LockUsage;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.metadata.schema.table.RelationIdentifier;
+import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
+import org.apache.carbondata.core.statusmanager.SegmentStatus;
+import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
+import org.apache.carbondata.core.util.CarbonProperties;
+import org.apache.carbondata.core.util.CarbonUtil;
+import org.apache.carbondata.core.util.path.CarbonTablePath;
+
+import com.google.gson.Gson;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.log4j.Logger;
+
+/**
+ * Stores mv schema in disk as json format
+ */
[e-mail]
+public class MaterializedViewProvider {
+
+  private static final Logger LOG = LogServiceFactory.getLogService(
+      MaterializedViewProvider.class.getCanonicalName());
+
+  private static final String STATUS_FILE_NAME = "mv_status";
+
+  private final String storeLocation;
+
+  private final Map<String, SchemaProvider> schemaProviders = new ConcurrentHashMap<>();
+
+  private MaterializedViewProvider(String storeLocation) {
+    this.storeLocation = storeLocation;
+  }
+
+  public static MaterializedViewProvider get() {
+    String storeLocation =
+        CarbonProperties.getInstance().getProperty(CarbonCommonConstants.STORE_LOCATION);
+    if (storeLocation == null) {
+      throw new RuntimeException(
+          "Property [" + CarbonCommonConstants.STORE_LOCATION + "] is not set.");
+    }
+    return new MaterializedViewProvider(storeLocation);
+  }
+
+  private static String getSchemaPath(String schemaRoot, String viewName) {
+    return schemaRoot + CarbonCommonConstants.FILE_SEPARATOR + "mv_schema." + viewName;
+  }
+
+  private SchemaProvider getSchemaProvider(String databaseName) {
+    String databaseNameUpper = databaseName.toUpperCase();
+    SchemaProvider schemaProvider = this.schemaProviders.get(databaseNameUpper);
+    if (schemaProvider == null) {
+      synchronized (this.schemaProviders) {
+        schemaProvider = this.schemaProviders.get(databaseNameUpper);
+        if (schemaProvider == null) {
+          String databaseLocation;
+          if (databaseNameUpper.equalsIgnoreCase("default")) {
 
 Review comment:
   Use CarbonCommonConstants.DATABASE_DEFAULT_NAME

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

Indhumathi27 commented on a change in pull request #3661: [WIP] Support materialized view
URL: https://github.com/apache/carbondata/pull/3661#discussion_r396358616
 
 

 ##########
 File path: integration/spark/src/main/scala/org/apache/spark/sql/listeners/ShowCacheEventListeners.scala
 ##########
 @@ -40,7 +40,7 @@ object ShowCachePreMVEventListener extends OperationEventListener {
       case showTableCacheEvent: ShowTableCacheEvent =>
         val carbonTable = showTableCacheEvent.carbonTable
         val internalCall = showTableCacheEvent.internalCall
-        if (carbonTable.isChildTableForMV && !internalCall) {
+        if (carbonTable.isMaterializedView && !internalCall) {
 
 Review comment:
   Show cache is not handled for mv after refactory. Please check

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

Indhumathi27 commented on a change in pull request #3661: [WIP] Support materialized view
URL: https://github.com/apache/carbondata/pull/3661#discussion_r396349479
 
 

 ##########
 File path: integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
 ##########
 @@ -1011,14 +1015,25 @@ object CarbonDataRDDFactory {
       throw new Exception(errorMessage)
     } else {
       DataMapStatusManager.disableAllLazyDataMaps(carbonTable)
+      val viewManager = MaterializedViewManagerInSpark.get(session)
+      val viewSchemas = new util.ArrayList[MaterializedViewSchema]()
+      for (viewSchema <- viewManager.getSchemasOnTable(carbonTable).asScala) {
+        if (viewSchema.isRefreshOnManual) {
+          viewSchemas.add(viewSchema)
+        }
+      }
+      viewManager.setStatus(viewSchemas, MaterializedViewStatus.DISABLED)
       if (overwriteTable) {
         val allDataMapSchemas = DataMapStoreManager.getInstance
           .getDataMapSchemasOfTable(carbonTable).asScala
           .filter(dataMapSchema => null != dataMapSchema.getRelationIdentifier &&
-                                   !dataMapSchema.isIndexDataMap).asJava
+                                   !dataMapSchema.isIndex).asJava
 
 Review comment:
   Can remove code from line:1027 to 1033, as it is added for mv

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

Indhumathi27 commented on a change in pull request #3661: [WIP] Support materialized view
URL: https://github.com/apache/carbondata/pull/3661#discussion_r396354903
 
 

 ##########
 File path: integration/spark/src/main/scala/org/apache/carbondata/view/MaterializedViewRefresher.scala
 ##########
 @@ -0,0 +1,398 @@
+/*
+ * 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.carbondata.view
+
+import java._
+import java.io.IOException
+
+import scala.collection.JavaConverters._
+
+import com.google.gson.Gson
+import org.apache.log4j.Logger
+import org.apache.spark.sql.{CarbonUtils, SparkSession}
+import org.apache.spark.sql.execution.command.management.CarbonInsertIntoCommand
+import org.apache.spark.sql.parser.MaterializedViewQueryParser
+
+import org.apache.carbondata.common.exceptions.sql.NoSuchMaterializedViewException
+import org.apache.carbondata.common.logging.LogServiceFactory
+import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.locks.ICarbonLock
+import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, RelationIdentifier}
+import org.apache.carbondata.core.statusmanager.{LoadMetadataDetails, SegmentStatus, SegmentStatusManager}
+import org.apache.carbondata.core.util.path.CarbonTablePath
+import org.apache.carbondata.core.view.{MaterializedViewSchema, MaterializedViewStatus}
+import org.apache.carbondata.processing.util.CarbonLoaderUtil
+
+class MaterializedViewRefresher{
+
+}
+
+object MaterializedViewRefresher {
+
+  private val LOGGER: Logger = LogServiceFactory.getLogService(
+    classOf[MaterializedViewRefresher].getCanonicalName)
+
+  /**
+   * Refresh the mv by loading all existing data from associated table
+   * This is called when refreshing the mv when
+   * 1. after mv creation and no "WITH DEFERRED REBUILD" defined
+   * 2. user manually trigger REFRESH MATERIALIZED VIEW command
+   */
+  @throws[IOException]
+  @throws[NoSuchMaterializedViewException]
+  def refresh(viewSchema: MaterializedViewSchema, session: SparkSession): Boolean = {
+    var newLoadName: String = ""
+    var segmentMap: String = ""
+    val viewTable: CarbonTable = CarbonTable.buildFromTablePath(
+      viewSchema.getIdentifier.getTableName,
+      viewSchema.getIdentifier.getDatabaseName,
+      viewSchema.getIdentifier.getTablePath,
+      viewSchema.getIdentifier.getTableId)
+    val viewIdentifier = viewSchema.getIdentifier
+    val viewTableIdentifier = viewTable.getAbsoluteTableIdentifier
+    // Clean up the old invalid segment data before creating a new entry for new load.
+    SegmentStatusManager.deleteLoadsAndUpdateMetadata(viewTable, false, null)
+    val segmentStatusManager: SegmentStatusManager = new SegmentStatusManager(viewTableIdentifier)
+    // Acquire table status lock to handle concurrent dataloading
+    val lock: ICarbonLock = segmentStatusManager.getTableStatusLock
+    val segmentMapping: util.Map[String, util.List[String]] =
+      new util.HashMap[String, util.List[String]]
+    val viewManager = MaterializedViewManagerInSpark.get(session)
+    try if (lock.lockWithRetries) {
+      LOGGER.info("Acquired lock for mv " + viewIdentifier + " for table status updation")
+      val viewTableMetadataPath: String =
+        CarbonTablePath.getMetadataPath(viewIdentifier.getTablePath)
+      val loadMetadataDetails = SegmentStatusManager.readLoadMetadata(viewTableMetadataPath)
+      val loadMetadataDetailList: util.List[LoadMetadataDetails] =
+        new util.ArrayList[LoadMetadataDetails](CarbonCommonConstants.DEFAULT_COLLECTION_SIZE)
+      // Mark for delete all stale loadMetadetail
+      for (loadMetadataDetail <- loadMetadataDetails) {
+        if (((loadMetadataDetail.getSegmentStatus eq SegmentStatus.INSERT_IN_PROGRESS) ||
+             (loadMetadataDetail.getSegmentStatus eq SegmentStatus.INSERT_OVERWRITE_IN_PROGRESS)) &&
+            loadMetadataDetail.getVisibility.equalsIgnoreCase("false")) {
+          loadMetadataDetail.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE)
+        }
+        loadMetadataDetailList.add(loadMetadataDetail)
+      }
+      if (viewSchema.isRefreshOnManual) {
+        // check if rebuild to mv is already in progress and throw exception
+        if (loadMetadataDetails.nonEmpty) {
+          for (loadMetaDetail <- loadMetadataDetails) {
+            if (((loadMetaDetail.getSegmentStatus eq SegmentStatus.INSERT_IN_PROGRESS) ||
+                 (loadMetaDetail.getSegmentStatus eq SegmentStatus.INSERT_OVERWRITE_IN_PROGRESS)) &&
+                SegmentStatusManager.isLoadInProgress(viewTableIdentifier,
+                  loadMetaDetail.getLoadName)) {
+              throw new RuntimeException(
+                "Rebuild to materialized view " + viewSchema.getIdentifier.getTableName +
+                  " is already in progress")
+            }
+          }
+        }
+      }
+      if (viewSchema.isRefreshIncremental) {
+        if (!getSpecificSegmentsTobeLoaded(viewSchema, segmentMapping, loadMetadataDetailList)) {
+          return false
+        }
+      } else {
+          // set segment mapping only for carbondata table
+          val associatedTableIds =
+            viewSchema.getAssociatedTables.asScala.filter(_.isCarbonDataTable)
+          for (associatedTableId <- associatedTableIds) {
+            val associatedTableSegmentList: util.List[String] =
+              SegmentStatusManager.getValidSegmentList(associatedTableId)
+            if (associatedTableSegmentList.isEmpty) {
+              return false
+            }
+            segmentMapping.put(associatedTableId.toString, associatedTableSegmentList)
+          }
+        }
+      segmentMap = new Gson().toJson(segmentMapping)
+      // To handle concurrent dataloading to mv, create new loadMetaEntry and
+      // set segmentMap to new loadMetaEntry and pass new segmentId with load command
+      val loadMetadataDetail: LoadMetadataDetails = new LoadMetadataDetails
+      val segmentId: String = String.valueOf(
+        SegmentStatusManager.createNewSegmentId(loadMetadataDetails))
+      loadMetadataDetail.setLoadName(segmentId)
+      loadMetadataDetail.setSegmentStatus(SegmentStatus.INSERT_IN_PROGRESS)
+      loadMetadataDetail.setExtraInfo(segmentMap)
+      loadMetadataDetailList.add(loadMetadataDetail)
+      newLoadName = segmentId
+      SegmentStatusManager.writeLoadDetailsIntoFile(CarbonTablePath.getTableStatusFilePath(
+        viewSchema.getIdentifier.getTablePath),
+        loadMetadataDetailList.toArray(new Array[LoadMetadataDetails](loadMetadataDetailList
+          .size)))
+    } else {
+      LOGGER.error("Not able to acquire the lock for Table status updation for table " +
+                   viewSchema.getIdentifier.getDatabaseName + "." +
+                   viewSchema.getIdentifier.getTableName)
+      viewManager.setStatus(viewSchema.getIdentifier, MaterializedViewStatus.DISABLED)
+      return false
+    } finally {
+      if (lock.unlock) {
+        LOGGER.info("Table unlocked successfully after table status updation" +
+                    viewSchema.getIdentifier.getDatabaseName + "." +
+                    viewSchema.getIdentifier.getTableName)
+      } else {
+        LOGGER.error("Unable to unlock Table lock for table" +
+                     viewSchema.getIdentifier.getDatabaseName + "." +
+                     viewSchema.getIdentifier.getTableName +
+                     " during table status updation")
+      }
+    }
+    refreshInternal(viewManager, viewSchema, viewTable, newLoadName, segmentMapping, session)
+  }
+
+  @throws[IOException]
+  private def refreshInternal(
+      viewManager: MaterializedViewManagerInSpark,
+      viewSchema: MaterializedViewSchema,
+      viewTable: CarbonTable,
+      newLoadName: String,
+      segmentMap: java.util.Map[String, java.util.List[String]],
+      session: SparkSession): Boolean = {
+    val query = viewSchema.getQuery
+    if (query != null) {
+      val viewIdentifier = viewSchema.getIdentifier
+      val updatedQuery = MaterializedViewQueryParser.getQuery(query, session)
+      val isFullRefresh = !viewSchema.isRefreshIncremental
+      // Set specified segments for incremental load
+      val segmentMapIterator = segmentMap.entrySet().iterator()
+      while (segmentMapIterator.hasNext) {
+        val entry = segmentMapIterator.next()
+        setInputSegments(entry.getKey, entry.getValue)
+      }
+      val header = viewTable.getTableInfo.getFactTable.getListOfColumns.asScala
+        .filter { column =>
+          !column.getColumnName
+            .equalsIgnoreCase(CarbonCommonConstants.DEFAULT_INVISIBLE_DUMMY_MEASURE)
+        }.sortBy(_.getSchemaOrdinal).map(_.getColumnName).mkString(",")
+      val insertIntoCommand = CarbonInsertIntoCommand(
+        databaseNameOp = Some(viewIdentifier.getDatabaseName),
+        tableName = viewIdentifier.getTableName,
+        options = scala.collection.immutable.Map("fileheader" -> header),
+        isFullRefresh,
+        logicalPlan = updatedQuery.queryExecution.analyzed,
+        tableInfo = viewTable.getTableInfo,
+        internalOptions = Map("mergedSegmentName" -> newLoadName,
+          CarbonCommonConstants.IS_INTERNAL_LOAD_CALL -> "true"),
+        partition = Map.empty)
+      try {
+        insertIntoCommand.run(session)
+      } catch {
+        case exception: Exception =>
+          // If load to dataMap table fails, disable the dataMap and if newLoad is still
+          // in INSERT_IN_PROGRESS state, mark for delete the newLoad and update table status file
+          viewManager.setStatus(viewSchema.getIdentifier, MaterializedViewStatus.DISABLED)
+          LOGGER.error("Data Load failed for DataMap: ", exception)
+          CarbonLoaderUtil.updateTableStatusInCaseOfFailure(
+            newLoadName,
+            viewTable.getAbsoluteTableIdentifier,
+            viewTable.getTableName,
+            viewTable.getDatabaseName,
+            viewTable.getTablePath,
+            viewTable.getMetadataPath)
+          throw exception
+      } finally {
+        unsetInputSegments(viewSchema)
+      }
+    }
+    true
+  }
+
+  /**
+   * This method will compare mainTable and dataMapTable segment List and loads only newly added
+   * segment from main table to dataMap table.
+   * In case if mainTable is compacted, then based on dataMap to mainTables segmentMapping, dataMap
+   * will be loaded
+   * Eg:
+   * case 1: Consider mainTableSegmentList: {0, 1, 2}, dataMapToMainTable segmentMap:
+   * { 0 -> 0, 1-> 1,2}. If (1, 2) segments of main table are compacted to 1.1 and new segment (3)
+   * is loaded to main table, then mainTableSegmentList will be updated to{0, 1.1, 3}.
+   * In this case, segment (1) of dataMap table will be marked for delete, and new segment
+   * {2 -> 1.1, 3} will be loaded to dataMap table
+   * case 2: Consider mainTableSegmentList: {0, 1, 2, 3}, dataMapToMainTable segmentMap:
+   * { 0 -> 0,1,2, 1-> 3}. If (1, 2) segments of main table are compacted to 1.1 and new segment
+   * (4) is loaded to main table, then mainTableSegmentList will be updated to {0, 1.1, 3, 4}.
+   * In this case, segment (0) of dataMap table will be marked for delete and segment (0) of
+   * main table will be added to validSegmentList which needs to be loaded again. Now, new dataMap
+   * table segment (2) with main table segmentList{2 -> 1.1, 4, 0} will be loaded to dataMap table.
+   * dataMapToMainTable segmentMap will be updated to {1 -> 3, 2 -> 1.1, 4, 0} after rebuild
+   */
+  @throws[IOException]
+  private def getSpecificSegmentsTobeLoaded(schema: MaterializedViewSchema,
+      segmentMapping: util.Map[String, util.List[String]],
+      listOfLoadFolderDetails: util.List[LoadMetadataDetails]): Boolean = {
+    val relationIdentifiers: util.List[RelationIdentifier] = schema.getAssociatedTables
+    // invalidDataMapSegmentList holds segment list which needs to be marked for delete
+    val invalidDataMapSegmentList: util.HashSet[String] = new util.HashSet[String]
 
 Review comment:
   Can check and rename all variables to mv

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

Indhumathi27 commented on a change in pull request #3661: [WIP] Support materialized view
URL: https://github.com/apache/carbondata/pull/3661#discussion_r396329129
 
 

 ##########
 File path: core/src/main/java/org/apache/carbondata/core/view/MaterializedViewProvider.java
 ##########
 @@ -0,0 +1,573 @@
+/*
+ * 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.carbondata.core.view;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStreamWriter;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.carbondata.common.annotations.InterfaceAudience;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.constants.CarbonCommonConstants;
+import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
+import org.apache.carbondata.core.datastore.filesystem.CarbonFileFilter;
+import org.apache.carbondata.core.datastore.impl.FileFactory;
+import org.apache.carbondata.core.fileoperations.AtomicFileOperationFactory;
+import org.apache.carbondata.core.fileoperations.AtomicFileOperations;
+import org.apache.carbondata.core.fileoperations.FileWriteOperation;
+import org.apache.carbondata.core.locks.CarbonLockFactory;
+import org.apache.carbondata.core.locks.CarbonLockUtil;
+import org.apache.carbondata.core.locks.ICarbonLock;
+import org.apache.carbondata.core.locks.LockUsage;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.metadata.schema.table.RelationIdentifier;
+import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
+import org.apache.carbondata.core.statusmanager.SegmentStatus;
+import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
+import org.apache.carbondata.core.util.CarbonProperties;
+import org.apache.carbondata.core.util.CarbonUtil;
+import org.apache.carbondata.core.util.path.CarbonTablePath;
+
+import com.google.gson.Gson;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.log4j.Logger;
+
+/**
+ * Stores mv schema in disk as json format
+ */
[e-mail]
+public class MaterializedViewProvider {
+
+  private static final Logger LOG = LogServiceFactory.getLogService(
+      MaterializedViewProvider.class.getCanonicalName());
+
+  private static final String STATUS_FILE_NAME = "mv_status";
+
+  private final String storeLocation;
+
+  private final Map<String, SchemaProvider> schemaProviders = new ConcurrentHashMap<>();
+
+  private MaterializedViewProvider(String storeLocation) {
+    this.storeLocation = storeLocation;
+  }
+
+  public static MaterializedViewProvider get() {
+    String storeLocation =
+        CarbonProperties.getInstance().getProperty(CarbonCommonConstants.STORE_LOCATION);
+    if (storeLocation == null) {
+      throw new RuntimeException(
+          "Property [" + CarbonCommonConstants.STORE_LOCATION + "] is not set.");
+    }
+    return new MaterializedViewProvider(storeLocation);
+  }
+
+  private static String getSchemaPath(String schemaRoot, String viewName) {
+    return schemaRoot + CarbonCommonConstants.FILE_SEPARATOR + "mv_schema." + viewName;
+  }
+
+  private SchemaProvider getSchemaProvider(String databaseName) {
+    String databaseNameUpper = databaseName.toUpperCase();
+    SchemaProvider schemaProvider = this.schemaProviders.get(databaseNameUpper);
+    if (schemaProvider == null) {
+      synchronized (this.schemaProviders) {
+        schemaProvider = this.schemaProviders.get(databaseNameUpper);
+        if (schemaProvider == null) {
+          String databaseLocation;
+          if (databaseNameUpper.equalsIgnoreCase("default")) {
+            databaseLocation = CarbonUtil.checkAndAppendHDFSUrl(this.storeLocation);
+          } else {
+            databaseLocation = CarbonUtil.checkAndAppendHDFSUrl(this.storeLocation +
+                CarbonCommonConstants.FILE_SEPARATOR + databaseName + ".db");
+          }
+          if (!FileFactory.getCarbonFile(databaseLocation).exists()) {
+            return null;
+          }
+          schemaProvider = new SchemaProvider(databaseLocation);
+          this.schemaProviders.put(databaseNameUpper, schemaProvider);
+        }
+      }
+    }
+    return schemaProvider;
+  }
+
+  public MaterializedViewSchema getSchema(MaterializedViewManager viewManager,
+      String databaseName, String viewName) throws IOException {
+    SchemaProvider schemaProvider = this.getSchemaProvider(databaseName);
+    if (schemaProvider == null) {
+      return null;
+    }
+    return schemaProvider.retrieveSchema(viewManager, viewName);
+  }
+
+  List<MaterializedViewSchema> getSchemas(MaterializedViewManager viewManager,
+      String databaseName, CarbonTable carbonTable) throws IOException {
+    SchemaProvider schemaProvider = this.getSchemaProvider(databaseName);
+    if (schemaProvider == null) {
+      return Collections.emptyList();
+    } else {
+      return schemaProvider.retrieveSchemas(viewManager, carbonTable);
+    }
+  }
+
+  List<MaterializedViewSchema> getSchemas(MaterializedViewManager viewManager,
+      String databaseName) throws IOException {
+    SchemaProvider schemaProvider = this.getSchemaProvider(databaseName);
+    if (schemaProvider == null) {
+      return Collections.emptyList();
+    } else {
+      return schemaProvider.retrieveAllSchemas(viewManager);
+    }
+  }
+
+  void saveSchema(MaterializedViewManager viewManager, String databaseName,
+      MaterializedViewSchema viewSchema) throws IOException {
+    SchemaProvider schemaProvider = this.getSchemaProvider(databaseName);
+    if (schemaProvider == null) {
+      throw new IOException("Database [" + databaseName + "] is not found.");
+    }
+    schemaProvider.saveSchema(viewManager, viewSchema);
+  }
+
+  public void dropSchema(String databaseName, String viewName)
+      throws IOException {
+    SchemaProvider schemaProvider = this.getSchemaProvider(databaseName);
+    if (schemaProvider == null) {
+      throw new IOException("Materialized view with name " + databaseName + "." + viewName +
+          " does not exists in storage");
+    }
+    schemaProvider.dropSchema(viewName);
+  }
+
+  private String getStatusFileName(String databaseName) {
+    if (databaseName.equalsIgnoreCase("default")) {
+      return this.storeLocation +
+          CarbonCommonConstants.FILE_SEPARATOR + "_system" +
+          CarbonCommonConstants.FILE_SEPARATOR + STATUS_FILE_NAME;
+    } else {
+      return this.storeLocation +
+          CarbonCommonConstants.FILE_SEPARATOR + databaseName + ".db" +
+          CarbonCommonConstants.FILE_SEPARATOR + "_system" +
+          CarbonCommonConstants.FILE_SEPARATOR + STATUS_FILE_NAME;
+    }
+  }
+
+  public List<MaterializedViewStatusDetail> getStatusDetails(String databaseName)
+      throws IOException {
+    String statusPath = this.getStatusFileName(databaseName);
+    Gson gsonObjectToRead = new Gson();
+    DataInputStream dataInputStream = null;
+    BufferedReader buffReader = null;
+    InputStreamReader inStream = null;
+    MaterializedViewStatusDetail[] statusDetails;
+    try {
+      if (!FileFactory.isFileExist(statusPath)) {
+        return Collections.emptyList();
+      }
+      dataInputStream = FileFactory.getDataInputStream(statusPath);
+      inStream = new InputStreamReader(dataInputStream,
+          Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET));
+      buffReader = new BufferedReader(inStream);
+      statusDetails = gsonObjectToRead.fromJson(buffReader,
+          MaterializedViewStatusDetail[].class);
+    } catch (IOException e) {
+      LOG.error("Failed to read datamap status", e);
+      throw e;
+    } finally {
+      CarbonUtil.closeStreams(buffReader, inStream, dataInputStream);
+    }
+    // if status details is null, return empty array
+    if (null == statusDetails) {
+      return Collections.emptyList();
+    }
+    return Arrays.asList(statusDetails);
+  }
+
+  private static ICarbonLock getStatusLock(String databaseName) {
+    return CarbonLockFactory
+        .getSystemLevelCarbonLockObj(
+            CarbonProperties.getInstance().getSystemFolderLocation(databaseName),
+            LockUsage.MATERIALIZED_VIEW_STATUS_LOCK);
+  }
+
+  /**
+   * Update or add the status of passed mvs with the given mv status. If the mv status
+   * given is enabled/disabled then updates/adds the mv, in case of drop it just removes it
+   * from the file.
+   * This method always overwrites the old file.
+   * @param schemaList schemas of which are need to be updated in mv status
+   * @param status  status to be updated for the mv schemas
+   */
+  public void updateStatus(List<MaterializedViewSchema> schemaList,
+      MaterializedViewStatus status) throws IOException {
+    if (schemaList == null || schemaList.size() == 0) {
+      // There is nothing to update
+      return;
+    }
+    final Map<String, List<MaterializedViewSchema>> schemasMapByDatabase = new HashMap<>();
+    for (MaterializedViewSchema schema : schemaList) {
+      String databaseName = schema.getIdentifier().getDatabaseName().toLowerCase();
+      List<MaterializedViewSchema> schemas = schemasMapByDatabase.get(databaseName);
+      if (schemas == null) {
+        schemas = new ArrayList<>();
+        schemasMapByDatabase.put(databaseName, schemas);
+      }
+      schemas.add(schema);
+    }
+    for (Map.Entry<String, List<MaterializedViewSchema>> entry : schemasMapByDatabase.entrySet()) {
+      this.updateStatus(entry.getKey(), entry.getValue(), status);
+    }
+  }
+
+  private void updateStatus(String databaseName, List<MaterializedViewSchema> schemaList,
+      MaterializedViewStatus status) throws IOException {
+    ICarbonLock carbonTableStatusLock = getStatusLock(databaseName);
+    boolean locked = false;
+    try {
+      locked = carbonTableStatusLock.lockWithRetries();
+      if (locked) {
+        LOG.info("Materialized view status lock has been successfully acquired.");
+        if (status == MaterializedViewStatus.ENABLED) {
+          // Enable mv only if mv tables and main table are in sync
+          if (!isViewCanBeEnabled(schemaList.get(0))) {
+            return;
+          }
+        }
+        List<MaterializedViewStatusDetail> statusDetailList =
+            new ArrayList<>(getStatusDetails(databaseName));
+        List<MaterializedViewStatusDetail> changedStatusDetails = new ArrayList<>();
+        List<MaterializedViewStatusDetail> newStatusDetails = new ArrayList<>();
+        for (MaterializedViewSchema schema : schemaList) {
+          boolean exists = false;
+          for (MaterializedViewStatusDetail statusDetail : statusDetailList) {
+            if (statusDetail.getIdentifier().equals(schema.getIdentifier())) {
+              statusDetail.setStatus(status);
+              changedStatusDetails.add(statusDetail);
+              exists = true;
+            }
+          }
+          if (!exists) {
+            newStatusDetails
+                .add(new MaterializedViewStatusDetail(schema.getIdentifier(),
+                    status));
+          }
+        }
+        // Add the newly added datamaps to the list.
+        if (newStatusDetails.size() > 0 &&
+            status != MaterializedViewStatus.DROPPED) {
+          statusDetailList.addAll(newStatusDetails);
+        }
+        // In case of dropped datamap, just remove from the list.
+        if (status == MaterializedViewStatus.DROPPED) {
+          statusDetailList.removeAll(changedStatusDetails);
+        }
+        writeLoadDetailsIntoFile(
+            this.getStatusFileName(databaseName),
+            statusDetailList.toArray(
+                new MaterializedViewStatusDetail[statusDetailList.size()]));
+      } else {
+        String errorMsg = "Upadating datamapstatus is failed due to another process taken the lock"
+            + " for updating it";
+        LOG.error(errorMsg);
+        throw new IOException(errorMsg + " Please try after some time.");
+      }
+    } finally {
+      if (locked) {
+        CarbonLockUtil.fileUnlock(carbonTableStatusLock, LockUsage.DATAMAP_STATUS_LOCK);
+      }
+    }
+  }
+
+  /**
+   * writes mv status details
+   */
+  private static void writeLoadDetailsIntoFile(String location,
+      MaterializedViewStatusDetail[] statusDetails) throws IOException {
+    FileFactory.touchFile(FileFactory.getCarbonFile(location),
+        new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
+    AtomicFileOperations fileWrite = AtomicFileOperationFactory.getAtomicFileOperations(location);
+    BufferedWriter brWriter = null;
+    DataOutputStream dataOutputStream = null;
+    Gson gsonObjectToWrite = new Gson();
+    // write the updated data into the mv status file.
+    try {
+      dataOutputStream = fileWrite.openForWrite(FileWriteOperation.OVERWRITE);
+      brWriter = new BufferedWriter(new OutputStreamWriter(dataOutputStream,
+          Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)));
+
+      String metadataInstance = gsonObjectToWrite.toJson(statusDetails);
+      brWriter.write(metadataInstance);
+    } catch (IOException ioe) {
+      LOG.error("Error message: " + ioe.getLocalizedMessage());
+      fileWrite.setFailed();
+      throw ioe;
+    } finally {
+      if (null != brWriter) {
+        brWriter.flush();
+      }
+      CarbonUtil.closeStreams(brWriter);
+      fileWrite.close();
+    }
+
+  }
+
+  /**
+   * This method checks if main table and mv table are synchronised or not. If synchronised
+   * return true to enable the mv
+   *
+   * @param schema of mv to be disabled or enabled
+   * @return flag to enable or disable mv
+   * @throws IOException
+   */
+  private static boolean isViewCanBeEnabled(MaterializedViewSchema schema)
+      throws IOException {
+    if (!schema.isRefreshIncremental()) {
+      return true;
+    }
+    boolean isViewCanBeEnabled = true;
+    String viewMetadataPath =
+        CarbonTablePath.getMetadataPath(schema.getIdentifier().getTablePath());
+    LoadMetadataDetails[] viewLoadMetadataDetails =
+        SegmentStatusManager.readLoadMetadata(viewMetadataPath);
+    Map<String, List<String>> viewSegmentMap = new HashMap<>();
+    for (LoadMetadataDetails loadMetadataDetail : viewLoadMetadataDetails) {
+      if (loadMetadataDetail.getSegmentStatus() == SegmentStatus.SUCCESS) {
+        Map<String, List<String>> segmentMap =
+            new Gson().fromJson(loadMetadataDetail.getExtraInfo(), Map.class);
+        if (viewSegmentMap.isEmpty()) {
+          viewSegmentMap.putAll(segmentMap);
+        } else {
+          for (Map.Entry<String, List<String>> entry : segmentMap.entrySet()) {
+            if (null != viewSegmentMap.get(entry.getKey())) {
+              viewSegmentMap.get(entry.getKey()).addAll(entry.getValue());
+            }
+          }
+        }
+      }
+    }
+    List<RelationIdentifier> associatedTables = schema.getAssociatedTables();
+    for (RelationIdentifier associatedTable : associatedTables) {
+      List<String> associatedTableSegmentList =
+          SegmentStatusManager.getValidSegmentList(associatedTable);
+      if (!associatedTableSegmentList.isEmpty()) {
+        if (viewSegmentMap.isEmpty()) {
+          isViewCanBeEnabled = false;
+        } else {
+          isViewCanBeEnabled = viewSegmentMap.get(
+              associatedTable.getDatabaseName() + CarbonCommonConstants.POINT +
+                  associatedTable.getTableName()).containsAll(associatedTableSegmentList);
+        }
+      }
+    }
+    return isViewCanBeEnabled;
+  }
+
+  /**
+   * Data map schema provider of a database.
+   */
+  private static final class SchemaProvider {
+
+    private String systemDirectory;
+
+    private String schemaIndexFilePath;
+
+    private long lastModifiedTime;
+
+    private Set<MaterializedViewSchema> schemas = new HashSet<>();
+
+    SchemaProvider(String databaseLocation) {
+      final String systemDirectory =
+          databaseLocation + CarbonCommonConstants.FILE_SEPARATOR + "_system";
+      this.systemDirectory = systemDirectory;
+      this.schemaIndexFilePath = systemDirectory + CarbonCommonConstants.FILE_SEPARATOR +
+          "mv_schema_index";
+    }
+
+    void saveSchema(MaterializedViewManager viewManager, MaterializedViewSchema viewSchema)
+        throws IOException {
+      BufferedWriter brWriter = null;
+      DataOutputStream dataOutputStream = null;
+      Gson gsonObjectToWrite = new Gson();
+      String schemaPath =
+          getSchemaPath(this.systemDirectory, viewSchema.getIdentifier().getTableName());
+      if (FileFactory.isFileExist(schemaPath)) {
 
 Review comment:
   Can remove this check, as if it is already present in CarbonCreateMaterializedViewCommand:78

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

Indhumathi27 commented on a change in pull request #3661: [WIP] Support materialized view
URL: https://github.com/apache/carbondata/pull/3661#discussion_r396362694
 
 

 ##########
 File path: integration/spark/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonCleanFilesCommand.scala
 ##########
 @@ -65,7 +66,7 @@ case class CarbonCleanFilesCommand(
       val allDataMapSchemas = DataMapStoreManager.getInstance
         .getDataMapSchemasOfTable(carbonTable).asScala
         .filter(dataMapSchema => null != dataMapSchema.getRelationIdentifier &&
-                                 !dataMapSchema.isIndexDataMap)
+                                 !dataMapSchema.isIndex)
 
 Review comment:
   Can remove this if check for mv

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

Indhumathi27 commented on a change in pull request #3661: [WIP] Support materialized view
URL: https://github.com/apache/carbondata/pull/3661#discussion_r396367410
 
 

 ##########
 File path: integration/spark/src/main/scala/org/apache/spark/sql/listeners/DropCacheEventListeners.scala
 ##########
 @@ -31,31 +32,28 @@ import org.apache.carbondata.core.metadata.schema.datamap.DataMapClassProvider
 import org.apache.carbondata.core.metadata.schema.table.DataMapSchema
 import org.apache.carbondata.events.{DropTableCacheEvent, Event, OperationContext, OperationEventListener}
 
-object DropCacheDataMapEventListener extends OperationEventListener {
+object DropCacheMVEventListener extends OperationEventListener {
 
-  val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
+  val LOGGER: Logger = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
 
   /**
    * Called on a specified event occurrence
-   *
-   * @param event
-   * @param operationContext
    */
   override protected def onEvent(event: Event, operationContext: OperationContext): Unit = {
     event match {
       case dropCacheEvent: DropTableCacheEvent =>
         val carbonTable = dropCacheEvent.carbonTable
         val sparkSession = dropCacheEvent.sparkSession
         val internalCall = dropCacheEvent.internalCall
-        if (carbonTable.isChildTableForMV && !internalCall) {
+        if (carbonTable.isMaterializedView && !internalCall) {
           throw new UnsupportedOperationException("Operation not allowed on child table.")
         }
 
         if (carbonTable.hasMVCreated) {
           val childrenSchemas = DataMapStoreManager.getInstance
             .getDataMapSchemasOfTable(carbonTable).asScala
             .filter(dataMapSchema => null != dataMapSchema.getRelationIdentifier &&
-                                     !dataMapSchema.isIndexDataMap)
+                                     !dataMapSchema.isIndex)
 
 Review comment:
   Remove this if check code added for mv, as drop cache handled in DropCacheCommand

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

Indhumathi27 commented on a change in pull request #3661: [WIP] Support materialized view
URL: https://github.com/apache/carbondata/pull/3661#discussion_r396363582
 
 

 ##########
 File path: integration/spark/src/main/scala/org/apache/spark/sql/execution/command/mutation/CarbonProjectForDeleteCommand.scala
 ##########
 @@ -136,11 +137,18 @@ private[sql] case class CarbonProjectForDeleteCommand(
       val allDataMapSchemas = DataMapStoreManager.getInstance
         .getDataMapSchemasOfTable(carbonTable).asScala
         .filter(dataMapSchema => null != dataMapSchema.getRelationIdentifier &&
-                                 !dataMapSchema.isIndexDataMap).asJava
+                                 !dataMapSchema.isIndex).asJava
 
 Review comment:
   Can remove code from line 137 to 143 added for mv, after this refactory

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

Indhumathi27 commented on a change in pull request #3661: [WIP] Support materialized view
URL: https://github.com/apache/carbondata/pull/3661#discussion_r396366706
 
 

 ##########
 File path: integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonAnalysisRules.scala
 ##########
 @@ -207,16 +219,26 @@ case class CarbonIUDAnalysisRule(sparkSession: SparkSession) extends Rule[Logica
         val projList = Seq(UnresolvedAlias(UnresolvedStar(alias.map(Seq(_)))), tupleId)
         val carbonTable = CarbonEnv.getCarbonTable(table.tableIdentifier)(sparkSession)
         if (carbonTable != null) {
-          if (carbonTable.isChildTableForMV) {
+          if (carbonTable.isMaterializedView) {
             throw new UnsupportedOperationException(
               "Delete operation is not supported for datamap table")
           }
+          val viewManager = MaterializedViewManagerInSpark.get(sparkSession)
+          val viewSchemas = viewManager.getSchemasOnTable(carbonTable)
+          if (!viewSchemas.isEmpty) {
+            viewSchemas.asScala.foreach { schema =>
+              viewManager.setStatus(
+                schema.getIdentifier,
+                MaterializedViewStatus.DISABLED
+              )
+            }
+          }
           val indexSchemas = DataMapStoreManager.getInstance().getDataMapSchemasOfTable(carbonTable)
           if (carbonTable.hasMVCreated) {
             val allDataMapSchemas = DataMapStoreManager.getInstance
               .getDataMapSchemasOfTable(carbonTable).asScala
               .filter(dataMapSchema => null != dataMapSchema.getRelationIdentifier &&
-                                       !dataMapSchema.isIndexDataMap).asJava
+                                       !dataMapSchema.isIndex).asJava
 
 Review comment:
   remove this if check code

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

Indhumathi27 commented on a change in pull request #3661: [WIP] Support materialized view
URL: https://github.com/apache/carbondata/pull/3661#discussion_r396370616
 
 

 ##########
 File path: integration/spark/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
 ##########
 @@ -101,6 +100,12 @@ class CarbonSpark2SqlParser extends CarbonDDLSqlParser {
     loadDataNew | explainPlan | alterTableColumnRenameAndModifyDataType |
     alterTableAddColumns
 
+  protected lazy val materializedViewCommands: Parser[LogicalPlan] =
 
 Review comment:
   As already a new Parser => `MVParser` is already present. Why it is added?

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

Indhumathi27 commented on a change in pull request #3661: [WIP] Support materialized view
URL: https://github.com/apache/carbondata/pull/3661#discussion_r396365929
 
 

 ##########
 File path: integration/spark/src/main/scala/org/apache/spark/sql/execution/strategy/DDLHelper.scala
 ##########
 @@ -379,13 +380,17 @@ object DDLHelper {
           throw new MalformedCarbonCommandException(
             "Streaming property value is incorrect")
         }
-        if (carbonTable.hasMVCreated) {
+        if (carbonTable.isMaterializedView) {
+          throw new MalformedCarbonCommandException(
+            "Datamap table does not support set streaming property")
 
 Review comment:
   Change the exception to mv

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

Indhumathi27 commented on a change in pull request #3661: [WIP] Support materialized view
URL: https://github.com/apache/carbondata/pull/3661#discussion_r396368302
 
 

 ##########
 File path: integration/spark/src/main/scala/org/apache/spark/sql/listeners/MVListeners.scala
 ##########
 @@ -67,7 +67,7 @@ object AlterDataMaptableCompactionPostListener extends OperationEventListener {
     val allDataMapSchemas = DataMapStoreManager.getInstance
       .getDataMapSchemasOfTable(carbonTable).asScala
       .filter(dataMapSchema => null != dataMapSchema.getRelationIdentifier &&
-                               !dataMapSchema.isIndexDataMap)
+                               !dataMapSchema.isIndex)
 
 Review comment:
   Can delete this class

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

Indhumathi27 commented on a change in pull request #3661: [WIP] Support materialized view
URL: https://github.com/apache/carbondata/pull/3661#discussion_r396366570
 
 

 ##########
 File path: integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonAnalysisRules.scala
 ##########
 @@ -65,20 +67,30 @@ case class CarbonIUDAnalysisRule(sparkSession: SparkSession) extends Rule[Logica
       val projList = Seq(UnresolvedAlias(UnresolvedStar(alias.map(Seq(_)))), tupleId)
       val carbonTable = CarbonEnv.getCarbonTable(table.tableIdentifier)(sparkSession)
       if (carbonTable != null) {
+        val viewManager = MaterializedViewManagerInSpark.get(sparkSession)
+        val viewSchemas = viewManager.getSchemasOnTable(carbonTable)
+        if (!viewSchemas.isEmpty) {
+          viewSchemas.asScala.foreach { schema =>
+            viewManager.setStatus(
+              schema.getIdentifier,
+              MaterializedViewStatus.DISABLED
+            )
+          }
+        }
         val indexSchemas = DataMapStoreManager.getInstance().getDataMapSchemasOfTable(carbonTable)
         if (carbonTable.hasMVCreated) {
           val allDataMapSchemas = DataMapStoreManager.getInstance
             .getDataMapSchemasOfTable(carbonTable).asScala
             .filter(dataMapSchema => null != dataMapSchema.getRelationIdentifier &&
-                                     !dataMapSchema.isIndexDataMap).asJava
+                                     !dataMapSchema.isIndex).asJava
 
 Review comment:
   remove this if check code

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services

[GitBox]

Indhumathi27 commented on a change in pull request #3661: [WIP] Support materialized view
URL: https://github.com/apache/carbondata/pull/3661#discussion_r396363915
 
 

 ##########
 File path: integration/spark/src/main/scala/org/apache/spark/sql/execution/command/mutation/CarbonProjectForUpdateCommand.scala
 ##########
 @@ -203,10 +204,18 @@ private[sql] case class CarbonProjectForUpdateCommand(
       val allDataMapSchemas = DataMapStoreManager.getInstance
         .getDataMapSchemasOfTable(carbonTable).asScala
         .filter(dataMapSchema => null != dataMapSchema.getRelationIdentifier &&
-                                 !dataMapSchema.isIndexDataMap).asJava
+                                 !dataMapSchema.isIndex).asJava
 
 Review comment:
   Can remove code from line 204 to 210 added for mv, after this refactory

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services