core/src/main/java/org/apache/carbondata/core/index/IndexStoreManager.java

/*
 * 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;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.carbondata.common.annotations.InterfaceAudience;
import org.apache.carbondata.common.exceptions.MetadataProcessException;
import org.apache.carbondata.common.exceptions.sql.MalformedIndexCommandException;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.block.SegmentPropertiesAndSchemaHolder;
import org.apache.carbondata.core.index.dev.IndexFactory;
import org.apache.carbondata.core.indexstore.BlockletDetailsFetcher;
import org.apache.carbondata.core.indexstore.SegmentPropertiesFetcher;
import org.apache.carbondata.core.indexstore.blockletindex.BlockletIndexFactory;
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.IndexSchema;
import org.apache.carbondata.core.mutate.UpdateVO;
import org.apache.carbondata.core.statusmanager.SegmentRefreshInfo;
import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
import org.apache.carbondata.core.statusmanager.SegmentUpdateStatusManager;
import org.apache.carbondata.core.util.CarbonProperties;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;

/**
 * It maintains all the Indexes in it.
 */
@InterfaceAudience.Internal
public final class IndexStoreManager {

  private static IndexStoreManager instance = new IndexStoreManager();

  public Map<String, List<TableIndex>> getTableIndexForAllTables() {
    return allIndexes;
  }

  /**
   * Contains the list of indexes for each table.
   */
  private Map<String, List<TableIndex>> allIndexes = new ConcurrentHashMap<>();

  /**
   * Contains the table name to the table path mapping.
   */
  private Map<String, String> tablePathMap = new ConcurrentHashMap<>();

  private Map<String, TableSegmentRefresher> segmentRefreshMap = new ConcurrentHashMap<>();

  private static final Logger LOGGER =
      LogServiceFactory.getLogService(IndexStoreManager.class.getName());

  private IndexStoreManager() {

  }

  /**
   * Collect's Coarse grain and Fine grain indexes on a table
   *
   * @return
   */
  public List<TableIndex> getAllCGAndFGIndexes(CarbonTable carbonTable) throws IOException {
    List<TableIndex> indexes = new ArrayList<>();
    // get bloom indexes and lucene indexes
    for (Map.Entry<String, Map<String, Map<String, String>>> providerEntry : carbonTable
        .getIndexesMap().entrySet()) {
      for (Map.Entry<String, Map<String, String>> indexEntry : providerEntry.getValue()
          .entrySet()) {
        IndexSchema indexSchema = new IndexSchema(indexEntry.getKey(),
            indexEntry.getValue().get(CarbonCommonConstants.INDEX_PROVIDER));
        indexSchema.setProperties(indexEntry.getValue());
        indexes.add(getIndex(carbonTable, indexSchema));
      }
    }
    return indexes;
  }

  /**
   * It gives the default index of the table. Default index of any table is BlockletIndex
   *
   * @param table
   * @return
   */
  public TableIndex getDefaultIndex(CarbonTable table) {
    return getIndex(table, BlockletIndexFactory.INDEX_SCHEMA);
  }

  /**
   * Get the index for reading data.
   */
  public TableIndex getIndex(CarbonTable table, IndexSchema indexSchema) {
    String tableId =
        table.getAbsoluteTableIdentifier().getCarbonTableIdentifier().getTableId();
    List<TableIndex> tableIndices = allIndexes.get(table.getTableId());
    if (tableIndices == null && !table.isTransactionalTable()) {
      String keyUsingTablePath = getKeyUsingTablePath(table.getTablePath());
      if (keyUsingTablePath != null) {
        tableId = keyUsingTablePath;
        tableIndices = allIndexes.get(tableId);
      }
    }
    // in case of file format or sdk, when table is dropped or schema is changed the indexes are
    // not cleared, they need to be cleared by using API, so compare the columns, if not same, clear
    // the indexes on that table
    if (allIndexes.size() > 0 && !CollectionUtils.isEmpty(allIndexes.get(tableId))
        && !allIndexes.get(tableId).get(0).getTable().getTableInfo().getFactTable()
        .getListOfColumns().equals(table.getTableInfo().getFactTable().getListOfColumns())) {
      clearIndex(tableId);
      tableIndices = null;
    }
    TableIndex index = null;
    if (tableIndices != null) {
      index = getTableIndex(indexSchema.getIndexName(), tableIndices);
    }
    if (index == null) {
      synchronized (tableId.intern()) {
        tableIndices = allIndexes.get(tableId);
        if (tableIndices != null) {
          index = getTableIndex(indexSchema.getIndexName(), tableIndices);
        }
        if (index == null) {
          try {
            index = createAndRegisterIndex(table, indexSchema);
          } catch (Exception e) {
            throw new RuntimeException(e);
          }
        }
      }
    }

    if (index == null) {
      throw new RuntimeException("Index does not exist");
    }
    // This is done to handle the scenario of stale cache because of which schema mismatch
    // exception can be thrown. Scenario: In case of carbondata used through FileFormat API,
    // once a table is dropped and recreated with the same name again then because the Index
    // contains the stale carbon table schema mismatch exception is thrown. To avoid such scenarios
    // it is always better to update the carbon table object retrieved
    index.getIndexFactory().setCarbonTable(table);
    index.getIndexFactory().setIndexSchema(indexSchema);
    index.setIndexSchema(indexSchema);
    return index;
  }

