| /* |
| * 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"); |
| } |
| } |
| } |
| } |
| |
| } |
