/*
 * 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.index.bloom;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.carbondata.common.annotations.InterfaceAudience;
import org.apache.carbondata.common.exceptions.sql.MalformedIndexCommandException;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.cache.Cache;
import org.apache.carbondata.core.cache.CacheProvider;
import org.apache.carbondata.core.cache.CacheType;
import org.apache.carbondata.core.constants.CarbonV3DataFormatConstants;
import org.apache.carbondata.core.datastore.block.SegmentProperties;
import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.features.TableOperation;
import org.apache.carbondata.core.index.IndexInputSplit;
import org.apache.carbondata.core.index.IndexLevel;
import org.apache.carbondata.core.index.IndexMeta;
import org.apache.carbondata.core.index.Segment;
import org.apache.carbondata.core.index.dev.IndexBuilder;
import org.apache.carbondata.core.index.dev.IndexFactory;
import org.apache.carbondata.core.index.dev.IndexWriter;
import org.apache.carbondata.core.index.dev.cgindex.CoarseGrainIndex;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.IndexSchema;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
import org.apache.carbondata.core.scan.filter.intf.ExpressionType;
import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.path.CarbonTablePath;
import org.apache.carbondata.events.Event;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;

/**
 * This class is for Bloom Filter for blocklet level
 */
@InterfaceAudience.Internal
public class BloomCoarseGrainIndexFactory extends IndexFactory<CoarseGrainIndex> {
  private static final Logger LOGGER = LogServiceFactory.getLogService(
      BloomCoarseGrainIndexFactory.class.getName());
  /**
   * property for size of bloom filter
   */
  private static final String BLOOM_SIZE = "bloom_size";
  /**
   * default size for bloom filter, cardinality of the column.
   */
  private static final int DEFAULT_BLOOM_FILTER_SIZE =
      CarbonV3DataFormatConstants.NUMBER_OF_ROWS_PER_BLOCKLET_COLUMN_PAGE_DEFAULT * 20;
  /**
   * property for fpp(false-positive-probability) of bloom filter
   */
  private static final String BLOOM_FPP = "bloom_fpp";
  /**
   * default value for fpp of bloom filter is 0.001%
   */
  private static final double DEFAULT_BLOOM_FILTER_FPP = 0.00001d;

  /**
   * property for compressing bloom while saving to disk.
   */
  private static final String COMPRESS_BLOOM = "bloom_compress";
  /**
   * Default value of compressing bloom while save to disk.
   */
  private static final boolean DEFAULT_BLOOM_COMPRESS = true;

  private IndexMeta indexMeta;
  private String indexName;
  private int bloomFilterSize;
  private double bloomFilterFpp;
  private boolean bloomCompress;
  private Cache<BloomCacheKeyValue.CacheKey, BloomCacheKeyValue.CacheValue> cache;
  // segmentId -> list of index file
  private Map<String, Set<String>> segmentMap = new ConcurrentHashMap<>();

  public BloomCoarseGrainIndexFactory(CarbonTable carbonTable, IndexSchema indexSchema)
      throws MalformedIndexCommandException {
    super(carbonTable, indexSchema);
    Objects.requireNonNull(carbonTable);
    Objects.requireNonNull(indexSchema);

    this.indexName = indexSchema.getIndexName();

    List<CarbonColumn> indexedColumns =
        carbonTable.getIndexedColumns(indexSchema.getIndexColumns());
    this.bloomFilterSize = validateAndGetBloomFilterSize(indexSchema);
    this.bloomFilterFpp = validateAndGetBloomFilterFpp(indexSchema);
    this.bloomCompress = validateAndGetBloomCompress(indexSchema);
    List<ExpressionType> optimizedOperations = new ArrayList<ExpressionType>();
    // todo: support more optimize operations
    optimizedOperations.add(ExpressionType.EQUALS);
    optimizedOperations.add(ExpressionType.IN);
    this.indexMeta = new IndexMeta(this.indexName, indexedColumns, optimizedOperations);
    LOGGER.info(String.format("Index %s works for %s with bloom size %d",
        this.indexName, this.indexMeta, this.bloomFilterSize));
    try {
      this.cache = CacheProvider.getInstance()
          .createCache(new CacheType("bloom_cache"), BloomIndexCache.class.getName());
    } catch (Exception e) {
      LOGGER.error(e.getMessage(), e);
      throw new MalformedIndexCommandException(e.getMessage());
    }
  }