  private String getKeyUsingTablePath(String tablePath) {
    if (tablePath != null) {
      // Try get using table path
      for (Map.Entry<String, String> entry : tablePathMap.entrySet()) {
        if (new Path(entry.getValue()).equals(new Path(tablePath))) {
          return entry.getKey();
        }
      }
    }
    return null;
  }

  /**
   * Return a new Index instance and registered in the store manager.
   * The Index is created using Index name, Index factory class and table identifier.
   */
  public IndexFactory getIndexFactoryClass(CarbonTable table, IndexSchema indexSchema)
      throws MalformedIndexCommandException {
    try {
      // try to create Index by reflection to test whether it is a valid IndexFactory class
      return (IndexFactory)
          Class.forName(indexSchema.getProviderName()).getConstructors()[0]
              .newInstance(table, indexSchema);
    } catch (ClassNotFoundException e) {
      // try to create IndexClassProvider instance by taking providerName as short name
      return IndexRegistry.getIndexFactoryByShortName(table, indexSchema);
    } catch (Throwable e) {
      throw new MetadataProcessException(
          "failed to get Index factory for'" + indexSchema.getProviderName() + "'", e);
    }
  }

  /**
   * registered in the store manager.
   * The Index is created using Index name, Index factory class and table identifier.
   */
  private TableIndex createAndRegisterIndex(CarbonTable table,
      IndexSchema indexSchema) throws MalformedIndexCommandException {
    IndexFactory indexFactory = getIndexFactoryClass(table, indexSchema);
    return registerIndex(table, indexSchema, indexFactory);
  }

  public TableIndex registerIndex(CarbonTable table,
      IndexSchema indexSchema,  IndexFactory indexFactory) {
    String tableUniqueName = table.getCarbonTableIdentifier().getTableUniqueName();
    // Just update the segmentRefreshMap with the table if not added.
    getTableSegmentRefresher(table);
    List<TableIndex> tableIndices = allIndexes.get(table.getTableId());
    if (tableIndices == null) {
      String keyUsingTablePath = getKeyUsingTablePath(table.getTablePath());
      if (keyUsingTablePath != null) {
        tableUniqueName = keyUsingTablePath;
        tableIndices = allIndexes.get(table.getTableId());
      }
    }
    if (tableIndices == null) {
      tableIndices = new ArrayList<>();
    }

    BlockletDetailsFetcher blockletDetailsFetcher;
    SegmentPropertiesFetcher segmentPropertiesFetcher = null;
    if (indexFactory instanceof BlockletDetailsFetcher) {
      blockletDetailsFetcher = (BlockletDetailsFetcher) indexFactory;
    } else {
      blockletDetailsFetcher = getBlockletDetailsFetcher(table);
    }
    segmentPropertiesFetcher = (SegmentPropertiesFetcher) blockletDetailsFetcher;
    TableIndex index = new TableIndex(table, indexSchema, indexFactory, blockletDetailsFetcher,
        segmentPropertiesFetcher);

    tableIndices.add(index);
    allIndexes.put(table.getTableId(), tableIndices);
    tablePathMap.put(table.getTableId(), table.getTablePath());
    return index;
  }

  private TableIndex getTableIndex(String indexName, List<TableIndex> tableIndices) {
    TableIndex index = null;
    for (TableIndex tableIndex : tableIndices) {
      if (tableIndex.getIndexSchema().getIndexName().equals(indexName)) {
        index = tableIndex;
        break;
      }
    }
    return index;
  }

