| /* |
| * 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.index.status; |
| |
| 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.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| |
| import org.apache.carbondata.common.logging.LogServiceFactory; |
| import org.apache.carbondata.core.constants.CarbonCommonConstants; |
| 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.index.IndexUtil; |
| 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.IndexSchema; |
| 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.log4j.Logger; |
| |
| /** |
| * It saves/serializes the array of {{@link IndexStatusDetail}} to disk in json format. |
| * It ensures the data consistance while concurrent write through write lock. It saves the status |
| * to the index status under the system folder. |
| */ |
| public class DiskBasedIndexStatusProvider implements IndexStatusStorageProvider { |
| |
| private static final Logger LOG = |
| LogServiceFactory.getLogService(DiskBasedIndexStatusProvider.class.getName()); |
| |
| private static final String INDEX_STATUS_FILE = "indexstatus"; |
| |
| @Override |
| public IndexStatusDetail[] getIndexStatusDetails() throws IOException { |
| String statusPath = CarbonProperties.getInstance().getSystemFolderLocation() |
| + CarbonCommonConstants.FILE_SEPARATOR + INDEX_STATUS_FILE; |
| Gson gsonObjectToRead = new Gson(); |
| DataInputStream dataInputStream = null; |
| BufferedReader buffReader = null; |
| InputStreamReader inStream = null; |
| IndexStatusDetail[] indexStatusDetails; |
| try { |
| if (!FileFactory.isFileExist(statusPath)) { |
| return new IndexStatusDetail[0]; |
| } |
| dataInputStream = FileFactory.getDataInputStream(statusPath); |
| inStream = new InputStreamReader(dataInputStream, |
| Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)); |
| buffReader = new BufferedReader(inStream); |
| indexStatusDetails = gsonObjectToRead.fromJson(buffReader, IndexStatusDetail[].class); |
| } catch (IOException e) { |
| LOG.error("Failed to read index status", e); |
| throw e; |
| } finally { |
| CarbonUtil.closeStreams(buffReader, inStream, dataInputStream); |
| } |
| |
| // if indexStatusDetails is null, return empty array |
| if (null == indexStatusDetails) { |
| return new IndexStatusDetail[0]; |
| } |
| |
| return indexStatusDetails; |
| } |
| |
| /** |
| * Update or add the status of passed indexes with the given indexstatus. If the indexstatus |
| * given is enabled/disabled then updates/adds the index, in case of drop it just removes it |
| * from the file. |
| * This method always overwrites the old file. |
| * @param indexSchemas schemas of which are need to be updated in index status |
| * @param indexStatus status to be updated for the index schemas |
| * @throws IOException |
| */ |
| @Override |
| public void updateIndexStatus(List<IndexSchema> indexSchemas, IndexStatus indexStatus) |
| throws IOException { |
| if (indexSchemas == null || indexSchemas.size() == 0) { |
| // There is nothing to update |
| return; |
| } |
| ICarbonLock carbonTableStatusLock = getIndexStatusLock(); |
| boolean locked = false; |
| try { |
| locked = carbonTableStatusLock.lockWithRetries(); |
| if (locked) { |
| LOG.info("index status lock has been successfully acquired."); |
| if (indexStatus == IndexStatus.ENABLED && !indexSchemas.get(0).isIndex()) { |
| // Enable index only if index tables and main table are in sync |
| if (!canIndexBeEnabled(indexSchemas.get(0))) { |
| return; |
| } |
| } |
| IndexStatusDetail[] indexStatusDetails = getIndexStatusDetails(); |
| List<IndexStatusDetail> indexStatusList = Arrays.asList(indexStatusDetails); |
| indexStatusList = new ArrayList<>(indexStatusList); |
| List<IndexStatusDetail> changedStatusDetails = new ArrayList<>(); |
| List<IndexStatusDetail> newStatusDetails = new ArrayList<>(); |
| for (IndexSchema indexSchema : indexSchemas) { |
| boolean exists = false; |
| for (IndexStatusDetail statusDetail : indexStatusList) { |
| if (statusDetail.getIndexName().equals(indexSchema.getIndexName())) { |
| statusDetail.setStatus(indexStatus); |
| changedStatusDetails.add(statusDetail); |
| exists = true; |
| } |
| } |
| if (!exists) { |
| newStatusDetails |
| .add(new IndexStatusDetail(indexSchema.getIndexName(), indexStatus)); |
| } |
| } |
| // Add the newly added index to the list. |
| if (newStatusDetails.size() > 0 && indexStatus != IndexStatus.DROPPED) { |
| indexStatusList.addAll(newStatusDetails); |
| } |
| // In case of dropped index, just remove from the list. |
| if (indexStatus == IndexStatus.DROPPED) { |
| indexStatusList.removeAll(changedStatusDetails); |
| } |
| writeLoadDetailsIntoFile(CarbonProperties.getInstance().getSystemFolderLocation() |
| + CarbonCommonConstants.FILE_SEPARATOR + INDEX_STATUS_FILE, |
| indexStatusList.toArray(new IndexStatusDetail[indexStatusList.size()])); |
| } else { |
| String errorMsg = "Upadating index status 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.INDEX_STATUS_LOCK); |
| } |
| } |
| } |
| |
| /** |
| * This method checks if main table and index table are synchronised or not. If synchronised |
| * return true to enable the index |
| * |
| * @param indexSchema of index to be disabled or enabled |
| * @return flag to enable or disable index |
| * @throws IOException |
| */ |
| private boolean canIndexBeEnabled(IndexSchema indexSchema) throws IOException { |
| boolean isIndexInSync = true; |
| String metaDataPath = |
| CarbonTablePath.getMetadataPath(indexSchema.getRelationIdentifier().getTablePath()); |
| LoadMetadataDetails[] indexLoadMetadataDetails = |
| SegmentStatusManager.readLoadMetadata(metaDataPath); |
| Map<String, List<String>> indexSegmentMap = new HashMap<>(); |
| for (LoadMetadataDetails loadMetadataDetail : indexLoadMetadataDetails) { |
| if (loadMetadataDetail.getSegmentStatus() == SegmentStatus.SUCCESS) { |
| Map<String, List<String>> segmentMap = |
| MVSegmentStatusUtil.getSegmentMap(loadMetadataDetail.getExtraInfo()); |
| if (indexSegmentMap.isEmpty()) { |
| indexSegmentMap.putAll(segmentMap); |
| } else { |
| for (Map.Entry<String, List<String>> entry : segmentMap.entrySet()) { |
| if (null != indexSegmentMap.get(entry.getKey())) { |
| indexSegmentMap.get(entry.getKey()).addAll(entry.getValue()); |
| } |
| } |
| } |
| } |
| } |
| List<RelationIdentifier> parentTables = indexSchema.getParentTables(); |
| for (RelationIdentifier parentTable : parentTables) { |
| List<String> mainTableValidSegmentList = |
| IndexUtil.getMainTableValidSegmentList(parentTable); |
| if (!mainTableValidSegmentList.isEmpty() && !indexSegmentMap.isEmpty()) { |
| isIndexInSync = indexSegmentMap.get( |
| parentTable.getDatabaseName() + CarbonCommonConstants.POINT + parentTable |
| .getTableName()).containsAll(mainTableValidSegmentList); |
| } else if (indexSegmentMap.isEmpty() && !mainTableValidSegmentList.isEmpty()) { |
| isIndexInSync = false; |
| } |
| } |
| return isIndexInSync; |
| } |
| |
| /** |
| * writes index status details |
| * |
| * @param indexStatusDetails |
| * @throws IOException |
| */ |
| private static void writeLoadDetailsIntoFile(String location, |
| IndexStatusDetail[] indexStatusDetails) throws IOException { |
| AtomicFileOperations fileWrite = AtomicFileOperationFactory.getAtomicFileOperations(location); |
| BufferedWriter brWriter = null; |
| DataOutputStream dataOutputStream = null; |
| Gson gsonObjectToWrite = new Gson(); |
| // write the updated data into the index status file. |
| try { |
| dataOutputStream = fileWrite.openForWrite(FileWriteOperation.OVERWRITE); |
| brWriter = new BufferedWriter(new OutputStreamWriter(dataOutputStream, |
| Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET))); |
| |
| String metadataInstance = gsonObjectToWrite.toJson(indexStatusDetails); |
| 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(); |
| } |
| |
| } |
| |
| private static ICarbonLock getIndexStatusLock() { |
| return CarbonLockFactory |
| .getSystemLevelCarbonLockObj(CarbonProperties.getInstance().getSystemFolderLocation(), |
| LockUsage.INDEX_STATUS_LOCK); |
| } |
| } |