  /**
   * validate Lucene Index BLOOM_SIZE
   * 1. BLOOM_SIZE property is optional, 32000 * 20 will be the default size.
   * 2. BLOOM_SIZE should be an integer that greater than 0
   */
  private int validateAndGetBloomFilterSize(IndexSchema dmSchema)
      throws MalformedIndexCommandException {
    String bloomFilterSizeStr = dmSchema.getProperties().get(BLOOM_SIZE);
    if (StringUtils.isBlank(bloomFilterSizeStr)) {
      LOGGER.warn(
          String.format("Bloom filter size is not configured for index %s, use default value %d",
              indexName, DEFAULT_BLOOM_FILTER_SIZE));
      return DEFAULT_BLOOM_FILTER_SIZE;
    }
    int bloomFilterSize;
    try {
      bloomFilterSize = Integer.parseInt(bloomFilterSizeStr);
    } catch (NumberFormatException e) {
      throw new MalformedIndexCommandException(
          String.format("Invalid value of bloom filter size '%s', it should be an integer",
              bloomFilterSizeStr));
    }
    // todo: reconsider the boundaries of bloom filter size
    if (bloomFilterSize <= 0) {
      throw new MalformedIndexCommandException(
          String.format("Invalid value of bloom filter size '%s', it should be greater than 0",
              bloomFilterSizeStr));
    }
    return bloomFilterSize;
  }

  /**
   * validate bloom Index BLOOM_FPP
   * 1. BLOOM_FPP property is optional, 0.00001 will be the default value.
   * 2. BLOOM_FPP should be (0, 1)
   */
  private double validateAndGetBloomFilterFpp(IndexSchema dmSchema)
      throws MalformedIndexCommandException {
    String bloomFilterFppStr = dmSchema.getProperties().get(BLOOM_FPP);
    if (StringUtils.isBlank(bloomFilterFppStr)) {
      LOGGER.warn(
          String.format("Bloom filter FPP is not configured for index %s, use default value %f",
              indexName, DEFAULT_BLOOM_FILTER_FPP));
      return DEFAULT_BLOOM_FILTER_FPP;
    }
    double bloomFilterFpp;
    try {
      bloomFilterFpp = Double.parseDouble(bloomFilterFppStr);
    } catch (NumberFormatException e) {
      throw new MalformedIndexCommandException(
          String.format("Invalid value of bloom filter fpp '%s', it should be an numeric",
              bloomFilterFppStr));
    }
    if (bloomFilterFpp < 0 || bloomFilterFpp - 1 >= 0) {
      throw new MalformedIndexCommandException(
          String.format("Invalid value of bloom filter fpp '%s', it should be in range 0~1",
              bloomFilterFppStr));
    }
    return bloomFilterFpp;
  }

  /**
   * validate bloom Index COMPRESS_BLOOM
   * Default value is true
   */
  private boolean validateAndGetBloomCompress(IndexSchema dmSchema) {
    String bloomCompress = dmSchema.getProperties().get(COMPRESS_BLOOM);
    if (StringUtils.isBlank(bloomCompress)) {
      LOGGER.warn(
          String.format("Bloom compress is not configured for index %s, use default value %b",
              indexName, DEFAULT_BLOOM_COMPRESS));
      return DEFAULT_BLOOM_COMPRESS;
    }
    return Boolean.parseBoolean(bloomCompress);
  }

  @Override
  public IndexWriter createWriter(Segment segment, String shardName,
      SegmentProperties segmentProperties) throws IOException {
    LOGGER.info(
        String.format("Data of BloomCoarseGrainIndex %s for table %s will be written to %s",
            this.indexName, getCarbonTable().getTableName() , shardName));
    return new BloomIndexWriter(getCarbonTable().getTablePath(), this.indexName,
        this.indexMeta.getIndexedColumns(), segment, shardName, this.bloomFilterSize,
        this.bloomFilterFpp, bloomCompress);
  }

  @Override
  public IndexBuilder createBuilder(Segment segment, String shardName,
      SegmentProperties segmentProperties) throws IOException {
    return new BloomIndexBuilder(getCarbonTable().getTablePath(), this.indexName,
        this.indexMeta.getIndexedColumns(), segment, shardName, this.bloomFilterSize,
        this.bloomFilterFpp, bloomCompress);
  }