  /**
   * Clear the invalid segments from all the indexes of the table
   *
   * @param carbonTable table for which the operation has to be performed.
   * @param segments segments which have to be cleared from cache.
   */
  public void clearInvalidSegments(CarbonTable carbonTable, List<String> segments)
      throws IOException {
    getDefaultIndex(carbonTable).clear(segments);
    List<TableIndex> indexes = getAllCGAndFGIndexes(carbonTable);
    for (TableIndex index: indexes) {
      index.clear(segments);
    }

  }

  public List<String> getSegmentsToBeRefreshed(CarbonTable carbonTable,
      List<Segment> filteredSegmentToAccess) {
    List<String> toBeCleanedSegments = new ArrayList<>();
    for (Segment filteredSegment : filteredSegmentToAccess) {
      boolean refreshNeeded = getTableSegmentRefresher(carbonTable).isRefreshNeeded(filteredSegment,
          SegmentUpdateStatusManager
              .getInvalidTimestampRange(filteredSegment.getLoadMetadataDetails()));
      if (refreshNeeded) {
        toBeCleanedSegments.add(filteredSegment.getSegmentNo());
      }
    }
    return toBeCleanedSegments;
  }

  public void refreshSegmentCacheIfRequired(CarbonTable carbonTable,
      SegmentUpdateStatusManager updateStatusManager, List<Segment> filteredSegmentToAccess)
      throws IOException {
    List<String> toBeCleanedSegments =
        getSegmentsToBeRefreshed(carbonTable, filteredSegmentToAccess);
    if (toBeCleanedSegments.size() > 0) {
      clearInvalidSegments(carbonTable, toBeCleanedSegments);
    }
  }

  /**
   * Clear the index/indexes of a table from memory
   *
   * @param identifier Table identifier
   */
  public void clearIndex(AbsoluteTableIdentifier identifier) {
    CarbonTable carbonTable = getCarbonTable(identifier);
    boolean launchJob = false;
    try {
      // launchJob will be true if either the table has a CGIndex or index server is enabled for
      // the specified table.
      launchJob = hasCGIndex(carbonTable) ||
          CarbonProperties.getInstance().isDistributedPruningEnabled(identifier.getDatabaseName(),
              identifier.getTableName());
    } catch (IOException e) {
      LOGGER.warn("Unable to launch job to clear indexes.", e);
    }
    clearIndexCache(identifier, launchJob);
  }

  /**
   * Clear the index/indexes of a table from memory
   *
   * @param identifier Table identifier
   */
  public void clearIndexCache(AbsoluteTableIdentifier identifier, boolean clearInAllWorkers) {
    String tableId = identifier.getCarbonTableIdentifier().getTableId();
    if (clearInAllWorkers) {
      // carbon table need to lookup only if launch job is set.
      CarbonTable carbonTable = getCarbonTable(identifier);
      if (null != carbonTable) {
        String jobClassName;
        if (CarbonProperties.getInstance()
            .isDistributedPruningEnabled(identifier.getDatabaseName(), identifier.getTableName())) {
          jobClassName = IndexUtil.DISTRIBUTED_JOB_NAME;
        } else {
          jobClassName = IndexUtil.EMBEDDED_JOB_NAME;
        }
        try {
          IndexUtil.executeClearIndexJob(carbonTable, jobClassName);
        } catch (IOException e) {
          LOGGER.error("clear index job failed", e);
          // ignoring the exception
        }
      }
    } else {
      // remove carbon table from meta cache if launchJob is false as this would be called in
      // executor side.
      CarbonMetadata.getInstance()
          .removeTable(identifier.getDatabaseName(), identifier.getTableName());
    }
    List<TableIndex> tableIndices =
        allIndexes.get(identifier.getCarbonTableIdentifier().getTableId());
    if (tableIndices == null) {
      String keyUsingTablePath = getKeyUsingTablePath(identifier.getTablePath());
      if (keyUsingTablePath != null) {
        tableId = keyUsingTablePath;
      }
    }
    segmentRefreshMap.remove(tableId);
    clearIndex(tableId);
    allIndexes.remove(tableId);
    tablePathMap.remove(tableId);
  }

