content: /*
   * 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.IOException;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  
  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.locks.ICarbonLock;
  import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
  import org.apache.carbondata.core.metadata.CarbonMetadata;
  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.path.CarbonTablePath;
  
  import com.google.gson.Gson;
  import org.apache.log4j.Logger;
  
  /**
   * It maintains all the mv schemas in it.
   */
  @InterfaceAudience.Internal
  public abstract class MVManager {
  
    private static final Logger LOGGER =
        LogServiceFactory.getLogService(MVManager.class.getName());
  
    private final MVProvider schemaProvider = new MVProvider();
  
    private volatile MVCatalog<?> catalog;
  
    private final Object lock = new Object();
  
    public MVManager() {
  
    }
  
    public abstract List<String> getDatabases();
  
    public abstract String getDatabaseLocation(String databaseName);
  
    public boolean hasSchemaOnTable(CarbonTable table) {
      return !table.getMVTablesMap().isEmpty();
    }
  
    public boolean isMVInSyncWithParentTables(MVSchema mvSchema) throws IOException {
      return schemaProvider.isViewCanBeEnabled(mvSchema, true);
    }
  
    /**
     * It gives all mv schemas of a given table.
     * For show mv command.
     */
    public List<MVSchema> getSchemasOnTable(CarbonTable table)
        throws IOException {
      return getSchemas(table.getMVTablesMap());
    }
  
    /**
     * It gives all mv schemas of a given table.
     * For show mv command.
     */
    public List<MVSchema> getSchemasOnTable(String databaseName,
                                            CarbonTable carbonTable) throws IOException {
      return schemaProvider.getSchemas(this, databaseName, carbonTable);
    }
  
    /**
     * It gives all mv schemas from store.
     */
    public List<MVSchema> getSchemas() throws IOException {
      List<MVSchema> schemas = new ArrayList<>();
      for (String database : this.getDatabases()) {
        try {
          schemas.addAll(this.getSchemas(database));
        } catch (IOException ex) {
          throw ex;
        } catch (Exception ex) {
          LOGGER.error("Exception Occurred: Skipping MV schemas from database: " + database);
          if (LOGGER.isDebugEnabled()) {
            LOGGER.debug(ex.getMessage());
          }
        }
      }
      return schemas;
    }
  
    /**
     * It gives all mv schemas from given databases in the store
     */
    public List<MVSchema> getSchemas(Map<String, List<String>> mvTablesMap) throws IOException {
      List<MVSchema> schemas = new ArrayList<>();
      for (Map.Entry<String, List<String>> databaseEntry : mvTablesMap.entrySet()) {
        String database = databaseEntry.getKey();
        List<String> mvTables = databaseEntry.getValue();
        for (String mvTable : mvTables) {
          try {
            schemas.add(this.getSchema(database, mvTable));
          } catch (IOException ex) {
            LOGGER.error("Error while fetching MV schema " + mvTable + " from database: " + database);
            throw ex;
          } catch (Exception ex) {
            LOGGER.error(
                "Exception Occurred: Skipping MV schema " + mvTable + " from database: " + database);
            if (LOGGER.isDebugEnabled()) {
              LOGGER.debug(ex.getMessage());
            }
          }
        }
      }
      return schemas;
    }
  
    /**
     * It gives all mv schemas from store.
     */
    public List<MVSchema> getSchemas(String databaseName) throws IOException {
      return schemaProvider.getSchemas(this, databaseName);
    }
  
    public MVSchema getSchema(String databaseName, String viewName) throws IOException {
      return schemaProvider.getSchema(this, databaseName, viewName, false);
    }
  
    public MVSchema getSchema(String databaseName, String viewName, boolean isRegisterMV)
        throws IOException {
      return schemaProvider.getSchema(this, databaseName, viewName, isRegisterMV);
    }
  
    /**
     * Saves the mv schema to storage
     *
     * @param viewSchema mv schema
     */
    public void createSchema(String databaseName, MVSchema viewSchema)
        throws IOException {
      schemaProvider.saveSchema(this, databaseName, viewSchema);
    }
  
    /**
     * Drops the mv schema from storage
     *
     * @param viewName mv name
     */
    public void deleteSchema(String databaseName, String viewName) throws IOException {
      schemaProvider.dropSchema(this, databaseName, viewName);
    }
  
    /**
     * Get the mv catalog.
     */
    public MVCatalog<?> getCatalog() {
      return catalog;
    }
  
    /**
     * Get the mv catalog.
     */
    public MVCatalog<?> getCatalog(
        MVCatalogFactory<?> catalogFactory,
        List<MVSchema> currSchemas) throws IOException {
      MVCatalog<?> catalog = this.catalog;
      synchronized (lock) {
        catalog = this.catalog;
        if (catalog == null) {
          catalog = catalogFactory.newCatalog();
        }
        List<MVSchema> schemas = getSchemas();
        if (schemas.size() == currSchemas.size() && currSchemas.containsAll(schemas)) {
          return catalog;
        }
        for (MVSchema schema : schemas) {
          try {
            // register the schemas that are not already present in catalog.
            if (!currSchemas.contains(schema)) {
              catalog.registerSchema(schema);
            }
          } catch (Exception e) {
            // Ignore the schema
            LOGGER.error(
                "Error while registering schema for mv: " + schema.getIdentifier().getTableName());
            if (LOGGER.isDebugEnabled()) {
              LOGGER.debug(e.getMessage());
            }
          }
        }
        for (MVSchema currSchema : currSchemas) {
          try {
            // deregister the schemas from catalog if not present in the path.
            if (!schemas.contains(currSchema)) {
              catalog.deregisterSchema(currSchema.getIdentifier());
            }
          } catch (Exception e) {
            // Ignore the schema
            LOGGER.error("Error while deregistering schema for mv: " + currSchema.getIdentifier()
                .getTableName());
            if (LOGGER.isDebugEnabled()) {
              LOGGER.debug(e.getMessage());
            }
          }
        }
        this.catalog = catalog;
      }
      return catalog;
    }
  
    /**
     * In case of compaction on mv table,this method will merge the segment list of main table
     * and return updated segment mapping
     *
     * @param mergedLoadName      to find which all segments are merged to new compacted segment
     * @param viewSchema       of mv table
     * @param viewLoadMetadataDetails of mv table
     * @return updated segment map after merging segment list
     */
    @SuppressWarnings("unchecked")
    public static String getUpdatedSegmentMap(String mergedLoadName,
        MVSchema viewSchema,
        LoadMetadataDetails[] viewLoadMetadataDetails) {
      Map<String, List<String>> segmentMapping = new HashMap<>();
      List<RelationIdentifier> relationIdentifiers = viewSchema.getRelatedTables();
      for (RelationIdentifier relationIdentifier : relationIdentifiers) {
        for (LoadMetadataDetails loadMetadataDetail : viewLoadMetadataDetails) {
          if (loadMetadataDetail.getSegmentStatus() == SegmentStatus.COMPACTED) {
            if (mergedLoadName.equalsIgnoreCase(loadMetadataDetail.getMergedLoadName())) {
              Map segmentMap = new Gson().fromJson(loadMetadataDetail.getExtraInfo(), Map.class);
              if (segmentMapping.isEmpty()) {
                segmentMapping.putAll(segmentMap);
              } else {
                segmentMapping.get(relationIdentifier.getDatabaseName() + CarbonCommonConstants.POINT
                    + relationIdentifier.getTableName()).addAll(
                    (List<String>) segmentMap.get(
                        relationIdentifier.getDatabaseName() + CarbonCommonConstants.POINT
                            + relationIdentifier.getTableName()));
              }
            }
          }
        }
      }
      Gson gson = new Gson();
      return gson.toJson(segmentMapping);
    }
  
    /**
     * Get enabled mv status details
     */
    public List<MVStatusDetail> getEnabledStatusDetails() throws IOException {
      List<MVStatusDetail> statusDetails = new ArrayList<>();
      for (String database : this.getDatabases()) {
        try {
          statusDetails.addAll(this.getEnabledStatusDetails(database));
        } catch (IOException ex) {
          throw ex;
        } catch (Exception ex) {
          if (LOGGER.isDebugEnabled()) {
            LOGGER.debug(ex.getMessage());
          }
        }
      }
      return statusDetails;
    }
  
    /**
     * Get enabled mv status details
     */
    List<MVStatusDetail> getEnabledStatusDetails(String databaseName)
        throws IOException {
      List<MVStatusDetail> statusDetails = schemaProvider.getStatusDetails(this, databaseName);
      List<MVStatusDetail> enabledStatusDetails = new ArrayList<>(statusDetails.size());
      for (MVStatusDetail statusDetail : statusDetails) {
        if (statusDetail.getStatus() == MVStatus.ENABLED) {
          enabledStatusDetails.add(statusDetail);
        }
      }
      return enabledStatusDetails;
    }
  
    public void setStatus(RelationIdentifier viewIdentifier, MVStatus viewStatus)
        throws IOException {
      MVSchema schema = getSchema(
          viewIdentifier.getDatabaseName(), viewIdentifier.getTableName());
      if (schema != null) {
        schemaProvider.updateStatus(this, Collections.singletonList(schema), viewStatus);
      }
    }
  
    public void setStatus(List<MVSchema> viewSchemas, MVStatus viewStatus)
        throws IOException {
      if (viewSchemas != null && !viewSchemas.isEmpty()) {
        schemaProvider.updateStatus(this, viewSchemas, viewStatus);
      }
    }
  
    public void onDrop(String databaseName, String viewName)
        throws IOException {
      MVSchema viewSchema = getSchema(databaseName, viewName);
      if (viewSchema != null) {
        schemaProvider.updateStatus(
            this, Collections.singletonList(viewSchema), MVStatus.DROPPED);
      }
    }
  
    /**
     * This method will remove all segments of MV table in case of Insert-Overwrite/Update/Delete
     * operations on main table
     *
     * @param schemas mv schemas
     */
    public void onTruncate(List<MVSchema> schemas)
        throws IOException {
      for (MVSchema schema : schemas) {
        if (!schema.isRefreshOnManual()) {
          setStatus(schema.identifier, MVStatus.DISABLED);
        }
        RelationIdentifier relationIdentifier = schema.getIdentifier();
        CarbonTable carbonTable = CarbonMetadata.getInstance()
            .getCarbonTable(relationIdentifier.getDatabaseName(), relationIdentifier.getTableName());
        String tblStatusVersion = "";
        if (null != carbonTable) {
          tblStatusVersion = carbonTable.getTableStatusVersion();
        }
        SegmentStatusManager segmentStatusManager = new SegmentStatusManager(AbsoluteTableIdentifier
            .from(relationIdentifier.getTablePath(),
                relationIdentifier.getDatabaseName(),
                relationIdentifier.getTableName()), tblStatusVersion);
        ICarbonLock carbonLock = segmentStatusManager.getTableStatusLock();
        try {
          if (carbonLock.lockWithRetries()) {
            LOGGER.info("Acquired lock for table" + relationIdentifier.getDatabaseName() + "."
                + relationIdentifier.getTableName() + " for table status update");
            String metaDataPath =
                CarbonTablePath.getMetadataPath(relationIdentifier.getTablePath());
            LoadMetadataDetails[] loadMetadataDetails =
                SegmentStatusManager.readLoadMetadata(metaDataPath, tblStatusVersion);
            for (LoadMetadataDetails entry : loadMetadataDetails) {
              entry.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE);
            }
            SegmentStatusManager.writeLoadDetailsIntoFile(
                CarbonTablePath.getTableStatusFilePath(relationIdentifier.getTablePath(),
                    tblStatusVersion), loadMetadataDetails);
          } else {
            LOGGER.error("Not able to acquire the lock for Table status update for table "
                + relationIdentifier.getDatabaseName() + "." + relationIdentifier
                .getTableName());
          }
        } finally {
          if (carbonLock.unlock()) {
            LOGGER.info(
                "Table unlocked successfully after table status update" + relationIdentifier
                    .getDatabaseName() + "." + relationIdentifier.getTableName());
          } else {
            LOGGER.error(
                "Unable to unlock Table lock for table" + relationIdentifier.getDatabaseName()
                    + "." + relationIdentifier.getTableName()
                    + " during table status update");
          }
        }
      }
    }
  
  }
  