8064

Build # 8064

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/84aadc2f70d5d77a09e0f753092240ff4f2c1ee0">84aadc2f7<a href="https://github.com/apache/carbondata/commit/84aadc2f70d5d77a09e0f753092240ff4f2c1ee0">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/30a861a92ff53df8befd14ee48b7a37499ab7c96">30a861a92</a>">30a861a92</a><a href="https://github.com/apache/carbondata/commit/84aadc2f70d5d77a09e0f753092240ff4f2c1ee0"> into 12ab57992">12ab57992</a>

Run Details

308 of 308 new or added lines in 13 files covered. (100.0%)

57710 of 71310 relevant lines covered (80.93%)

1.07 hits per line

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

80.87

/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.File;
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.List;
import java.util.Map;
import java.util.TimeZone;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.carbondata.common.annotations.InterfaceAudience;
import org.apache.carbondata.common.logging.LogService;
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.DataType;
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.filter.resolver.FilterResolverIntf;
import org.apache.carbondata.core.util.CarbonProperties;
import org.apache.carbondata.core.util.CarbonUtil;
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.commons.collections.CollectionUtils;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.util.bloom.CarbonBloomFilter;
import org.apache.hadoop.util.bloom.Key;

/**
 * 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 LogService LOGGER =
      LogServiceFactory.getLogService(BloomCoarseGrainDataMap.class.getName());
  public static final String BLOOM_INDEX_SUFFIX = ".bloomindex";
  private Map<String, CarbonColumn> name2Col;
  private Cache<BloomCacheKeyValue.CacheKey, BloomCacheKeyValue.CacheValue> cache;
  private String shardName;
  private Path indexPath;
  /**
   * 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();
    }
  }

  /**
   * 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);
        this.name2Converters.put(indexedColumn.get(i).getColName(), fieldConverter);
      }
    } catch (IOException e) {
      LOGGER.error(e, "Exception occurs while init index columns");
      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 {
    // we will store the result in hitBlocklets. If it is not first pruning, we will do
    // intersection with the previous result.
    boolean isFirstPruning = true;
    List<Blocklet> hitBlocklets = new ArrayList<>();
    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(e, "Exception occurs while creating query model");
      throw new RuntimeException(e);
    }
    // the relationship for BloomQueryModels here is AND, so we do intersection to reduce
    // pruned result. We will return immediately if the datamap filter out all the blocklets
    for (BloomQueryModel bloomQueryModel : bloomQueryModels) {
      List<Blocklet> prunedResult = pruneByBloomQueryModel(bloomQueryModel);
      if (prunedResult.isEmpty()) {
        hitBlocklets = prunedResult;
        break;
      }
      if (isFirstPruning) {
        hitBlocklets = prunedResult;
        isFirstPruning = false;
      } else {
        hitBlocklets = (List) CollectionUtils.intersection(hitBlocklets, prunedResult);
        if (hitBlocklets.isEmpty()) {
          break;
        }
      }
    }
    return new ArrayList<>(hitBlocklets);
  }

  private List<Blocklet> pruneByBloomQueryModel(BloomQueryModel bloomQueryModel)
      throws IOException {
    List<Blocklet> hitBlocklets = new ArrayList<>();
    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) {
      boolean scanRequired = bloomFilter.membershipTest(new Key(bloomQueryModel.filterValue));
      if (scanRequired) {
        LOGGER.debug(
            String.format("BloomCoarseGrainDataMap: Need to scan -> shard-%s & blocklet#%s",
                shardName, String.valueOf(bloomFilter.getBlockletNo())));
        Blocklet blocklet = new Blocklet(shardName, String.valueOf(bloomFilter.getBlockletNo()));
        hitBlocklets.add(blocklet);
      } else {
        LOGGER.debug(
            String.format("BloomCoarseGrainDataMap: Skip scan -> shard-%s & blocklet#%s",
                shardName, String.valueOf(bloomFilter.getBlockletNo())));
      }
    }
    return 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)) {
          List<BloomQueryModel> models =
              buildQueryModelForIn((ColumnExpression) left, (ListExpression) right);
          queryModels.addAll(models);
        }
        return queryModels;
      } else if (left instanceof ListExpression && right instanceof ColumnExpression) {
        column = ((ColumnExpression) right).getColumnName();
        if (this.name2Col.containsKey(column)) {
          List<BloomQueryModel> models =
              buildQueryModelForIn((ColumnExpression) right, (ListExpression) left);
          queryModels.addAll(models);
        }
        return queryModels;
      } else {
        String errorMsg = "BloomFilter can only support the 'in' filter like 'Col in PlainValue'";
        LOGGER.warn(errorMsg);
        throw new RuntimeException(errorMsg);
      }
    }

    for (Expression child : expression.getChildren()) {
      queryModels.addAll(createQueryModel(child));
    }
    return queryModels;
  }

  private BloomQueryModel buildQueryModelForEqual(ColumnExpression ce,
      LiteralExpression le) throws DictionaryGenerationException, UnsupportedEncodingException {
    String columnName = ce.getColumnName();
    DataType dataType = ce.getDataType();
    Object expressionValue = le.getLiteralExpValue();
    Object literalValue;
    // note that if the datatype is date/timestamp, the expressionValue is long type.
    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 buildQueryModelInternal(this.name2Col.get(columnName), literalValue, dataType);
  }

  /**
   * for `in` expressions, we use `equal` to handle it.
   * 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 List<BloomQueryModel> buildQueryModelForIn(ColumnExpression ce, ListExpression le)
      throws DictionaryGenerationException, UnsupportedEncodingException {
    List<BloomQueryModel> queryModels = new ArrayList<>();
    for (Expression child : le.getChildren()) {
      queryModels.add(buildQueryModelForEqual(ce, (LiteralExpression) child));
    }
    return queryModels;
  }

  private BloomQueryModel buildQueryModelInternal(CarbonColumn carbonColumn,
      Object filterLiteralValue, DataType filterValueDataType) throws
      DictionaryGenerationException, UnsupportedEncodingException {
    // convert the filter value to string and apply convertes 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, is already bytes,
      internalFilterValue = (byte[]) convertedValue;
    }
    if (internalFilterValue.length == 0) {
      internalFilterValue = CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY;
    }
    return new BloomQueryModel(carbonColumn.getColName(), internalFilterValue);
  }

  @Override
  public boolean isScanRequired(FilterResolverIntf filterExp) {
    return true;
  }

  @Override
  public void clear() {
  }

  /**
   * get bloom index file
   * @param shardPath path for the shard
   * @param colName index column name
   */
  public static String getBloomIndexFile(String shardPath, String colName) {
    return shardPath.concat(File.separator).concat(colName).concat(BLOOM_INDEX_SUFFIX);
  }
  static class BloomQueryModel {
    private String columnName;
    private byte[] filterValue;

    /**
     * represent an query model will be applyied on bloom index
     *
     * @param columnName bloom index column
     * @param filterValue key for the bloom index,
     *                   this value is converted from user specified filter value in query
     */
    private BloomQueryModel(String columnName, byte[] filterValue) {
      this.columnName = columnName;
      this.filterValue = filterValue;
    }

    @Override
    public String toString() {
      final StringBuilder sb = new StringBuilder("BloomQueryModel{");
      sb.append("columnName='").append(columnName).append('\'');
      sb.append(", filterValue=").append(Arrays.toString(filterValue));
      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.File;
21	import java.io.IOException;
22	import java.io.UnsupportedEncodingException;
23	import java.text.DateFormat;
24	import java.text.SimpleDateFormat;
25	import java.util.ArrayList;
26	import java.util.Arrays;
27	import java.util.Date;
28	import java.util.HashMap;
29	import java.util.List;
30	import java.util.Map;
31	import java.util.TimeZone;
32	import java.util.concurrent.ConcurrentHashMap;
33
34	import org.apache.carbondata.common.annotations.InterfaceAudience;
35	import org.apache.carbondata.common.logging.LogService;
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.DataType;
50	import org.apache.carbondata.core.metadata.datatype.DataTypes;
51	import org.apache.carbondata.core.metadata.encoder.Encoding;
52	import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
53	import org.apache.carbondata.core.metadata.schema.table.RelationIdentifier;
54	import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
55	import org.apache.carbondata.core.scan.expression.ColumnExpression;
56	import org.apache.carbondata.core.scan.expression.Expression;
57	import org.apache.carbondata.core.scan.expression.LiteralExpression;
58	import org.apache.carbondata.core.scan.expression.conditional.EqualToExpression;
59	import org.apache.carbondata.core.scan.expression.conditional.InExpression;
60	import org.apache.carbondata.core.scan.expression.conditional.ListExpression;
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.processing.loading.DataField;
65	import org.apache.carbondata.processing.loading.converter.BadRecordLogHolder;
66	import org.apache.carbondata.processing.loading.converter.FieldConverter;
67	import org.apache.carbondata.processing.loading.converter.impl.FieldEncoderFactory;
68
69	import org.apache.commons.collections.CollectionUtils;
70	import org.apache.hadoop.fs.Path;
71	import org.apache.hadoop.util.bloom.CarbonBloomFilter;
72	import org.apache.hadoop.util.bloom.Key;
73
74	/**
75	* BloomDataCoarseGrainMap is constructed in blocklet level. For each indexed column,
76	* a bloom filter is constructed to indicate whether a value belongs to this blocklet.
77	* More information of the index file can be found in the corresponding datamap writer.
78	*/
79	@InterfaceAudience.Internal
80	public class BloomCoarseGrainDataMap extends CoarseGrainDataMap {	1✔
81	private static final LogService LOGGER =	1✔
82	LogServiceFactory.getLogService(BloomCoarseGrainDataMap.class.getName());	1✔
83	public static final String BLOOM_INDEX_SUFFIX = ".bloomindex";
84	private Map<String, CarbonColumn> name2Col;
85	private Cache<BloomCacheKeyValue.CacheKey, BloomCacheKeyValue.CacheValue> cache;
86	private String shardName;
87	private Path indexPath;
88	/**
89	* This is used to convert literal filter value to internal carbon value
90	*/
91	private Map<String, FieldConverter> name2Converters;
92	private BadRecordLogHolder badRecordLogHolder;
93
94	@Override
95	public void init(DataMapModel dataMapModel) throws IOException {
96	this.indexPath = FileFactory.getPath(dataMapModel.getFilePath());	1✔
97	this.shardName = indexPath.getName();	1✔
98	if (dataMapModel instanceof BloomDataMapModel) {	1✔
99	BloomDataMapModel model = (BloomDataMapModel) dataMapModel;	1✔
100	this.cache = model.getCache();	1✔
101	}
102	}	1✔
103
104	/**
105	* init field converters for index columns
106	*/
107	public void initIndexColumnConverters(CarbonTable carbonTable, List<CarbonColumn> indexedColumn) {
108	this.name2Col = new HashMap<>(indexedColumn.size());	1✔
109	for (CarbonColumn col : indexedColumn) {	1✔
110	this.name2Col.put(col.getColName(), col);	1✔
111	}	1✔
112	String parentTablePath = getAncestorTablePath(carbonTable);	1✔
113
114	try {
115	this.name2Converters = new HashMap<>(indexedColumn.size());	1✔
116	AbsoluteTableIdentifier absoluteTableIdentifier = AbsoluteTableIdentifier	1✔
117	.from(carbonTable.getTablePath(), carbonTable.getCarbonTableIdentifier());	1✔
118	String nullFormat = "\\N";	1✔
119	Map<Object, Integer>[] localCaches = new Map[indexedColumn.size()];	1✔
120
121	for (int i = 0; i < indexedColumn.size(); i++) {	1✔
122	localCaches[i] = new ConcurrentHashMap<>();	1✔
123	DataField dataField = new DataField(indexedColumn.get(i));	1✔
124	String dateFormat = CarbonProperties.getInstance().getProperty(	1✔
125	CarbonCommonConstants.CARBON_DATE_FORMAT,
126	CarbonCommonConstants.CARBON_DATE_DEFAULT_FORMAT);
127	dataField.setDateFormat(dateFormat);	1✔
128	String tsFormat = CarbonProperties.getInstance().getProperty(	1✔
129	CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT,
130	CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT);
131	dataField.setTimestampFormat(tsFormat);	1✔
132	FieldConverter fieldConverter = FieldEncoderFactory.getInstance()	1✔
133	.createFieldEncoder(dataField, absoluteTableIdentifier, i, nullFormat, null, false,	1✔
134	localCaches[i], false, parentTablePath);
135	this.name2Converters.put(indexedColumn.get(i).getColName(), fieldConverter);	1✔
136	}
137	} catch (IOException e) {	×
138	LOGGER.error(e, "Exception occurs while init index columns");	×
139	throw new RuntimeException(e);	×
140	}	1✔
141	this.badRecordLogHolder = new BadRecordLogHolder();	1✔
142	this.badRecordLogHolder.setLogged(false);	1✔
143	}	1✔
144
145	/**
146	* recursively find the ancestor's table path. This is used for dictionary scenario
147	* where preagg will use the dictionary of the parent table.
148	*/
149	private String getAncestorTablePath(CarbonTable currentTable) {
150	if (!currentTable.isChildDataMap()) {	1✔
151	return currentTable.getTablePath();	1✔
152	}
153
154	RelationIdentifier parentIdentifier =	×
155	currentTable.getTableInfo().getParentRelationIdentifiers().get(0);	×
156	CarbonTable parentTable = CarbonMetadata.getInstance().getCarbonTable(	×
157	parentIdentifier.getDatabaseName(), parentIdentifier.getTableName());	×
158	return getAncestorTablePath(parentTable);	×
159	}
160
161	@Override
162	public List<Blocklet> prune(FilterResolverIntf filterExp, SegmentProperties segmentProperties,
163	List<PartitionSpec> partitions) throws IOException {
164	// we will store the result in hitBlocklets. If it is not first pruning, we will do
165	// intersection with the previous result.
166	boolean isFirstPruning = true;	1✔
167	List<Blocklet> hitBlocklets = new ArrayList<>();	1✔
168	if (filterExp == null) {	1✔
169	// null is different from empty here. Empty means after pruning, no blocklet need to scan.
170	return null;	×
171	}
172
173	List<BloomQueryModel> bloomQueryModels;
174	try {
175	bloomQueryModels = createQueryModel(filterExp.getFilterExpression());	1✔
176	} catch (DictionaryGenerationException \| UnsupportedEncodingException e) {	×
177	LOGGER.error(e, "Exception occurs while creating query model");	×
178	throw new RuntimeException(e);	×
179	}	1✔
180	// the relationship for BloomQueryModels here is AND, so we do intersection to reduce
181	// pruned result. We will return immediately if the datamap filter out all the blocklets
182	for (BloomQueryModel bloomQueryModel : bloomQueryModels) {	1✔
183	List<Blocklet> prunedResult = pruneByBloomQueryModel(bloomQueryModel);	1✔
184	if (prunedResult.isEmpty()) {	1✔
185	hitBlocklets = prunedResult;	1✔
186	break;	1✔
187	}
188	if (isFirstPruning) {	1✔
189	hitBlocklets = prunedResult;	1✔
190	isFirstPruning = false;	1✔
191	} else {
192	hitBlocklets = (List) CollectionUtils.intersection(hitBlocklets, prunedResult);	1✔
193	if (hitBlocklets.isEmpty()) {	1✔
194	break;	×
195	}
196	}
197	}	1✔
198	return new ArrayList<>(hitBlocklets);	1✔
199	}
200
201	private List<Blocklet> pruneByBloomQueryModel(BloomQueryModel bloomQueryModel)
202	throws IOException {
203	List<Blocklet> hitBlocklets = new ArrayList<>();	1✔
204	LOGGER.debug("prune blocklet for query: " + bloomQueryModel);	1✔
205	BloomCacheKeyValue.CacheKey cacheKey = new BloomCacheKeyValue.CacheKey(	1✔
206	this.indexPath.toString(), bloomQueryModel.columnName);	1✔
207	BloomCacheKeyValue.CacheValue cacheValue = cache.get(cacheKey);	1✔
208	List<CarbonBloomFilter> bloomIndexList = cacheValue.getBloomFilters();	1✔
209	for (CarbonBloomFilter bloomFilter : bloomIndexList) {	1✔
210	boolean scanRequired = bloomFilter.membershipTest(new Key(bloomQueryModel.filterValue));	1✔
211	if (scanRequired) {	1✔
212	LOGGER.debug(	1✔
213	String.format("BloomCoarseGrainDataMap: Need to scan -> shard-%s & blocklet#%s",	1✔
214	shardName, String.valueOf(bloomFilter.getBlockletNo())));	1✔
215	Blocklet blocklet = new Blocklet(shardName, String.valueOf(bloomFilter.getBlockletNo()));	1✔
216	hitBlocklets.add(blocklet);	1✔
217	} else {	1✔
218	LOGGER.debug(	1✔
219	String.format("BloomCoarseGrainDataMap: Skip scan -> shard-%s & blocklet#%s",	1✔
220	shardName, String.valueOf(bloomFilter.getBlockletNo())));	1✔
221	}
222	}	1✔
223	return hitBlocklets;	1✔
224	}
225
226	private List<BloomQueryModel> createQueryModel(Expression expression)
227	throws DictionaryGenerationException, UnsupportedEncodingException {
228	List<BloomQueryModel> queryModels = new ArrayList<BloomQueryModel>();	1✔
229	// bloomdatamap only support equalTo and In operators now
230	if (expression instanceof EqualToExpression) {	1✔
231	Expression left = ((EqualToExpression) expression).getLeft();	1✔
232	Expression right = ((EqualToExpression) expression).getRight();	1✔
233	String column;
234	if (left instanceof ColumnExpression && right instanceof LiteralExpression) {	1✔
235	column = ((ColumnExpression) left).getColumnName();	1✔
236	if (this.name2Col.containsKey(column)) {	1✔
237	BloomQueryModel bloomQueryModel =	1✔
238	buildQueryModelForEqual((ColumnExpression) left, (LiteralExpression) right);	1✔
239	queryModels.add(bloomQueryModel);	1✔
240	}
241	return queryModels;	1✔
242	} else if (left instanceof LiteralExpression && right instanceof ColumnExpression) {	×
243	column = ((ColumnExpression) right).getColumnName();	×
244	if (this.name2Col.containsKey(column)) {	×
245	BloomQueryModel bloomQueryModel =	×
246	buildQueryModelForEqual((ColumnExpression) right, (LiteralExpression) left);	×
247	queryModels.add(bloomQueryModel);	×
248	}
249	return queryModels;	×
250	} else {
251	String errorMsg = "BloomFilter can only support the 'equal' filter like 'Col = PlainValue'";	×
252	LOGGER.warn(errorMsg);	×
253	throw new RuntimeException(errorMsg);	×
254	}
255	} else if (expression instanceof InExpression) {	1✔
256	Expression left = ((InExpression) expression).getLeft();	1✔
257	Expression right = ((InExpression) expression).getRight();	1✔
258	String column;
259	if (left instanceof ColumnExpression && right instanceof ListExpression) {	1✔
260	column = ((ColumnExpression) left).getColumnName();	1✔
261	if (this.name2Col.containsKey(column)) {	1✔
262	List<BloomQueryModel> models =	1✔
263	buildQueryModelForIn((ColumnExpression) left, (ListExpression) right);	1✔
264	queryModels.addAll(models);	1✔
265	}
266	return queryModels;	1✔
267	} else if (left instanceof ListExpression && right instanceof ColumnExpression) {	×
268	column = ((ColumnExpression) right).getColumnName();	×
269	if (this.name2Col.containsKey(column)) {	×
270	List<BloomQueryModel> models =	×
271	buildQueryModelForIn((ColumnExpression) right, (ListExpression) left);	×
272	queryModels.addAll(models);	×
273	}
274	return queryModels;	×
275	} else {
276	String errorMsg = "BloomFilter can only support the 'in' filter like 'Col in PlainValue'";	×
277	LOGGER.warn(errorMsg);	×
278	throw new RuntimeException(errorMsg);	×
279	}
280	}
281
282	for (Expression child : expression.getChildren()) {	1✔
283	queryModels.addAll(createQueryModel(child));	1✔
284	}	1✔
285	return queryModels;	1✔
286	}
287
288	private BloomQueryModel buildQueryModelForEqual(ColumnExpression ce,
289	LiteralExpression le) throws DictionaryGenerationException, UnsupportedEncodingException {
290	String columnName = ce.getColumnName();	1✔
291	DataType dataType = ce.getDataType();	1✔
292	Object expressionValue = le.getLiteralExpValue();	1✔
293	Object literalValue;
294	// note that if the datatype is date/timestamp, the expressionValue is long type.
295	if (null == expressionValue) {	1✔
296	literalValue = null;	1✔
297	} else if (le.getLiteralExpDataType() == DataTypes.DATE) {	1✔
298	DateFormat format = new SimpleDateFormat(CarbonCommonConstants.CARBON_DATE_DEFAULT_FORMAT);	1✔
299	// the below settings are set statically according to DateDirectDirectionaryGenerator
300	format.setLenient(false);	1✔
301	format.setTimeZone(TimeZone.getTimeZone("GMT"));	1✔
302
303	literalValue = format.format(new Date((long) expressionValue / 1000));	1✔
304	} else if (le.getLiteralExpDataType() == DataTypes.TIMESTAMP) {	1✔
305	DateFormat format =	1✔
306	new SimpleDateFormat(CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT);
307	// the below settings are set statically according to TimeStampDirectDirectionaryGenerator
308	format.setLenient(false);	1✔
309	literalValue = format.format(new Date((long) expressionValue / 1000));	1✔
310	} else {	1✔
311	literalValue = expressionValue;	1✔
312	}
313
314	return buildQueryModelInternal(this.name2Col.get(columnName), literalValue, dataType);	1✔
315	}
316
317	/**
318	* for `in` expressions, we use `equal` to handle it.
319	* Note that `in` operator needs at least one match not exactly match. since while doing pruning,
320	* we collect all the blocklets that will match the querymodel, this will not be a problem.
321	*/
322	private List<BloomQueryModel> buildQueryModelForIn(ColumnExpression ce, ListExpression le)
323	throws DictionaryGenerationException, UnsupportedEncodingException {
324	List<BloomQueryModel> queryModels = new ArrayList<>();	1✔
325	for (Expression child : le.getChildren()) {	1✔
326	queryModels.add(buildQueryModelForEqual(ce, (LiteralExpression) child));	1✔
327	}	1✔
328	return queryModels;	1✔
329	}
330
331	private BloomQueryModel buildQueryModelInternal(CarbonColumn carbonColumn,
332	Object filterLiteralValue, DataType filterValueDataType) throws
333	DictionaryGenerationException, UnsupportedEncodingException {
334	// convert the filter value to string and apply convertes on it to get carbon internal value
335	String strFilterValue = null;	1✔
336	if (null != filterLiteralValue) {	1✔
337	strFilterValue = String.valueOf(filterLiteralValue);	1✔
338	}
339
340	Object convertedValue = this.name2Converters.get(carbonColumn.getColName()).convert(	1✔
341	strFilterValue, badRecordLogHolder);
342
343	byte[] internalFilterValue;
344	if (carbonColumn.isMeasure()) {	1✔
345	// for measures, the value is already the type, just convert it to bytes.
346	if (convertedValue == null) {	1✔
347	convertedValue = DataConvertUtil.getNullValueForMeasure(carbonColumn.getDataType(),	1✔
348	carbonColumn.getColumnSchema().getScale());	1✔
349	}
350	// Carbon stores boolean as byte. Here we convert it for `getValueAsBytes`
351	if (carbonColumn.getDataType().equals(DataTypes.BOOLEAN)) {	1✔
352	convertedValue = BooleanConvert.boolean2Byte((Boolean)convertedValue);	1✔
353	}
354	internalFilterValue = CarbonUtil.getValueAsBytes(carbonColumn.getDataType(), convertedValue);	1✔
355	} else if (carbonColumn.hasEncoding(Encoding.DIRECT_DICTIONARY) \|\|	1✔
356	carbonColumn.hasEncoding(Encoding.DICTIONARY)) {	1✔
357	// for dictionary/date columns, convert the surrogate key to bytes
358	internalFilterValue = CarbonUtil.getValueAsBytes(DataTypes.INT, convertedValue);	1✔
359	} else {
360	// for non dictionary dimensions, is already bytes,
361	internalFilterValue = (byte[]) convertedValue;	1✔
362	}
363	if (internalFilterValue.length == 0) {	1✔
364	internalFilterValue = CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY;	1✔
365	}
366	return new BloomQueryModel(carbonColumn.getColName(), internalFilterValue);	1✔
367	}
368
369	@Override
370	public boolean isScanRequired(FilterResolverIntf filterExp) {
371	return true;	×
372	}
373
374	@Override
375	public void clear() {
376	}	×
377
378	/**
379	* get bloom index file
380	* @param shardPath path for the shard
381	* @param colName index column name
382	*/
383	public static String getBloomIndexFile(String shardPath, String colName) {
384	return shardPath.concat(File.separator).concat(colName).concat(BLOOM_INDEX_SUFFIX);	1✔
385	}
386	static class BloomQueryModel {
387	private String columnName;
388	private byte[] filterValue;
389
390	/**
391	* represent an query model will be applyied on bloom index
392	*
393	* @param columnName bloom index column
394	* @param filterValue key for the bloom index,
395	* this value is converted from user specified filter value in query
396	*/
397	private BloomQueryModel(String columnName, byte[] filterValue) {	1✔
398	this.columnName = columnName;	1✔
399	this.filterValue = filterValue;	1✔
400	}	1✔
401
402	@Override
403	public String toString() {
404	final StringBuilder sb = new StringBuilder("BloomQueryModel{");	1✔
405	sb.append("columnName='").append(columnName).append('\'');	1✔
406	sb.append(", filterValue=").append(Arrays.toString(filterValue));	1✔
407	sb.append('}');	1✔
408	return sb.toString();	1✔
409	}
410	}
411
412	@Override
413	public void finish() {
414
415	}	1✔
416	}

apache / carbondata / 8064

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