  /**
   * This method returns the carbonTable from identifier
   * @param identifier
   * @return
   */
  public CarbonTable getCarbonTable(AbsoluteTableIdentifier identifier) {
    CarbonTable carbonTable = null;
    carbonTable = CarbonMetadata.getInstance()
        .getCarbonTable(identifier.getDatabaseName(), identifier.getTableName());
    if (carbonTable == null) {
      try {
        carbonTable = CarbonTable
            .buildFromTablePath(identifier.getTableName(), identifier.getDatabaseName(),
                identifier.getTablePath(), identifier.getCarbonTableIdentifier().getTableId());
      } catch (IOException e) {
        LOGGER.warn("failed to get carbon table from table Path" + e.getMessage(), e);
        // ignoring exception
      }
    }
    return carbonTable;
  }

  /**
   * this methods clears the index of table from memory
   */
  public void clearIndex(String tableId) {
    List<TableIndex> tableIndices = allIndexes.get(tableId);
    if (tableIndices != null) {
      for (TableIndex tableIndex : tableIndices) {
        if (tableIndex != null) {
          // clear the segmentMap in BlockletDetailsFetcher,else the Segment will remain in executor
          // and the query fails as we will check whether the blocklet contains in the index or not
          tableIndex.getBlockletDetailsFetcher().clear();
          tableIndex.clear();
        }
      }
    }
    allIndexes.remove(tableId);
    tablePathMap.remove(tableId);
  }




  /**
   * Clear the index/indexes of a table from memory and disk
   */
  public void deleteIndex(CarbonTable carbonTable, String indexName) {
    if (carbonTable == null) {
      // If carbon table is null then it means table is already deleted, therefore return without
      // doing any further changes.
      return;
    }
    String tableId = carbonTable.getTableId();
    if (CarbonProperties.getInstance()
        .isDistributedPruningEnabled(carbonTable.getDatabaseName(), carbonTable.getTableName())) {
      try {
        IndexUtil
            .executeClearIndexJob(carbonTable, IndexUtil.DISTRIBUTED_JOB_NAME, indexName);
      } catch (IOException e) {
        LOGGER.error("clear index job failed", e);
        // ignoring the exception
      }
    } else {
      List<TableIndex> tableIndices = allIndexes.get(tableId);
      if (tableIndices != null) {
        int i = 0;
        for (TableIndex tableIndex : tableIndices) {
          if (tableIndex != null && indexName
              .equalsIgnoreCase(tableIndex.getIndexSchema().getIndexName())) {
            try {
              IndexUtil
                  .executeClearIndexJob(carbonTable, IndexUtil.EMBEDDED_JOB_NAME, indexName);
              tableIndex.clear();
            } catch (IOException e) {
              LOGGER.error("clear index job failed", e);
              // ignoring the exception
            }
            tableIndex.deleteIndexData();
            tableIndices.remove(i);
            break;
          }
          i++;
        }
        allIndexes.put(tableId, tableIndices);
      }
    }
  }

  /**
   * Get the blocklet index factory to get the detail information of blocklets
   *
   * @param table
   * @return
   */
  private BlockletDetailsFetcher getBlockletDetailsFetcher(CarbonTable table) {
    TableIndex index = getIndex(table, BlockletIndexFactory.INDEX_SCHEMA);
    return (BlockletDetailsFetcher) index.getIndexFactory();
  }

  /**
   * Returns the singleton instance
   *
   * @return
   */
  public static IndexStoreManager getInstance() {
    return instance;
  }

  /**
   * Get the TableSegmentRefresher for the table. If not existed then add one and return.
   */
  public TableSegmentRefresher getTableSegmentRefresher(CarbonTable table) {
    String tableId = table.getAbsoluteTableIdentifier().getCarbonTableIdentifier().getTableId();
    if (segmentRefreshMap.get(tableId) == null) {
      segmentRefreshMap.put(tableId, new TableSegmentRefresher(table));
    }
    return segmentRefreshMap.get(tableId);
  }

  /**
   * Keep track of the segment refresh time.
   */
  public static class TableSegmentRefresher {

    // This map stores the latest segment refresh time.So in case of update/delete we check the
    // time against this map.
    private Map<String, SegmentRefreshInfo> segmentRefreshTime = new HashMap<>();

    // This map keeps the manual refresh entries from users. It is mainly used for partition
    // altering.
    private Map<String, Boolean> manualSegmentRefresh = new HashMap<>();

