2259

Build # 2259

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/b4f4607bda4e734b9a8ea584c12d4de5acb74fff">b4f4607bd<a href="https://github.com/apache/carbondata/commit/b4f4607bda4e734b9a8ea584c12d4de5acb74fff">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/ed1a96f2874a3d6a2e75c855511a4ddcfcf6308d">ed1a96f28</a>">ed1a96f28</a><a href="https://github.com/apache/carbondata/commit/b4f4607bda4e734b9a8ea584c12d4de5acb74fff"> into d5a2c698c">d5a2c698c</a>

Run Details

7 of 7 new or added lines in 2 files covered. (100.0%)

61547 of 77964 relevant lines covered (78.94%)

1.05 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.43

/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMap.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.datamap.bloom;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TimeZone;
import java.util.concurrent.ConcurrentHashMap;

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.datamap.dev.DataMapModel;
import org.apache.carbondata.core.datamap.dev.cgdatamap.CoarseGrainDataMap;
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.devapi.DictionaryGenerationException;
import org.apache.carbondata.core.indexstore.Blocklet;
import org.apache.carbondata.core.indexstore.PartitionSpec;
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
import org.apache.carbondata.core.metadata.CarbonMetadata;
import org.apache.carbondata.core.metadata.datatype.DataTypes;
import org.apache.carbondata.core.metadata.encoder.Encoding;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.RelationIdentifier;
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.logical.AndExpression;
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;

/**
 * BloomDataCoarseGrainMap 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 datamap writer.
 */
