| /* |
| * 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.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) throws IOException { |
| List<MVSchema> schemas = getSchemas(); |
| for (MVSchema schema : schemas) { |
| for (RelationIdentifier relatedTable : schema.getRelatedTables()) { |
| if (relatedTable.getDatabaseName().equalsIgnoreCase(table.getDatabaseName()) && |
| relatedTable.getTableName().equalsIgnoreCase(table.getTableName())) { |
| return true; |
| } |
| } |
| } |
| return false; |
| } |
| |
| /** |
| * It gives all mv schemas of a given table. |
| * For show mv command. |
| */ |
| public List<MVSchema> getSchemasOnTable(CarbonTable table) |
| throws IOException { |
| List<MVSchema> schemasOnTable = new ArrayList<>(); |
| List<MVSchema> schemas = getSchemas(); |
| for (MVSchema schema : schemas) { |
| boolean isSchemaOnTable = false; |
| for (RelationIdentifier relatedTable : schema.getRelatedTables()) { |
| if (relatedTable.getDatabaseName().equalsIgnoreCase(table.getDatabaseName()) && |
| relatedTable.getTableName().equalsIgnoreCase(table.getTableName())) { |
| isSchemaOnTable = true; |
| break; |
| } |
| } |
| if (isSchemaOnTable) { |
| schemasOnTable.add(schema); |
| } |
| } |
| return schemasOnTable; |
| } |
| |
| /** |
| * 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()) { |
| schemas.addAll(this.getSchemas(database)); |
| } |
| 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); |
| } |
| |
| /** |
| * 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 index 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, |
| boolean reload) throws IOException { |
| MVCatalog<?> catalog = this.catalog; |
| if (reload || catalog == null) { |
| synchronized (lock) { |
| catalog = this.catalog; |
| if (reload || catalog == null) { |
| catalog = catalogFactory.newCatalog(); |
| List<MVSchema> schemas = getSchemas(); |
| if (null == catalog) { |
| throw new RuntimeException("Internal Error."); |
| } |
| for (MVSchema schema : schemas) { |
| try { |
| catalog.registerSchema(schema); |
| } catch (Exception e) { |
| // Ignore the schema |
| LOGGER.error("Error while registering schema", e); |
| } |
| } |
| 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()) { |
| statusDetails.addAll(this.getEnabledStatusDetails(database)); |
| } |
| 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(); |
| SegmentStatusManager segmentStatusManager = new SegmentStatusManager(AbsoluteTableIdentifier |
| .from(relationIdentifier.getTablePath(), |
| relationIdentifier.getDatabaseName(), |
| relationIdentifier.getTableName())); |
| 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); |
| for (LoadMetadataDetails entry : loadMetadataDetails) { |
| entry.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE); |
| } |
| SegmentStatusManager.writeLoadDetailsIntoFile( |
| CarbonTablePath.getTableStatusFilePath(relationIdentifier.getTablePath()), |
| 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"); |
| } |
| } |
| } |
| } |
| |
| } |