  /**
   * returns all shard directories of bloom index files for query
   * if bloom index files are merged we should get only one shard path
   */
  public static Set<String> getAllShardPaths(String tablePath, String segmentId, String indexName) {
    String indexStorePath = CarbonTablePath.getIndexesStorePath(
        tablePath, segmentId, indexName);
    CarbonFile[] carbonFiles = FileFactory.getCarbonFile(indexStorePath).listFiles();
    Set<String> shardPaths = new HashSet<>();
    boolean mergeShardInProgress = false;
    CarbonFile mergeShardFile = null;
    for (CarbonFile carbonFile : carbonFiles) {
      if (carbonFile.getName().equals(BloomIndexFileStore.MERGE_BLOOM_INDEX_SHARD_NAME)) {
        mergeShardFile = carbonFile;
      } else if (carbonFile.getName().equals(BloomIndexFileStore.MERGE_INPROGRESS_FILE)) {
        mergeShardInProgress = true;
      } else if (carbonFile.isDirectory()) {
        shardPaths.add(FileFactory.getPath(carbonFile.getAbsolutePath()).toString());
      }
    }
    if (mergeShardFile != null && !mergeShardInProgress) {
      // should only get one shard path if mergeShard is generated successfully
      shardPaths.clear();
      shardPaths.add(FileFactory.getPath(mergeShardFile.getAbsolutePath()).toString());
    }
    return shardPaths;
  }

  @Override
  public List<CoarseGrainIndex> getIndexes(Segment segment) throws IOException {
    List<CoarseGrainIndex> indexes = new ArrayList<>();
    try {
      Set<String> shardPaths = segmentMap.get(segment.getSegmentNo());
      if (shardPaths == null) {
        shardPaths =
            getAllShardPaths(getCarbonTable().getTablePath(), segment.getSegmentNo(), indexName);
        segmentMap.put(segment.getSegmentNo(), shardPaths);
      }
      Set<String> filteredShards = segment.getFilteredIndexShardNames();
      for (String shard : shardPaths) {
        if (shard.endsWith(BloomIndexFileStore.MERGE_BLOOM_INDEX_SHARD_NAME) ||
            filteredShards.contains(new File(shard).getName())) {
          // Filter out the tasks which are filtered through Main index.
          // for merge shard, shard pruning delay to be done before pruning blocklet
          BloomCoarseGrainIndex bloomDM = new BloomCoarseGrainIndex();
          bloomDM.init(new BloomIndexModel(shard, cache, segment.getConfiguration()));
          bloomDM.initIndexColumnConverters(getCarbonTable(), indexMeta.getIndexedColumns());
          bloomDM.setFilteredShard(filteredShards);
          indexes.add(bloomDM);
        }
      }
    } catch (Exception e) {
      throw new IOException("Error occurs while init Bloom Index", e);
    }
    return indexes;
  }

  @Override
  public List<CoarseGrainIndex> getIndexes(Segment segment, Set<Path> partitionLocations)
      throws IOException {
    return getIndexes(segment);
  }

  @Override
  public List<CoarseGrainIndex> getIndexes(IndexInputSplit distributable) {
    List<CoarseGrainIndex> indexes = new ArrayList<>();
    String indexPath = ((BloomIndexInputSplit) distributable).getIndexPath();
    Set<String> filteredShards = ((BloomIndexInputSplit) distributable).getFilteredShards();
    BloomCoarseGrainIndex bloomDM = new BloomCoarseGrainIndex();
    bloomDM.init(new BloomIndexModel(indexPath, cache, FileFactory.getConfiguration()));
    bloomDM.initIndexColumnConverters(getCarbonTable(), indexMeta.getIndexedColumns());
    bloomDM.setFilteredShard(filteredShards);
    indexes.add(bloomDM);
    return indexes;
  }

  @Override
  public List<IndexInputSplit> toDistributable(Segment segment) {
    List<IndexInputSplit> indexInputSplitList = new ArrayList<>();
    Set<String> shardPaths = segmentMap.get(segment.getSegmentNo());
    if (shardPaths == null) {
      shardPaths =
          getAllShardPaths(getCarbonTable().getTablePath(), segment.getSegmentNo(), indexName);
      segmentMap.put(segment.getSegmentNo(), shardPaths);
    }
    Set<String> filteredShards = segment.getFilteredIndexShardNames();
    for (String shardPath : shardPaths) {
      // Filter out the tasks which are filtered through Main index.
      // for merge shard, shard pruning delay to be done before pruning blocklet
      if (shardPath.endsWith(BloomIndexFileStore.MERGE_BLOOM_INDEX_SHARD_NAME) ||
          filteredShards.contains(new File(shardPath).getName())) {
        IndexInputSplit bloomIndexInputSplit =
            new BloomIndexInputSplit(shardPath, filteredShards);
        bloomIndexInputSplit.setSegment(segment);
        bloomIndexInputSplit.setIndexSchema(getIndexSchema());
        indexInputSplitList.add(bloomIndexInputSplit);
      }
    }
    return indexInputSplitList;
  }