    TableSegmentRefresher(CarbonTable table) {
      SegmentStatusManager segmentStatusManager =
          new SegmentStatusManager(table.getAbsoluteTableIdentifier(),
              table.getTableStatusVersion());
      List<Segment> validSegments;
      try {
        validSegments = segmentStatusManager.getValidAndInvalidSegments().getValidSegments();
      } catch (IOException e) {
        LOGGER.error("Error while getting the valid segments.", e);
        throw new RuntimeException(e);
      }
      for (Segment segment : validSegments) {
        UpdateVO updateVO =
            SegmentUpdateStatusManager.getInvalidTimestampRange(segment.getLoadMetadataDetails());
        SegmentRefreshInfo segmentRefreshInfo;
        String segmentFileName = segment.getSegmentFileName();
        if ((updateVO != null && updateVO.getLatestUpdateTimestamp() != null)
            || segmentFileName != null) {
          // Do not use getLastModifiedTime API on segment file carbon file object as it will
          // slow down operation in Object stores like S3. Now the segment file is always written
          // for operations which was overwriting earlier, so this timestamp can be checked always
          // to check whether to refresh the cache or not.
          long segmentFileTimeStamp = Long.parseLong(segmentFileName
              .substring(segmentFileName.indexOf(CarbonCommonConstants.UNDERSCORE) + 1,
                  segmentFileName.lastIndexOf(CarbonCommonConstants.POINT)));
          segmentRefreshInfo =
              new SegmentRefreshInfo(updateVO.getLatestUpdateTimestamp(), 0, segmentFileTimeStamp);
        } else {
          segmentRefreshInfo = new SegmentRefreshInfo(0L, 0, 0L);
        }
        segmentRefreshTime.put(segment.getSegmentNo(), segmentRefreshInfo);
      }
    }

    public boolean isRefreshNeeded(Segment seg, UpdateVO updateVo) {
      SegmentRefreshInfo segmentRefreshInfo =
          seg.getSegmentRefreshInfo(updateVo);
      String segmentId = seg.getSegmentNo();
      if (segmentRefreshInfo.getSegmentUpdatedTimestamp() == null
          && segmentRefreshInfo.getSegmentFileTimestamp() == 0) {
        return false;
      }

      if (segmentRefreshTime.get(segmentId) == null) {
        if (segmentRefreshInfo.getSegmentUpdatedTimestamp() != null
            && segmentRefreshInfo.getSegmentUpdatedTimestamp() != 0) {
          segmentRefreshTime.put(segmentId, segmentRefreshInfo);
          return true;
        }
        if (segmentRefreshInfo.getSegmentFileTimestamp() != 0) {
          segmentRefreshTime.put(segmentId, segmentRefreshInfo);
          return true;
        }
      }

      if (manualSegmentRefresh.get(segmentId) != null && manualSegmentRefresh.get(segmentId)) {
        manualSegmentRefresh.put(segmentId, false);
        return true;
      }

      boolean isRefresh = segmentRefreshInfo.compare(segmentRefreshTime.get(segmentId));
      if (isRefresh) {
        segmentRefreshTime.remove(segmentId);
      }
      return isRefresh;
    }
  }

  public synchronized void clearInvalidIndex(CarbonTable carbonTable, List<String> segmentNos,
      String indexToClear) throws IOException {
    List<TableIndex> indexes = getAllCGAndFGIndexes(carbonTable);
    List<TableIndex> remainingIndexes = new ArrayList<>();
    if (StringUtils.isNotEmpty(indexToClear)) {
      for (TableIndex tableIndex : indexes) {
        if (indexToClear.equalsIgnoreCase(tableIndex.getIndexSchema().getIndexName())) {
          for (String segment : segmentNos) {
            tableIndex.deleteSegmentIndexData(segment);
          }
          tableIndex.clear();
        } else {
          remainingIndexes.add(tableIndex);
        }
      }
      allIndexes.put(carbonTable.getTableId(), remainingIndexes);
    } else {
      clearIndex(carbonTable.getTableId());
      // clear the segment properties cache from executor
      SegmentPropertiesAndSchemaHolder.getInstance()
          .invalidate(carbonTable.getAbsoluteTableIdentifier());
    }
  }

  private boolean hasCGIndex(CarbonTable carbonTable) throws IOException {
    // In case of spark file format flow, carbon table will be null
    if (null == carbonTable) {
      return false;
    }
    for (TableIndex tableIndex : carbonTable.getAllVisibleIndexes()) {
      if (tableIndex.getIndexFactory().getIndexLevel().equals(IndexLevel.CG)) {
        return true;
      }
    }
    return false;
  }

}