content: /*
   * 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.util.ArrayList;
  import java.util.Arrays;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  
  import org.apache.carbondata.common.annotations.InterfaceAudience;
  import org.apache.carbondata.common.logging.LogServiceFactory;
  import org.apache.carbondata.core.cache.Cache;
  import org.apache.carbondata.core.constants.CarbonCommonConstants;
  import org.apache.carbondata.core.datastore.block.SegmentProperties;
  import org.apache.carbondata.core.datastore.impl.FileFactory;
  import org.apache.carbondata.core.datastore.page.encoding.bool.BooleanConvert;
  import org.apache.carbondata.core.index.dev.IndexModel;
  import org.apache.carbondata.core.index.dev.cgindex.CoarseGrainIndex;
  import org.apache.carbondata.core.indexstore.Blocklet;
  import org.apache.carbondata.core.metadata.datatype.DataTypes;
  import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
  import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
  import org.apache.carbondata.core.scan.expression.ColumnExpression;
  import org.apache.carbondata.core.scan.expression.Expression;
  import org.apache.carbondata.core.scan.expression.LiteralExpression;
  import org.apache.carbondata.core.scan.expression.conditional.EqualToExpression;
  import org.apache.carbondata.core.scan.expression.conditional.InExpression;
  import org.apache.carbondata.core.scan.expression.conditional.ListExpression;
  import org.apache.carbondata.core.scan.expression.exception.FilterIllegalMemberException;
  import org.apache.carbondata.core.scan.expression.logical.AndExpression;
  import org.apache.carbondata.core.scan.filter.executer.FilterExecutor;
  import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
  import org.apache.carbondata.core.util.CarbonProperties;
  import org.apache.carbondata.core.util.CarbonUtil;
  import org.apache.carbondata.core.util.DataTypeUtil;
  import org.apache.carbondata.processing.loading.DataField;
  import org.apache.carbondata.processing.loading.converter.BadRecordLogHolder;
  import org.apache.carbondata.processing.loading.converter.FieldConverter;
  import org.apache.carbondata.processing.loading.converter.impl.FieldEncoderFactory;
  
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.util.bloom.CarbonBloomFilter;
  import org.apache.hadoop.util.bloom.Key;
  import org.apache.log4j.Logger;
  
  /**
   * BloomDataCoarseGrainIndex is constructed in blocklet level. For each indexed column,
   * a bloom filter is constructed to indicate whether a value belongs to this blocklet.
   * More information of the index file can be found in the corresponding index writer.
   */
  @InterfaceAudience.Internal
  public class BloomCoarseGrainIndex extends CoarseGrainIndex {
    private static final Logger LOGGER =
        LogServiceFactory.getLogService(BloomCoarseGrainIndex.class.getName());
    private Map<String, CarbonColumn> name2Col;
    private Cache<BloomCacheKeyValue.CacheKey, BloomCacheKeyValue.CacheValue> cache;
    private String shardName;
    private Path indexPath;
    private Set<String> filteredShard;
    private boolean needShardPrune;
    /**
     * This is used to convert literal filter value to internal carbon value
     */
    private Map<String, FieldConverter> name2Converters;
    private BadRecordLogHolder badRecordLogHolder;
  
    @Override
    public void init(IndexModel indexModel) {
      this.indexPath = FileFactory.getPath(indexModel.getFilePath());
      this.shardName = indexPath.getName();
      if (indexModel instanceof BloomIndexModel) {
        BloomIndexModel model = (BloomIndexModel) indexModel;
        this.cache = model.getCache();
      }
    }
  
    public void setFilteredShard(Set<String> filteredShard) {
      this.filteredShard = filteredShard;
      // do shard prune when pruning only if bloom index files are merged
      this.needShardPrune = filteredShard != null &&
              shardName.equals(BloomIndexFileStore.MERGE_BLOOM_INDEX_SHARD_NAME);
    }
  
    /**
     * init field converters for index columns
     */
    public void initIndexColumnConverters(CarbonTable carbonTable, List<CarbonColumn> indexedColumn) {
      this.name2Col = new HashMap<>(indexedColumn.size());
      for (CarbonColumn col : indexedColumn) {
        this.name2Col.put(col.getColName(), col);
      }
  
      this.name2Converters = new HashMap<>(indexedColumn.size());
      String nullFormat = "\\N";
  
      for (int i = 0; i < indexedColumn.size(); i++) {
        DataField dataField = new DataField(indexedColumn.get(i));
        String dateFormat = CarbonProperties.getInstance().getProperty(
            CarbonCommonConstants.CARBON_DATE_FORMAT,
            CarbonCommonConstants.CARBON_DATE_DEFAULT_FORMAT);
        dataField.setDateFormat(dateFormat);
        String tsFormat = CarbonProperties.getInstance().getProperty(
            CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT,
            CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT);
        dataField.setTimestampFormat(tsFormat);
        FieldConverter fieldConverter = FieldEncoderFactory.getInstance()
            .createFieldEncoder(dataField, i, nullFormat, false,
                false, carbonTable.getTablePath(), null);
        this.name2Converters.put(indexedColumn.get(i).getColName(), fieldConverter);
      }
      this.badRecordLogHolder = new BadRecordLogHolder();
      this.badRecordLogHolder.setLogged(false);
    }
  
    @Override
    public List<Blocklet> prune(FilterResolverIntf filterExp, SegmentProperties segmentProperties,
        FilterExecutor filterExecutor, CarbonTable carbonTable) {
      Set<Blocklet> hitBlocklets = null;
      if (filterExp == null) {
        // null is different from empty here. Empty means after pruning, no blocklet need to scan.
        return null;
      }
      if (filteredShard.isEmpty()) {
        LOGGER.info("Bloom filtered shards is empty");
        return new ArrayList<>();
      }
  
      List<BloomQueryModel> bloomQueryModels;
      bloomQueryModels = createQueryModel(filterExp.getFilterExpression());
      for (BloomQueryModel bloomQueryModel : bloomQueryModels) {
        Set<Blocklet> tempHitBlockletsResult = new HashSet<>();
        if (LOGGER.isDebugEnabled()) {
          LOGGER.debug("prune blocklet for query: " + bloomQueryModel);
        }
        Long expiration_time = CarbonUtil.getExpiration_time(carbonTable);
        BloomCacheKeyValue.CacheKey cacheKey =
            new BloomCacheKeyValue.CacheKey(this.indexPath.toString(), bloomQueryModel.columnName,
                expiration_time);
        BloomCacheKeyValue.CacheValue cacheValue = cache.get(cacheKey);
        List<CarbonBloomFilter> bloomIndexList = cacheValue.getBloomFilters();
        for (CarbonBloomFilter bloomFilter : bloomIndexList) {
          if (needShardPrune && !filteredShard.contains(bloomFilter.getShardName())) {
            // skip shard which has been pruned in Main index
            continue;
          }
          boolean scanRequired = false;
          for (byte[] value: bloomQueryModel.filterValues) {
            scanRequired = bloomFilter.membershipTest(new Key(value));
            if (scanRequired) {
              // if any filter value hit this bloomfilter
              // no need to check other filter values
              break;
            }
          }
          if (scanRequired) {
            if (LOGGER.isDebugEnabled()) {
              LOGGER.debug(String.format("BloomCoarseGrainIndex: Need to scan -> blocklet#%s",
                  String.valueOf(bloomFilter.getBlockletNo())));
            }
            Blocklet blocklet = new Blocklet(bloomFilter.getShardName(),
                String.valueOf(bloomFilter.getBlockletNo()));
            tempHitBlockletsResult.add(blocklet);
          } else if (LOGGER.isDebugEnabled()) {
            LOGGER.debug(String.format("BloomCoarseGrainIndex: Skip scan -> blocklet#%s",
                String.valueOf(bloomFilter.getBlockletNo())));
          }
          // get intersect result between query models
          // pre-condition: only And/In/EqualTo expression exists in single bloom index
          if (null == hitBlocklets) {
            hitBlocklets = tempHitBlockletsResult;
          } else {
            hitBlocklets.retainAll(tempHitBlockletsResult);
          }
        }
      }
      if (hitBlocklets == null) {
        LOGGER.warn(String.format("HitBlocklets is empty in bloom filter prune method. " +
                "bloomQueryModels size is %d, filterShards size if %d",
                bloomQueryModels.size(), filteredShard.size()));
        return new ArrayList<>();
      }
      return new ArrayList<>(hitBlocklets);
    }
  
    private List<BloomQueryModel> createQueryModel(Expression expression) {
      List<BloomQueryModel> queryModels = new ArrayList<BloomQueryModel>();
      // bloom index only support equalTo and In operators now
      if (expression instanceof EqualToExpression) {
        Expression left = ((EqualToExpression) expression).getLeft();
        Expression right = ((EqualToExpression) expression).getRight();
        String column;
        if (left instanceof ColumnExpression && right instanceof LiteralExpression) {
          column = ((ColumnExpression) left).getColumnName();
          if (this.name2Col.containsKey(column)) {
            BloomQueryModel bloomQueryModel =
                buildQueryModelForEqual((ColumnExpression) left, (LiteralExpression) right);
            queryModels.add(bloomQueryModel);
          }
          return queryModels;
        } else if (left instanceof LiteralExpression && right instanceof ColumnExpression) {
          column = ((ColumnExpression) right).getColumnName();
          if (this.name2Col.containsKey(column)) {
            BloomQueryModel bloomQueryModel =
                buildQueryModelForEqual((ColumnExpression) right, (LiteralExpression) left);
            queryModels.add(bloomQueryModel);
          }
          return queryModels;
        } else {
          String errorMsg = "BloomFilter can only support the 'equal' filter like 'Col = PlainValue'";
          LOGGER.warn(errorMsg);
          throw new RuntimeException(errorMsg);
        }
      } else if (expression instanceof InExpression) {
        Expression left = ((InExpression) expression).getLeft();
        Expression right = ((InExpression) expression).getRight();
        String column;
        if (left instanceof ColumnExpression && right instanceof ListExpression) {
          column = ((ColumnExpression) left).getColumnName();
          if (this.name2Col.containsKey(column)) {
            BloomQueryModel bloomQueryModel =
                buildQueryModelForIn((ColumnExpression) left, (ListExpression) right);
            queryModels.add(bloomQueryModel);
          }
          return queryModels;
        } else if (left instanceof ListExpression && right instanceof ColumnExpression) {
          column = ((ColumnExpression) right).getColumnName();
          if (this.name2Col.containsKey(column)) {
            BloomQueryModel bloomQueryModel =
                buildQueryModelForIn((ColumnExpression) right, (ListExpression) left);
            queryModels.add(bloomQueryModel);
          }
          return queryModels;
        } else {
          String errorMsg = "BloomFilter can only support the 'in' filter like 'Col in PlainValue'";
          LOGGER.warn(errorMsg);
          throw new RuntimeException(errorMsg);
        }
      }  else if (expression instanceof AndExpression) {
        queryModels.addAll(createQueryModel(((AndExpression) expression).getLeft()));
        queryModels.addAll(createQueryModel(((AndExpression) expression).getRight()));
        return queryModels;
      }
  
      return queryModels;
    }
  
    private BloomQueryModel buildQueryModelForEqual(ColumnExpression ce, LiteralExpression le) {
      List<byte[]> filterValues = new ArrayList<>();
      byte[] internalFilterValue = getInternalFilterValue(this.name2Col.get(ce.getColumnName()), le);
      filterValues.add(internalFilterValue);
      return new BloomQueryModel(ce.getColumnName(), filterValues);
    }
  
    /**
     * Note that `in` operator needs at least one match not exactly match. since while doing pruning,
     * we collect all the blocklets that will match the querymodel, this will not be a problem.
     */
    private BloomQueryModel buildQueryModelForIn(ColumnExpression ce, ListExpression le) {
      List<byte[]> filterValues = new ArrayList<>();
      for (Expression child : le.getChildren()) {
        byte[] internalFilterValue = getInternalFilterValue(
            this.name2Col.get(ce.getColumnName()), (LiteralExpression) child);
        filterValues.add(internalFilterValue);
      }
      return new BloomQueryModel(ce.getColumnName(), filterValues);
    }
  
    private byte[] getInternalFilterValue(CarbonColumn carbonColumn, LiteralExpression le) {
      // convert the filter value to string and apply converters on it to get carbon internal value
      String strFilterValue = null;
      try {
        strFilterValue = le.getExpressionResult().getString();
      } catch (FilterIllegalMemberException e) {
        throw new RuntimeException("Error while resolving filter expression", e);
      }
  
      Object convertedValue = this.name2Converters.get(carbonColumn.getColName()).convert(
          strFilterValue, badRecordLogHolder);
  
      byte[] internalFilterValue;
      if (carbonColumn.isMeasure()) {
        // for measures, the value is already the type, just convert it to bytes.
        if (convertedValue == null) {
          convertedValue = DataConvertUtil.getNullValueForMeasure(carbonColumn.getDataType(),
              carbonColumn.getColumnSchema().getScale());
        }
        // Carbon stores boolean as byte. Here we convert it for `getValueAsBytes`
        if (carbonColumn.getDataType().equals(DataTypes.BOOLEAN)) {
          convertedValue = BooleanConvert.boolean2Byte((Boolean)convertedValue);
        }
        internalFilterValue = CarbonUtil.getValueAsBytes(carbonColumn.getDataType(), convertedValue);
      } else if (carbonColumn.getDataType() == DataTypes.DATE) {
        // for dictionary/date columns, convert the surrogate key to bytes
        internalFilterValue = CarbonUtil.getValueAsBytes(DataTypes.INT, convertedValue);
      } else {
        // for non dictionary dimensions, numeric columns will be of original data,
        // so convert the data to bytes
        if (DataTypeUtil.isPrimitiveColumn(carbonColumn.getDataType())) {
          if (convertedValue == null) {
            convertedValue = DataConvertUtil.getNullValueForMeasure(carbonColumn.getDataType(),
                carbonColumn.getColumnSchema().getScale());
          }
          internalFilterValue =
              CarbonUtil.getValueAsBytes(carbonColumn.getDataType(), convertedValue);
        } else {
          internalFilterValue = (byte[]) convertedValue;
        }
      }
      if (internalFilterValue.length == 0) {
        internalFilterValue = CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY;
      }
      return internalFilterValue;
    }
  
    @Override
    public boolean isScanRequired(FilterResolverIntf filterExp) {
      return true;
    }
  
    @Override
    public void clear() {
    }
  
    static class BloomQueryModel {
      private String columnName;
      private List<byte[]> filterValues;
  
      /**
       * represent an query model will be applyied on bloom index
       *
       * @param columnName bloom index column
       * @param filterValues key for the bloom index,
       *                   this value is converted from user specified filter value in query
       */
      private BloomQueryModel(String columnName, List<byte[]> filterValues) {
        this.columnName = columnName;
        this.filterValues = filterValues;
      }
  
      @Override
      public String toString() {
        final StringBuilder sb = new StringBuilder("BloomQueryModel{");
        sb.append("columnName='").append(columnName).append('\'');
        sb.append(", filterValues=");
        for (byte[] value: filterValues) {
          sb.append(Arrays.toString(value));
        }
        sb.append('}');
        return sb.toString();
      }
    }
  
    @Override
    public void finish() {
  
    }
  }
  