  @Override
  public void fireEvent(Event event) {

  }

  @Override
  public void clear(String segment) {
    Set<String> shards = segmentMap.remove(segment);
    if (shards != null) {
      for (String shard : shards) {
        for (CarbonColumn carbonColumn : indexMeta.getIndexedColumns()) {
          cache.invalidate(new BloomCacheKeyValue.CacheKey(shard, carbonColumn.getColName()));
        }
      }
    }
  }

  @Override
  public synchronized void clear() {
    if (segmentMap.size() > 0) {
      List<String> segments = new ArrayList<>(segmentMap.keySet());
      for (String segmentId : segments) {
        clear(segmentId);
      }
    }
  }

  @Override
  public void deleteIndexData(Segment segment) throws IOException {
    deleteSegmentIndexData(segment.getSegmentNo());
  }

  @Override
  public void deleteSegmentIndexData(String segmentId) throws IOException {
    try {
      String indexPath = CarbonTablePath
          .getIndexesStorePath(getCarbonTable().getTablePath(), segmentId, indexName);
      if (FileFactory.isFileExist(indexPath)) {
        CarbonFile file = FileFactory.getCarbonFile(indexPath);
        CarbonUtil.deleteFoldersAndFilesSilent(file);
      }
      clear(segmentId);
    } catch (InterruptedException ex) {
      throw new IOException("Failed to delete index for segment_" + segmentId);
    }
  }

  @Override
  public void deleteIndexData() {
    SegmentStatusManager ssm =
        new SegmentStatusManager(getCarbonTable().getAbsoluteTableIdentifier());
    try {
      List<Segment> validSegments =
          ssm.getValidAndInvalidSegments(getCarbonTable().isMV()).getValidSegments();
      for (Segment segment : validSegments) {
        deleteIndexData(segment);
      }
    } catch (IOException e) {
      LOGGER.error("drop index failed, failed to delete index directory");
    }
  }

  @Override
  public boolean willBecomeStale(TableOperation operation) {
    switch (operation) {
      case ALTER_RENAME:
        return false;
      case ALTER_DROP:
        return true;
      case ALTER_ADD_COLUMN:
        return false;
      case ALTER_CHANGE_DATATYPE:
        return true;
      case ALTER_COLUMN_RENAME:
        return true;
      case STREAMING:
        return false;
      case DELETE:
        return true;
      case UPDATE:
        return true;
      case PARTITION:
        return true;
      default:
        return false;
    }
  }

  @Override
  public boolean isOperationBlocked(TableOperation operation, Object... targets) {
    switch (operation) {
      case ALTER_DROP: {
        // alter table drop columns
        // will be blocked if the columns in bloomfilter index
        List<String> columnsToDrop = (List<String>) targets[0];
        List<String> indexedColumnNames = indexMeta.getIndexedColumnNames();
        for (String indexedcolumn : indexedColumnNames) {
          for (String column : columnsToDrop) {
            if (column.equalsIgnoreCase(indexedcolumn)) {
              return true;
            }
          }
        }
        return false;
      }
      case ALTER_CHANGE_DATATYPE:
      case ALTER_COLUMN_RENAME: {
        // alter table change one column datatype, or rename
        // will be blocked if the column in bloomfilter index
        String columnToChangeDatatype = (String) targets[0];
        List<String> indexedColumnNames = indexMeta.getIndexedColumnNames();
        for (String indexedcolumn : indexedColumnNames) {
          if (indexedcolumn.equalsIgnoreCase(columnToChangeDatatype)) {
            return true;
          }
        }
        return false;
      }
      default:
        return false;
    }
  }

  @Override
  public IndexMeta getMeta() {
    return this.indexMeta;
  }

  @Override
  public IndexLevel getIndexLevel() {
    return IndexLevel.CG;
  }

  @Override
  public String getCacheSize() {
    long sum = 0L;
    for (Map.Entry<String, Set<String>> entry : segmentMap.entrySet()) {
      for (String shardName : entry.getValue()) {
        for (CarbonColumn carbonColumn : indexMeta.getIndexedColumns()) {
          BloomCacheKeyValue.CacheValue cacheValue = cache
              .getIfPresent(new BloomCacheKeyValue.CacheKey(shardName, carbonColumn.getColName()));
          if (cacheValue != null) {
            sum += cacheValue.getMemorySize();
          }
        }
      }
    } return 0 + ":" + sum;
  }
}