9454

Build # 9454

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/bb1e87797e452bc317cd047917450646e9982a6e">bb1e87797<a href="https://github.com/apache/carbondata/commit/bb1e87797e452bc317cd047917450646e9982a6e">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/3259d25bf74a37c3316b8357abab2d9654570652">3259d25bf</a>">3259d25bf</a><a href="https://github.com/apache/carbondata/commit/bb1e87797e452bc317cd047917450646e9982a6e"> into 94a4f8314">94a4f8314</a>

Run Details

47 of 47 new or added lines in 9 files covered. (100.0%)

61805 of 77120 relevant lines covered (80.14%)

1.06 hits per line

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

81.31

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

    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<>();
      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) {
          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 {
          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);
        }
      }
    }
    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
179	List<BloomQueryModel> bloomQueryModels;
180	try {
181	bloomQueryModels = createQueryModel(filterExp.getFilterExpression());	1✔
182	} catch (DictionaryGenerationException \| UnsupportedEncodingException e) {	×
183	LOGGER.error("Exception occurs while creating query model", e);	×
184	throw new RuntimeException(e);	×
185	}	1✔
186	for (BloomQueryModel bloomQueryModel : bloomQueryModels) {	1✔
187	Set<Blocklet> tempHitBlockletsResult = new HashSet<>();	1✔
188	LOGGER.debug("prune blocklet for query: " + bloomQueryModel);	1✔
189	BloomCacheKeyValue.CacheKey cacheKey = new BloomCacheKeyValue.CacheKey(	1✔
190	this.indexPath.toString(), bloomQueryModel.columnName);	1✔
191	BloomCacheKeyValue.CacheValue cacheValue = cache.get(cacheKey);	1✔
192	List<CarbonBloomFilter> bloomIndexList = cacheValue.getBloomFilters();	1✔
193	for (CarbonBloomFilter bloomFilter : bloomIndexList) {	1✔
194	if (needShardPrune && !filteredShard.contains(bloomFilter.getShardName())) {	1✔
195	// skip shard which has been pruned in Main datamap
196	continue;	×
197	}
198	boolean scanRequired = false;	1✔
199	for (byte[] value: bloomQueryModel.filterValues) {	1✔
200	scanRequired = bloomFilter.membershipTest(new Key(value));	1✔
201	if (scanRequired) {	1✔
202	// if any filter value hit this bloomfilter
203	// no need to check other filter values
204	break;	1✔
205	}
206	}	1✔
207	if (scanRequired) {	1✔
208	LOGGER.debug(String.format("BloomCoarseGrainDataMap: Need to scan -> blocklet#%s",	1✔
209	String.valueOf(bloomFilter.getBlockletNo())));	1✔
210	Blocklet blocklet = new Blocklet(bloomFilter.getShardName(),	1✔
211	String.valueOf(bloomFilter.getBlockletNo()));	1✔
212	tempHitBlockletsResult.add(blocklet);	1✔
213	} else {	1✔
214	LOGGER.debug(String.format("BloomCoarseGrainDataMap: Skip scan -> blocklet#%s",	1✔
215	String.valueOf(bloomFilter.getBlockletNo())));	1✔
216	}
217	// get intersect result between query models
218	// pre-condition: only And/In/EqualTo expression exists in single bloom datamap
219	if (null == hitBlocklets) {	1✔
220	hitBlocklets = tempHitBlockletsResult;	1✔
221	} else {
222	hitBlocklets.retainAll(tempHitBlockletsResult);	1✔
223	}
224	}	1✔
225	}	1✔
226	return new ArrayList<>(hitBlocklets);	1✔
227	}
228
229	private List<BloomQueryModel> createQueryModel(Expression expression)
230	throws DictionaryGenerationException, UnsupportedEncodingException {
231	List<BloomQueryModel> queryModels = new ArrayList<BloomQueryModel>();	1✔
232	// bloomdatamap only support equalTo and In operators now
233	if (expression instanceof EqualToExpression) {	1✔
234	Expression left = ((EqualToExpression) expression).getLeft();	1✔
235	Expression right = ((EqualToExpression) expression).getRight();	1✔
236	String column;
237	if (left instanceof ColumnExpression && right instanceof LiteralExpression) {	1✔
238	column = ((ColumnExpression) left).getColumnName();	1✔
239	if (this.name2Col.containsKey(column)) {	1✔
240	BloomQueryModel bloomQueryModel =	1✔
241	buildQueryModelForEqual((ColumnExpression) left, (LiteralExpression) right);	1✔
242	queryModels.add(bloomQueryModel);	1✔
243	}
244	return queryModels;	1✔
245	} else if (left instanceof LiteralExpression && right instanceof ColumnExpression) {	×
246	column = ((ColumnExpression) right).getColumnName();	×
247	if (this.name2Col.containsKey(column)) {	×
248	BloomQueryModel bloomQueryModel =	×
249	buildQueryModelForEqual((ColumnExpression) right, (LiteralExpression) left);	×
250	queryModels.add(bloomQueryModel);	×
251	}
252	return queryModels;	×
253	} else {
254	String errorMsg = "BloomFilter can only support the 'equal' filter like 'Col = PlainValue'";	×
255	LOGGER.warn(errorMsg);	×
256	throw new RuntimeException(errorMsg);	×
257	}
258	} else if (expression instanceof InExpression) {	1✔
259	Expression left = ((InExpression) expression).getLeft();	1✔
260	Expression right = ((InExpression) expression).getRight();	1✔
261	String column;
262	if (left instanceof ColumnExpression && right instanceof ListExpression) {	1✔
263	column = ((ColumnExpression) left).getColumnName();	1✔
264	if (this.name2Col.containsKey(column)) {	1✔
265	BloomQueryModel bloomQueryModel =	1✔
266	buildQueryModelForIn((ColumnExpression) left, (ListExpression) right);	1✔
267	queryModels.add(bloomQueryModel);	1✔
268	}
269	return queryModels;	1✔
270	} else if (left instanceof ListExpression && right instanceof ColumnExpression) {	×
271	column = ((ColumnExpression) right).getColumnName();	×
272	if (this.name2Col.containsKey(column)) {	×
273	BloomQueryModel bloomQueryModel =	×
274	buildQueryModelForIn((ColumnExpression) right, (ListExpression) left);	×
275	queryModels.add(bloomQueryModel);	×
276	}
277	return queryModels;	×
278	} else {
279	String errorMsg = "BloomFilter can only support the 'in' filter like 'Col in PlainValue'";	×
280	LOGGER.warn(errorMsg);	×
281	throw new RuntimeException(errorMsg);	×
282	}
283	} else if (expression instanceof AndExpression) {	1✔
284	queryModels.addAll(createQueryModel(((AndExpression) expression).getLeft()));	1✔
285	queryModels.addAll(createQueryModel(((AndExpression) expression).getRight()));	1✔
286	return queryModels;	1✔
287	}
288
289	return queryModels;	×
290	}
291
292	/**
293	* Here preprocessed NULL and date/timestamp data type.
294	*
295	* Note that if the datatype is date/timestamp, the expressionValue is long type.
296	*/
297	private Object getLiteralExpValue(LiteralExpression le) {
298	Object expressionValue = le.getLiteralExpValue();	1✔
299	Object literalValue;
300
301	if (null == expressionValue) {	1✔
302	literalValue = null;	1✔
303	} else if (le.getLiteralExpDataType() == DataTypes.DATE) {	1✔
304	DateFormat format = new SimpleDateFormat(CarbonCommonConstants.CARBON_DATE_DEFAULT_FORMAT);	1✔
305	// the below settings are set statically according to DateDirectDirectionaryGenerator
306	format.setLenient(false);	1✔
307	format.setTimeZone(TimeZone.getTimeZone("GMT"));	1✔
308	literalValue = format.format(new Date((long) expressionValue / 1000));	1✔
309	} else if (le.getLiteralExpDataType() == DataTypes.TIMESTAMP) {	1✔
310	DateFormat format =	1✔
311	new SimpleDateFormat(CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT);
312	// the below settings are set statically according to TimeStampDirectDirectionaryGenerator
313	format.setLenient(false);	1✔
314	literalValue = format.format(new Date((long) expressionValue / 1000));	1✔
315	} else {	1✔
316	literalValue = expressionValue;	1✔
317	}
318	return literalValue;	1✔
319	}
320
321
322	private BloomQueryModel buildQueryModelForEqual(ColumnExpression ce,
323	LiteralExpression le) throws DictionaryGenerationException, UnsupportedEncodingException {
324	List<byte[]> filterValues = new ArrayList<>();	1✔
325	byte[] internalFilterValue = getInternalFilterValue(this.name2Col.get(ce.getColumnName()), le);	1✔
326	filterValues.add(internalFilterValue);	1✔
327	return new BloomQueryModel(ce.getColumnName(), filterValues);	1✔
328	}
329
330	/**
331	* Note that `in` operator needs at least one match not exactly match. since while doing pruning,
332	* we collect all the blocklets that will match the querymodel, this will not be a problem.
333	*/
334	private BloomQueryModel buildQueryModelForIn(ColumnExpression ce, ListExpression le)
335	throws DictionaryGenerationException, UnsupportedEncodingException {
336	List<byte[]> filterValues = new ArrayList<>();	1✔
337	for (Expression child : le.getChildren()) {	1✔
338	byte[] internalFilterValue = getInternalFilterValue(	1✔
339	this.name2Col.get(ce.getColumnName()), (LiteralExpression) child);	1✔
340	filterValues.add(internalFilterValue);	1✔
341	}	1✔
342	return new BloomQueryModel(ce.getColumnName(), filterValues);	1✔
343	}
344
345	private byte[] getInternalFilterValue(CarbonColumn carbonColumn, LiteralExpression le) throws
346	DictionaryGenerationException, UnsupportedEncodingException {
347	Object filterLiteralValue = getLiteralExpValue(le);	1✔
348	// convert the filter value to string and apply converters on it to get carbon internal value
349	String strFilterValue = null;	1✔
350	if (null != filterLiteralValue) {	1✔
351	strFilterValue = String.valueOf(filterLiteralValue);	1✔
352	}
353
354	Object convertedValue = this.name2Converters.get(carbonColumn.getColName()).convert(	1✔
355	strFilterValue, badRecordLogHolder);
356
357	byte[] internalFilterValue;
358	if (carbonColumn.isMeasure()) {	1✔
359	// for measures, the value is already the type, just convert it to bytes.
360	if (convertedValue == null) {	1✔
361	convertedValue = DataConvertUtil.getNullValueForMeasure(carbonColumn.getDataType(),	1✔
362	carbonColumn.getColumnSchema().getScale());	1✔
363	}
364	// Carbon stores boolean as byte. Here we convert it for `getValueAsBytes`
365	if (carbonColumn.getDataType().equals(DataTypes.BOOLEAN)) {	1✔
366	convertedValue = BooleanConvert.boolean2Byte((Boolean)convertedValue);	1✔
367	}
368	internalFilterValue = CarbonUtil.getValueAsBytes(carbonColumn.getDataType(), convertedValue);	1✔
369	} else if (carbonColumn.hasEncoding(Encoding.DIRECT_DICTIONARY) \|\|	1✔
370	carbonColumn.hasEncoding(Encoding.DICTIONARY)) {	1✔
371	// for dictionary/date columns, convert the surrogate key to bytes
372	internalFilterValue = CarbonUtil.getValueAsBytes(DataTypes.INT, convertedValue);	1✔
373	} else {
374	// for non dictionary dimensions, numeric columns will be of original data,
375	// so convert the data to bytes
376	if (DataTypeUtil.isPrimitiveColumn(carbonColumn.getDataType())) {	1✔
377	if (convertedValue == null) {	1✔
378	convertedValue = DataConvertUtil.getNullValueForMeasure(carbonColumn.getDataType(),	1✔
379	carbonColumn.getColumnSchema().getScale());	1✔
380	}
381	internalFilterValue =	1✔
382	CarbonUtil.getValueAsBytes(carbonColumn.getDataType(), convertedValue);	1✔
383	} else {
384	internalFilterValue = (byte[]) convertedValue;	1✔
385	}
386	}
387	if (internalFilterValue.length == 0) {	1✔
388	internalFilterValue = CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY;	×
389	}
390	return internalFilterValue;	1✔
391	}
392
393	@Override
394	public boolean isScanRequired(FilterResolverIntf filterExp) {
395	return true;	×
396	}
397
398	@Override
399	public void clear() {
400	}	×
401
402	static class BloomQueryModel {
403	private String columnName;
404	private List<byte[]> filterValues;
405
406	/**
407	* represent an query model will be applyied on bloom index
408	*
409	* @param columnName bloom index column
410	* @param filterValues key for the bloom index,
411	* this value is converted from user specified filter value in query
412	*/
413	private BloomQueryModel(String columnName, List<byte[]> filterValues) {	1✔
414	this.columnName = columnName;	1✔
415	this.filterValues = filterValues;	1✔
416	}	1✔
417
418	@Override
419	public String toString() {
420	final StringBuilder sb = new StringBuilder("BloomQueryModel{");	1✔
421	sb.append("columnName='").append(columnName).append('\'');	1✔
422	sb.append(", filterValues=");	1✔
423	for (byte[] value: filterValues) {	1✔
424	sb.append(Arrays.toString(value));	1✔
425	}	1✔
426	sb.append('}');	1✔
427	return sb.toString();	1✔
428	}
429	}
430
431	@Override
432	public void finish() {
433
434	}	1✔
435	}

apache / carbondata / 9454

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