@InterfaceAudience.Internal
public class BloomCoarseGrainDataMap extends CoarseGrainDataMap {
  private static final Logger LOGGER =
      LogServiceFactory.getLogService(BloomCoarseGrainDataMap.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(DataMapModel dataMapModel) throws IOException {
    this.indexPath = FileFactory.getPath(dataMapModel.getFilePath());
    this.shardName = indexPath.getName();
    if (dataMapModel instanceof BloomDataMapModel) {
      BloomDataMapModel model = (BloomDataMapModel) dataMapModel;
      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);
    }
    String parentTablePath = getAncestorTablePath(carbonTable);

    try {
      this.name2Converters = new HashMap<>(indexedColumn.size());
      AbsoluteTableIdentifier absoluteTableIdentifier = AbsoluteTableIdentifier
          .from(carbonTable.getTablePath(), carbonTable.getCarbonTableIdentifier());
      String nullFormat = "\\N";
      Map<Object, Integer>[] localCaches = new Map[indexedColumn.size()];

      for (int i = 0; i < indexedColumn.size(); i++) {
        localCaches[i] = new ConcurrentHashMap<>();
        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, absoluteTableIdentifier, i, nullFormat, null, false,
                localCaches[i], false, parentTablePath, false);
        this.name2Converters.put(indexedColumn.get(i).getColName(), fieldConverter);
      }
    } catch (IOException e) {
      LOGGER.error("Exception occurs while init index columns", e);
      throw new RuntimeException(e);
    }
    this.badRecordLogHolder = new BadRecordLogHolder();
    this.badRecordLogHolder.setLogged(false);
  }

  /**
   * recursively find the ancestor's table path. This is used for dictionary scenario
   * where preagg will use the dictionary of the parent table.
   */
  private String getAncestorTablePath(CarbonTable currentTable) {
    if (!currentTable.isChildDataMap()) {
      return currentTable.getTablePath();
    }

    RelationIdentifier parentIdentifier =
        currentTable.getTableInfo().getParentRelationIdentifiers().get(0);
    CarbonTable parentTable = CarbonMetadata.getInstance().getCarbonTable(
        parentIdentifier.getDatabaseName(), parentIdentifier.getTableName());
    return getAncestorTablePath(parentTable);
  }

  @Override
  public List<Blocklet> prune(FilterResolverIntf filterExp, SegmentProperties segmentProperties,
      List<PartitionSpec> partitions) throws IOException {
    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;
    try {
      bloomQueryModels = createQueryModel(filterExp.getFilterExpression());
    } catch (DictionaryGenerationException | UnsupportedEncodingException e) {
      LOGGER.error("Exception occurs while creating query model", e);
      throw new RuntimeException(e);
    }
    for (BloomQueryModel bloomQueryModel : bloomQueryModels) {
      Set<Blocklet> tempHitBlockletsResult = new HashSet<>();
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("prune blocklet for query: " + bloomQueryModel);
      }
      BloomCacheKeyValue.CacheKey cacheKey = new BloomCacheKeyValue.CacheKey(
          this.indexPath.toString(), bloomQueryModel.columnName);
      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 datamap
          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("BloomCoarseGrainDataMap: 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("BloomCoarseGrainDataMap: 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 datamap
        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 null;
    }
    return new ArrayList<>(hitBlocklets);
  }

  private List<BloomQueryModel> createQueryModel(Expression expression)
      throws DictionaryGenerationException, UnsupportedEncodingException {
    List<BloomQueryModel> queryModels = new ArrayList<BloomQueryModel>();
    // bloomdatamap 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;
  }

  /**
   * Here preprocessed NULL and date/timestamp data type.
   *
   * Note that if the datatype is date/timestamp, the expressionValue is long type.
   */
  private Object getLiteralExpValue(LiteralExpression le) {
    Object expressionValue = le.getLiteralExpValue();
    Object literalValue;

    if (null == expressionValue) {
      literalValue = null;
    } else if (le.getLiteralExpDataType() == DataTypes.DATE) {
      DateFormat format = new SimpleDateFormat(CarbonCommonConstants.CARBON_DATE_DEFAULT_FORMAT);
      // the below settings are set statically according to DateDirectDirectionaryGenerator
      format.setLenient(false);
      format.setTimeZone(TimeZone.getTimeZone("GMT"));
      literalValue = format.format(new Date((long) expressionValue / 1000));
    } else if (le.getLiteralExpDataType() == DataTypes.TIMESTAMP) {
      DateFormat format =
          new SimpleDateFormat(CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT);
      // the below settings are set statically according to TimeStampDirectDirectionaryGenerator
      format.setLenient(false);
      literalValue = format.format(new Date((long) expressionValue / 1000));
    } else {
      literalValue = expressionValue;
    }
    return literalValue;
  }


  private BloomQueryModel buildQueryModelForEqual(ColumnExpression ce,
      LiteralExpression le) throws DictionaryGenerationException, UnsupportedEncodingException {
    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)
      throws DictionaryGenerationException, UnsupportedEncodingException {
    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) throws
      DictionaryGenerationException, UnsupportedEncodingException {
    Object filterLiteralValue = getLiteralExpValue(le);
    // convert the filter value to string and apply converters on it to get carbon internal value
    String strFilterValue = null;
    if (null != filterLiteralValue) {
      strFilterValue = String.valueOf(filterLiteralValue);
    }

    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.hasEncoding(Encoding.DIRECT_DICTIONARY) ||
        carbonColumn.hasEncoding(Encoding.DICTIONARY)) {
      // 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() {

  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.carbondata.datamap.bloom;
19
20	import java.io.IOException;
21	import java.io.UnsupportedEncodingException;
22	import java.text.DateFormat;
23	import java.text.SimpleDateFormat;
24	import java.util.ArrayList;
25	import java.util.Arrays;
26	import java.util.Date;
27	import java.util.HashMap;
28	import java.util.HashSet;
29	import java.util.List;
30	import java.util.Map;
31	import java.util.Set;
32	import java.util.TimeZone;
33	import java.util.concurrent.ConcurrentHashMap;
34
35	import org.apache.carbondata.common.annotations.InterfaceAudience;
36	import org.apache.carbondata.common.logging.LogServiceFactory;
37	import org.apache.carbondata.core.cache.Cache;
38	import org.apache.carbondata.core.constants.CarbonCommonConstants;
39	import org.apache.carbondata.core.datamap.dev.DataMapModel;
40	import org.apache.carbondata.core.datamap.dev.cgdatamap.CoarseGrainDataMap;
41	import org.apache.carbondata.core.datastore.block.SegmentProperties;
42	import org.apache.carbondata.core.datastore.impl.FileFactory;
43	import org.apache.carbondata.core.datastore.page.encoding.bool.BooleanConvert;
44	import org.apache.carbondata.core.devapi.DictionaryGenerationException;
45	import org.apache.carbondata.core.indexstore.Blocklet;
46	import org.apache.carbondata.core.indexstore.PartitionSpec;
47	import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
48	import org.apache.carbondata.core.metadata.CarbonMetadata;
49	import org.apache.carbondata.core.metadata.datatype.DataTypes;
50	import org.apache.carbondata.core.metadata.encoder.Encoding;
51	import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
52	import org.apache.carbondata.core.metadata.schema.table.RelationIdentifier;
53	import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
54	import org.apache.carbondata.core.scan.expression.ColumnExpression;
55	import org.apache.carbondata.core.scan.expression.Expression;
56	import org.apache.carbondata.core.scan.expression.LiteralExpression;
57	import org.apache.carbondata.core.scan.expression.conditional.EqualToExpression;
58	import org.apache.carbondata.core.scan.expression.conditional.InExpression;
59	import org.apache.carbondata.core.scan.expression.conditional.ListExpression;
60	import org.apache.carbondata.core.scan.expression.logical.AndExpression;
61	import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
62	import org.apache.carbondata.core.util.CarbonProperties;
63	import org.apache.carbondata.core.util.CarbonUtil;
64	import org.apache.carbondata.core.util.DataTypeUtil;
65	import org.apache.carbondata.processing.loading.DataField;
66	import org.apache.carbondata.processing.loading.converter.BadRecordLogHolder;
67	import org.apache.carbondata.processing.loading.converter.FieldConverter;
68	import org.apache.carbondata.processing.loading.converter.impl.FieldEncoderFactory;
69
70	import org.apache.hadoop.fs.Path;
71	import org.apache.hadoop.util.bloom.CarbonBloomFilter;
72	import org.apache.hadoop.util.bloom.Key;
73	import org.apache.log4j.Logger;
74
75	/**
76	* BloomDataCoarseGrainMap is constructed in blocklet level. For each indexed column,
77	* a bloom filter is constructed to indicate whether a value belongs to this blocklet.
78	* More information of the index file can be found in the corresponding datamap writer.
79	*/
80	@InterfaceAudience.Internal
81	public class BloomCoarseGrainDataMap extends CoarseGrainDataMap {	1✔
82	private static final Logger LOGGER =	1✔
83	LogServiceFactory.getLogService(BloomCoarseGrainDataMap.class.getName());	1✔
84	private Map<String, CarbonColumn> name2Col;
85	private Cache<BloomCacheKeyValue.CacheKey, BloomCacheKeyValue.CacheValue> cache;
86	private String shardName;
87	private Path indexPath;
88	private Set<String> filteredShard;
89	private boolean needShardPrune;
90	/**
91	* This is used to convert literal filter value to internal carbon value
92	*/
93	private Map<String, FieldConverter> name2Converters;
94	private BadRecordLogHolder badRecordLogHolder;
95
96	@Override
97	public void init(DataMapModel dataMapModel) throws IOException {
98	this.indexPath = FileFactory.getPath(dataMapModel.getFilePath());	1✔
99	this.shardName = indexPath.getName();	1✔
100	if (dataMapModel instanceof BloomDataMapModel) {	1✔
101	BloomDataMapModel model = (BloomDataMapModel) dataMapModel;	1✔
102	this.cache = model.getCache();	1✔
103	}
104	}	1✔
105
106	public void setFilteredShard(Set<String> filteredShard) {
107	this.filteredShard = filteredShard;	1✔
108	// do shard prune when pruning only if bloom index files are merged
109	this.needShardPrune = filteredShard != null &&	1✔
110	shardName.equals(BloomIndexFileStore.MERGE_BLOOM_INDEX_SHARD_NAME);	1✔
111	}	1✔
112
113	/**
114	* init field converters for index columns
115	*/
116	public void initIndexColumnConverters(CarbonTable carbonTable, List<CarbonColumn> indexedColumn) {
117	this.name2Col = new HashMap<>(indexedColumn.size());	1✔
118	for (CarbonColumn col : indexedColumn) {	1✔
119	this.name2Col.put(col.getColName(), col);	1✔
120	}	1✔
121	String parentTablePath = getAncestorTablePath(carbonTable);	1✔
122
123	try {
124	this.name2Converters = new HashMap<>(indexedColumn.size());	1✔
125	AbsoluteTableIdentifier absoluteTableIdentifier = AbsoluteTableIdentifier	1✔
126	.from(carbonTable.getTablePath(), carbonTable.getCarbonTableIdentifier());	1✔
127	String nullFormat = "\\N";	1✔
128	Map<Object, Integer>[] localCaches = new Map[indexedColumn.size()];	1✔
129
130	for (int i = 0; i < indexedColumn.size(); i++) {	1✔
131	localCaches[i] = new ConcurrentHashMap<>();	1✔
132	DataField dataField = new DataField(indexedColumn.get(i));	1✔
133	String dateFormat = CarbonProperties.getInstance().getProperty(	1✔
134	CarbonCommonConstants.CARBON_DATE_FORMAT,
135	CarbonCommonConstants.CARBON_DATE_DEFAULT_FORMAT);
136	dataField.setDateFormat(dateFormat);	1✔
137	String tsFormat = CarbonProperties.getInstance().getProperty(	1✔
138	CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT,
139	CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT);
140	dataField.setTimestampFormat(tsFormat);	1✔
141	FieldConverter fieldConverter = FieldEncoderFactory.getInstance()	1✔
142	.createFieldEncoder(dataField, absoluteTableIdentifier, i, nullFormat, null, false,	1✔
143	localCaches[i], false, parentTablePath, false);
144	this.name2Converters.put(indexedColumn.get(i).getColName(), fieldConverter);	1✔
145	}
146	} catch (IOException e) {	×
147	LOGGER.error("Exception occurs while init index columns", e);	×
148	throw new RuntimeException(e);	×
149	}	1✔
150	this.badRecordLogHolder = new BadRecordLogHolder();	1✔
151	this.badRecordLogHolder.setLogged(false);	1✔
152	}	1✔
153
154	/**
155	* recursively find the ancestor's table path. This is used for dictionary scenario
156	* where preagg will use the dictionary of the parent table.
157	*/
158	private String getAncestorTablePath(CarbonTable currentTable) {
159	if (!currentTable.isChildDataMap()) {	1✔
160	return currentTable.getTablePath();	1✔
161	}
162
163	RelationIdentifier parentIdentifier =	×
164	currentTable.getTableInfo().getParentRelationIdentifiers().get(0);	×
165	CarbonTable parentTable = CarbonMetadata.getInstance().getCarbonTable(	×
166	parentIdentifier.getDatabaseName(), parentIdentifier.getTableName());	×
167	return getAncestorTablePath(parentTable);	×
168	}
169
170	@Override
171	public List<Blocklet> prune(FilterResolverIntf filterExp, SegmentProperties segmentProperties,
172	List<PartitionSpec> partitions) throws IOException {
173	Set<Blocklet> hitBlocklets = null;	1✔
174	if (filterExp == null) {	1✔
175	// null is different from empty here. Empty means after pruning, no blocklet need to scan.
176	return null;	×
177	}
178	if (filteredShard.isEmpty()) {	1✔
179	LOGGER.info("Bloom filtered shards is empty");	×
180	return new ArrayList<>();	×
181	}
182
183	List<BloomQueryModel> bloomQueryModels;
184	try {
185	bloomQueryModels = createQueryModel(filterExp.getFilterExpression());	1✔
186	} catch (DictionaryGenerationException \| UnsupportedEncodingException e) {	×
187	LOGGER.error("Exception occurs while creating query model", e);	×
188	throw new RuntimeException(e);	×
189	}	1✔
190	for (BloomQueryModel bloomQueryModel : bloomQueryModels) {	1✔
191	Set<Blocklet> tempHitBlockletsResult = new HashSet<>();	1✔
192	if (LOGGER.isDebugEnabled()) {	1✔
193	LOGGER.debug("prune blocklet for query: " + bloomQueryModel);	×
194	}
195	BloomCacheKeyValue.CacheKey cacheKey = new BloomCacheKeyValue.CacheKey(	1✔
196	this.indexPath.toString(), bloomQueryModel.columnName);	1✔
197	BloomCacheKeyValue.CacheValue cacheValue = cache.get(cacheKey);	1✔
198	List<CarbonBloomFilter> bloomIndexList = cacheValue.getBloomFilters();	1✔
199	for (CarbonBloomFilter bloomFilter : bloomIndexList) {	1✔
200	if (needShardPrune && !filteredShard.contains(bloomFilter.getShardName())) {	1✔
201	// skip shard which has been pruned in Main datamap
202	continue;	×
203	}
204	boolean scanRequired = false;	1✔
205	for (byte[] value: bloomQueryModel.filterValues) {	1✔
206	scanRequired = bloomFilter.membershipTest(new Key(value));	1✔
207	if (scanRequired) {	1✔
208	// if any filter value hit this bloomfilter
209	// no need to check other filter values
210	break;	1✔
211	}
212	}	1✔
213	if (scanRequired) {	1✔
214	if (LOGGER.isDebugEnabled()) {	1✔
215	LOGGER.debug(String.format("BloomCoarseGrainDataMap: Need to scan -> blocklet#%s",	×
216	String.valueOf(bloomFilter.getBlockletNo())));	×
217	}
218	Blocklet blocklet = new Blocklet(bloomFilter.getShardName(),	1✔
219	String.valueOf(bloomFilter.getBlockletNo()));	1✔
220	tempHitBlockletsResult.add(blocklet);	1✔
221	} else if (LOGGER.isDebugEnabled()) {	1✔
222	LOGGER.debug(String.format("BloomCoarseGrainDataMap: Skip scan -> blocklet#%s",	×
223	String.valueOf(bloomFilter.getBlockletNo())));	×
224	}
225	// get intersect result between query models
226	// pre-condition: only And/In/EqualTo expression exists in single bloom datamap
227	if (null == hitBlocklets) {	1✔
228	hitBlocklets = tempHitBlockletsResult;	1✔
229	} else {
230	hitBlocklets.retainAll(tempHitBlockletsResult);	1✔
231	}
232	}	1✔
233	}	1✔
234	if (hitBlocklets == null) {	1✔
235	LOGGER.warn(String.format("HitBlocklets is empty in bloom filter prune method. " +	×
236	"bloomQueryModels size is %d, filterShards size if %d",
237	bloomQueryModels.size(), filteredShard.size()));	×
238	return null;	×
239	}
240	return new ArrayList<>(hitBlocklets);	1✔
241	}
242
243	private List<BloomQueryModel> createQueryModel(Expression expression)
244	throws DictionaryGenerationException, UnsupportedEncodingException {
245	List<BloomQueryModel> queryModels = new ArrayList<BloomQueryModel>();	1✔
246	// bloomdatamap only support equalTo and In operators now
247	if (expression instanceof EqualToExpression) {	1✔
248	Expression left = ((EqualToExpression) expression).getLeft();	1✔
249	Expression right = ((EqualToExpression) expression).getRight();	1✔
250	String column;
251	if (left instanceof ColumnExpression && right instanceof LiteralExpression) {	1✔
252	column = ((ColumnExpression) left).getColumnName();	1✔
253	if (this.name2Col.containsKey(column)) {	1✔
254	BloomQueryModel bloomQueryModel =	1✔
255	buildQueryModelForEqual((ColumnExpression) left, (LiteralExpression) right);	1✔
256	queryModels.add(bloomQueryModel);	1✔
257	}
258	return queryModels;	1✔
259	} else if (left instanceof LiteralExpression && right instanceof ColumnExpression) {	×
260	column = ((ColumnExpression) right).getColumnName();	×
261	if (this.name2Col.containsKey(column)) {	×
262	BloomQueryModel bloomQueryModel =	×
263	buildQueryModelForEqual((ColumnExpression) right, (LiteralExpression) left);	×
264	queryModels.add(bloomQueryModel);	×
265	}
266	return queryModels;	×
267	} else {
268	String errorMsg = "BloomFilter can only support the 'equal' filter like 'Col = PlainValue'";	×
269	LOGGER.warn(errorMsg);	×
270	throw new RuntimeException(errorMsg);	×
271	}
272	} else if (expression instanceof InExpression) {	1✔
273	Expression left = ((InExpression) expression).getLeft();	1✔
274	Expression right = ((InExpression) expression).getRight();	1✔
275	String column;
276	if (left instanceof ColumnExpression && right instanceof ListExpression) {	1✔
277	column = ((ColumnExpression) left).getColumnName();	1✔
278	if (this.name2Col.containsKey(column)) {	1✔
279	BloomQueryModel bloomQueryModel =	1✔
280	buildQueryModelForIn((ColumnExpression) left, (ListExpression) right);	1✔
281	queryModels.add(bloomQueryModel);	1✔
282	}
283	return queryModels;	1✔
284	} else if (left instanceof ListExpression && right instanceof ColumnExpression) {	×
285	column = ((ColumnExpression) right).getColumnName();	×
286	if (this.name2Col.containsKey(column)) {	×
287	BloomQueryModel bloomQueryModel =	×
288	buildQueryModelForIn((ColumnExpression) right, (ListExpression) left);	×
289	queryModels.add(bloomQueryModel);	×
290	}
291	return queryModels;	×
292	} else {
293	String errorMsg = "BloomFilter can only support the 'in' filter like 'Col in PlainValue'";	×
294	LOGGER.warn(errorMsg);	×
295	throw new RuntimeException(errorMsg);	×
296	}
297	} else if (expression instanceof AndExpression) {	1✔
298	queryModels.addAll(createQueryModel(((AndExpression) expression).getLeft()));	1✔
299	queryModels.addAll(createQueryModel(((AndExpression) expression).getRight()));	1✔
300	return queryModels;	1✔
301	}
302
303	return queryModels;	×
304	}
305
306	/**
307	* Here preprocessed NULL and date/timestamp data type.
308	*
309	* Note that if the datatype is date/timestamp, the expressionValue is long type.
310	*/
311	private Object getLiteralExpValue(LiteralExpression le) {
312	Object expressionValue = le.getLiteralExpValue();	1✔
313	Object literalValue;
314
315	if (null == expressionValue) {	1✔
316	literalValue = null;	1✔
317	} else if (le.getLiteralExpDataType() == DataTypes.DATE) {	1✔
318	DateFormat format = new SimpleDateFormat(CarbonCommonConstants.CARBON_DATE_DEFAULT_FORMAT);	1✔
319	// the below settings are set statically according to DateDirectDirectionaryGenerator
320	format.setLenient(false);	1✔
321	format.setTimeZone(TimeZone.getTimeZone("GMT"));	1✔
322	literalValue = format.format(new Date((long) expressionValue / 1000));	1✔
323	} else if (le.getLiteralExpDataType() == DataTypes.TIMESTAMP) {	1✔
324	DateFormat format =	1✔
325	new SimpleDateFormat(CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT);
326	// the below settings are set statically according to TimeStampDirectDirectionaryGenerator
327	format.setLenient(false);	1✔
328	literalValue = format.format(new Date((long) expressionValue / 1000));	1✔
329	} else {	1✔
330	literalValue = expressionValue;	1✔
331	}
332	return literalValue;	1✔
333	}
334
335
336	private BloomQueryModel buildQueryModelForEqual(ColumnExpression ce,
337	LiteralExpression le) throws DictionaryGenerationException, UnsupportedEncodingException {
338	List<byte[]> filterValues = new ArrayList<>();	1✔
339	byte[] internalFilterValue = getInternalFilterValue(this.name2Col.get(ce.getColumnName()), le);	1✔
340	filterValues.add(internalFilterValue);	1✔
341	return new BloomQueryModel(ce.getColumnName(), filterValues);	1✔
342	}
343
344	/**
345	* Note that `in` operator needs at least one match not exactly match. since while doing pruning,
346	* we collect all the blocklets that will match the querymodel, this will not be a problem.
347	*/
348	private BloomQueryModel buildQueryModelForIn(ColumnExpression ce, ListExpression le)
349	throws DictionaryGenerationException, UnsupportedEncodingException {
350	List<byte[]> filterValues = new ArrayList<>();	1✔
351	for (Expression child : le.getChildren()) {	1✔
352	byte[] internalFilterValue = getInternalFilterValue(	1✔
353	this.name2Col.get(ce.getColumnName()), (LiteralExpression) child);	1✔
354	filterValues.add(internalFilterValue);	1✔
355	}	1✔
356	return new BloomQueryModel(ce.getColumnName(), filterValues);	1✔
357	}
358
359	private byte[] getInternalFilterValue(CarbonColumn carbonColumn, LiteralExpression le) throws
360	DictionaryGenerationException, UnsupportedEncodingException {
361	Object filterLiteralValue = getLiteralExpValue(le);	1✔
362	// convert the filter value to string and apply converters on it to get carbon internal value
363	String strFilterValue = null;	1✔
364	if (null != filterLiteralValue) {	1✔
365	strFilterValue = String.valueOf(filterLiteralValue);	1✔
366	}
367
368	Object convertedValue = this.name2Converters.get(carbonColumn.getColName()).convert(	1✔
369	strFilterValue, badRecordLogHolder);
370
371	byte[] internalFilterValue;
372	if (carbonColumn.isMeasure()) {	1✔
373	// for measures, the value is already the type, just convert it to bytes.
374	if (convertedValue == null) {	1✔
375	convertedValue = DataConvertUtil.getNullValueForMeasure(carbonColumn.getDataType(),	1✔
376	carbonColumn.getColumnSchema().getScale());	1✔
377	}
378	// Carbon stores boolean as byte. Here we convert it for `getValueAsBytes`
379	if (carbonColumn.getDataType().equals(DataTypes.BOOLEAN)) {	1✔
380	convertedValue = BooleanConvert.boolean2Byte((Boolean)convertedValue);	1✔
381	}
382	internalFilterValue = CarbonUtil.getValueAsBytes(carbonColumn.getDataType(), convertedValue);	1✔
383	} else if (carbonColumn.hasEncoding(Encoding.DIRECT_DICTIONARY) \|\|	1✔
384	carbonColumn.hasEncoding(Encoding.DICTIONARY)) {	1✔
385	// for dictionary/date columns, convert the surrogate key to bytes
386	internalFilterValue = CarbonUtil.getValueAsBytes(DataTypes.INT, convertedValue);	1✔
387	} else {
388	// for non dictionary dimensions, numeric columns will be of original data,
389	// so convert the data to bytes
390	if (DataTypeUtil.isPrimitiveColumn(carbonColumn.getDataType())) {	1✔
391	if (convertedValue == null) {	1✔
392	convertedValue = DataConvertUtil.getNullValueForMeasure(carbonColumn.getDataType(),	1✔
393	carbonColumn.getColumnSchema().getScale());	1✔
394	}
395	internalFilterValue =	1✔
396	CarbonUtil.getValueAsBytes(carbonColumn.getDataType(), convertedValue);	1✔
397	} else {
398	internalFilterValue = (byte[]) convertedValue;	1✔
399	}
400	}
401	if (internalFilterValue.length == 0) {	1✔
402	internalFilterValue = CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY;	×
403	}
404	return internalFilterValue;	1✔
405	}
406
407	@Override
408	public boolean isScanRequired(FilterResolverIntf filterExp) {
409	return true;	×
410	}
411
412	@Override
413	public void clear() {
414	}	×
415
416	static class BloomQueryModel {
417	private String columnName;
418	private List<byte[]> filterValues;
419
420	/**
421	* represent an query model will be applyied on bloom index
422	*
423	* @param columnName bloom index column
424	* @param filterValues key for the bloom index,
425	* this value is converted from user specified filter value in query
426	*/
427	private BloomQueryModel(String columnName, List<byte[]> filterValues) {	1✔
428	this.columnName = columnName;	1✔
429	this.filterValues = filterValues;	1✔
430	}	1✔
431
432	@Override
433	public String toString() {
434	final StringBuilder sb = new StringBuilder("BloomQueryModel{");	×
435	sb.append("columnName='").append(columnName).append('\'');	×
436	sb.append(", filterValues=");	×
437	for (byte[] value: filterValues) {	×
438	sb.append(Arrays.toString(value));	×
439	}	×
440	sb.append('}');	×
441	return sb.toString();	×
442	}
443	}
444
445	@Override
446	public void finish() {
447
448	}	1✔
449	}

apache / carbondata / 2259

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous