12089

Build # 12089

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/ea8225337f4b7ed03d2af0c72e0ece7d6912121b">ea8225337<a href="https://github.com/apache/carbondata/commit/ea8225337f4b7ed03d2af0c72e0ece7d6912121b">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/c0492570bfa2beccca4e87d92b91994e74be37fb">c0492570b</a>">c0492570b</a><a href="https://github.com/apache/carbondata/commit/ea8225337f4b7ed03d2af0c72e0ece7d6912121b"> into ebe78dca1">ebe78dca1</a>

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

65024 of 83616 relevant lines covered (77.77%)

1.02 hits per line

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

76.67

/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapFactory.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.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.carbondata.common.annotations.InterfaceAudience;
import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.cache.Cache;
import org.apache.carbondata.core.cache.CacheProvider;
import org.apache.carbondata.core.cache.CacheType;
import org.apache.carbondata.core.constants.CarbonV3DataFormatConstants;
import org.apache.carbondata.core.datamap.DataMapDistributable;
import org.apache.carbondata.core.datamap.DataMapLevel;
import org.apache.carbondata.core.datamap.DataMapMeta;
import org.apache.carbondata.core.datamap.Segment;
import org.apache.carbondata.core.datamap.dev.DataMapBuilder;
import org.apache.carbondata.core.datamap.dev.DataMapFactory;
import org.apache.carbondata.core.datamap.dev.DataMapWriter;
import org.apache.carbondata.core.datamap.dev.cgdatamap.CoarseGrainDataMap;
import org.apache.carbondata.core.datastore.block.SegmentProperties;
import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.features.TableOperation;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
import org.apache.carbondata.core.scan.filter.intf.ExpressionType;
import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.path.CarbonTablePath;
import org.apache.carbondata.events.Event;

import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Logger;

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

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

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

  public BloomCoarseGrainDataMapFactory(CarbonTable carbonTable, DataMapSchema dataMapSchema)
      throws MalformedDataMapCommandException {
    super(carbonTable, dataMapSchema);
    Objects.requireNonNull(carbonTable);
    Objects.requireNonNull(dataMapSchema);

    this.dataMapName = dataMapSchema.getDataMapName();

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

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

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

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

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

  @Override
  public DataMapBuilder createBuilder(Segment segment, String shardName,
      SegmentProperties segmentProperties) throws IOException {
    return new BloomDataMapBuilder(getCarbonTable().getTablePath(), this.dataMapName,
        this.dataMapMeta.getIndexedColumns(), segment, shardName, segmentProperties,
        this.bloomFilterSize, this.bloomFilterFpp, bloomCompress);
  }

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

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

  @Override
  public List<CoarseGrainDataMap> getDataMaps(DataMapDistributable distributable)
      throws IOException {
    List<CoarseGrainDataMap> dataMaps = new ArrayList<>();
    String indexPath = ((BloomDataMapDistributable) distributable).getIndexPath();
    Set<String> filteredShards = ((BloomDataMapDistributable) distributable).getFilteredShards();
    BloomCoarseGrainDataMap bloomDM = new BloomCoarseGrainDataMap();
    bloomDM.init(new BloomDataMapModel(indexPath, cache, FileFactory.getConfiguration()));
    bloomDM.initIndexColumnConverters(getCarbonTable(), dataMapMeta.getIndexedColumns());
    bloomDM.setFilteredShard(filteredShards);
    dataMaps.add(bloomDM);
    return dataMaps;
  }


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

  @Override
  public void fireEvent(Event event) {

  }

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

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

  @Override
  public void deleteDatamapData(Segment segment) throws IOException {
    deleteSegmentDatamapData(segment.getSegmentNo());
  }

  @Override
  public void deleteSegmentDatamapData(String segmentId) throws IOException {
    try {
      String datamapPath = CarbonTablePath
          .getDataMapStorePath(getCarbonTable().getTablePath(), segmentId, dataMapName);
      if (FileFactory.isFileExist(datamapPath)) {
        CarbonFile file = FileFactory.getCarbonFile(datamapPath,
            FileFactory.getFileType(datamapPath));
        CarbonUtil.deleteFoldersAndFilesSilent(file);
      }
      clear(segmentId);
    } catch (InterruptedException ex) {
      throw new IOException("Failed to delete datamap for segment_" + segmentId);
    }
  }

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

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

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

  @Override
  public DataMapMeta getMeta() {
    return this.dataMapMeta;
  }

  @Override
  public DataMapLevel getDataMapLevel() {
    return DataMapLevel.CG;
  }

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

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	package org.apache.carbondata.datamap.bloom;
18
19	import java.io.File;
20	import java.io.IOException;
21	import java.util.ArrayList;
22	import java.util.HashSet;
23	import java.util.List;
24	import java.util.Map;
25	import java.util.Objects;
26	import java.util.Set;
27	import java.util.concurrent.ConcurrentHashMap;
28
29	import org.apache.carbondata.common.annotations.InterfaceAudience;
30	import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException;
31	import org.apache.carbondata.common.logging.LogServiceFactory;
32	import org.apache.carbondata.core.cache.Cache;
33	import org.apache.carbondata.core.cache.CacheProvider;
34	import org.apache.carbondata.core.cache.CacheType;
35	import org.apache.carbondata.core.constants.CarbonV3DataFormatConstants;
36	import org.apache.carbondata.core.datamap.DataMapDistributable;
37	import org.apache.carbondata.core.datamap.DataMapLevel;
38	import org.apache.carbondata.core.datamap.DataMapMeta;
39	import org.apache.carbondata.core.datamap.Segment;
40	import org.apache.carbondata.core.datamap.dev.DataMapBuilder;
41	import org.apache.carbondata.core.datamap.dev.DataMapFactory;
42	import org.apache.carbondata.core.datamap.dev.DataMapWriter;
43	import org.apache.carbondata.core.datamap.dev.cgdatamap.CoarseGrainDataMap;
44	import org.apache.carbondata.core.datastore.block.SegmentProperties;
45	import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
46	import org.apache.carbondata.core.datastore.impl.FileFactory;
47	import org.apache.carbondata.core.features.TableOperation;
48	import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
49	import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
50	import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
51	import org.apache.carbondata.core.scan.filter.intf.ExpressionType;
52	import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
53	import org.apache.carbondata.core.util.CarbonUtil;
54	import org.apache.carbondata.core.util.path.CarbonTablePath;
55	import org.apache.carbondata.events.Event;
56
57	import org.apache.commons.lang3.StringUtils;
58	import org.apache.log4j.Logger;
59
60	/**
61	* This class is for Bloom Filter for blocklet level
62	*/
63	@InterfaceAudience.Internal
64	public class BloomCoarseGrainDataMapFactory extends DataMapFactory<CoarseGrainDataMap> {
65	private static final Logger LOGGER = LogServiceFactory.getLogService(	1✔
66	BloomCoarseGrainDataMapFactory.class.getName());	1✔
67	/**
68	* property for size of bloom filter
69	*/
70	private static final String BLOOM_SIZE = "bloom_size";
71	/**
72	* default size for bloom filter, cardinality of the column.
73	*/
74	private static final int DEFAULT_BLOOM_FILTER_SIZE =
75	CarbonV3DataFormatConstants.NUMBER_OF_ROWS_PER_BLOCKLET_COLUMN_PAGE_DEFAULT * 20;
76	/**
77	* property for fpp(false-positive-probability) of bloom filter
78	*/
79	private static final String BLOOM_FPP = "bloom_fpp";
80	/**
81	* default value for fpp of bloom filter is 0.001%
82	*/
83	private static final double DEFAULT_BLOOM_FILTER_FPP = 0.00001d;
84
85	/**
86	* property for compressing bloom while saving to disk.
87	*/
88	private static final String COMPRESS_BLOOM = "bloom_compress";
89	/**
90	* Default value of compressing bloom while save to disk.
91	*/
92	private static final boolean DEFAULT_BLOOM_COMPRESS = true;
93
94	private DataMapMeta dataMapMeta;
95	private String dataMapName;
96	private int bloomFilterSize;
97	private double bloomFilterFpp;
98	private boolean bloomCompress;
99	private Cache<BloomCacheKeyValue.CacheKey, BloomCacheKeyValue.CacheValue> cache;
100	// segmentId -> list of index file
101	private Map<String, Set<String>> segmentMap = new ConcurrentHashMap<>();	1✔
102
103	public BloomCoarseGrainDataMapFactory(CarbonTable carbonTable, DataMapSchema dataMapSchema)
104	throws MalformedDataMapCommandException {
105	super(carbonTable, dataMapSchema);	1✔
106	Objects.requireNonNull(carbonTable);	1✔
107	Objects.requireNonNull(dataMapSchema);	1✔
108
109	this.dataMapName = dataMapSchema.getDataMapName();	1✔
110
111	List<CarbonColumn> indexedColumns = carbonTable.getIndexedColumns(dataMapSchema);	1✔
112	this.bloomFilterSize = validateAndGetBloomFilterSize(dataMapSchema);	1✔
113	this.bloomFilterFpp = validateAndGetBloomFilterFpp(dataMapSchema);	1✔
114	this.bloomCompress = validateAndGetBloomCompress(dataMapSchema);	1✔
115	List<ExpressionType> optimizedOperations = new ArrayList<ExpressionType>();	1✔
116	// todo: support more optimize operations
117	optimizedOperations.add(ExpressionType.EQUALS);	1✔
118	optimizedOperations.add(ExpressionType.IN);	1✔
119	this.dataMapMeta = new DataMapMeta(this.dataMapName, indexedColumns, optimizedOperations);	1✔
120	LOGGER.info(String.format("DataMap %s works for %s with bloom size %d",	1✔
121	this.dataMapName, this.dataMapMeta, this.bloomFilterSize));	1✔
122	try {
123	this.cache = CacheProvider.getInstance()	1✔
124	.createCache(new CacheType("bloom_cache"), BloomDataMapCache.class.getName());	1✔
125	} catch (Exception e) {	×
126	LOGGER.error(e.getMessage(), e);	×
127	throw new MalformedDataMapCommandException(e.getMessage());	×
128	}	1✔
129	}	1✔
130
131	/**
132	* validate Lucene DataMap BLOOM_SIZE
133	* 1. BLOOM_SIZE property is optional, 32000 * 20 will be the default size.
134	* 2. BLOOM_SIZE should be an integer that greater than 0
135	*/
136	private int validateAndGetBloomFilterSize(DataMapSchema dmSchema)
137	throws MalformedDataMapCommandException {
138	String bloomFilterSizeStr = dmSchema.getProperties().get(BLOOM_SIZE);	1✔
139	if (StringUtils.isBlank(bloomFilterSizeStr)) {	1✔
140	LOGGER.warn(	1✔
141	String.format("Bloom filter size is not configured for datamap %s, use default value %d",	1✔
142	dataMapName, DEFAULT_BLOOM_FILTER_SIZE));	1✔
143	return DEFAULT_BLOOM_FILTER_SIZE;	1✔
144	}
145	int bloomFilterSize;
146	try {
147	bloomFilterSize = Integer.parseInt(bloomFilterSizeStr);	1✔
148	} catch (NumberFormatException e) {	×
149	throw new MalformedDataMapCommandException(	×
150	String.format("Invalid value of bloom filter size '%s', it should be an integer",	×
151	bloomFilterSizeStr));
152	}	1✔
153	// todo: reconsider the boundaries of bloom filter size
154	if (bloomFilterSize <= 0) {	1✔
155	throw new MalformedDataMapCommandException(	×
156	String.format("Invalid value of bloom filter size '%s', it should be greater than 0",	×
157	bloomFilterSizeStr));
158	}
159	return bloomFilterSize;	1✔
160	}
161
162	/**
163	* validate bloom DataMap BLOOM_FPP
164	* 1. BLOOM_FPP property is optional, 0.00001 will be the default value.
165	* 2. BLOOM_FPP should be (0, 1)
166	*/
167	private double validateAndGetBloomFilterFpp(DataMapSchema dmSchema)
168	throws MalformedDataMapCommandException {
169	String bloomFilterFppStr = dmSchema.getProperties().get(BLOOM_FPP);	1✔
170	if (StringUtils.isBlank(bloomFilterFppStr)) {	1✔
171	LOGGER.warn(	1✔
172	String.format("Bloom filter FPP is not configured for datamap %s, use default value %f",	1✔
173	dataMapName, DEFAULT_BLOOM_FILTER_FPP));	1✔
174	return DEFAULT_BLOOM_FILTER_FPP;	1✔
175	}
176	double bloomFilterFpp;
177	try {
178	bloomFilterFpp = Double.parseDouble(bloomFilterFppStr);	1✔
179	} catch (NumberFormatException e) {	×
180	throw new MalformedDataMapCommandException(	×
181	String.format("Invalid value of bloom filter fpp '%s', it should be an numeric",	×
182	bloomFilterFppStr));
183	}	1✔
184	if (bloomFilterFpp < 0 \|\| bloomFilterFpp - 1 >= 0) {	1✔
185	throw new MalformedDataMapCommandException(	×
186	String.format("Invalid value of bloom filter fpp '%s', it should be in range 0~1",	×
187	bloomFilterFppStr));
188	}
189	return bloomFilterFpp;	1✔
190	}
191
192	/**
193	* validate bloom DataMap COMPRESS_BLOOM
194	* Default value is true
195	*/
196	private boolean validateAndGetBloomCompress(DataMapSchema dmSchema) {
197	String bloomCompress = dmSchema.getProperties().get(COMPRESS_BLOOM);	1✔
198	if (StringUtils.isBlank(bloomCompress)) {	1✔
199	LOGGER.warn(	1✔
200	String.format("Bloom compress is not configured for datamap %s, use default value %b",	1✔
201	dataMapName, DEFAULT_BLOOM_COMPRESS));	1✔
202	return DEFAULT_BLOOM_COMPRESS;	1✔
203	}
204	return Boolean.parseBoolean(bloomCompress);	1✔
205	}
206
207	@Override
208	public DataMapWriter createWriter(Segment segment, String shardName,
209	SegmentProperties segmentProperties) throws IOException {
210	LOGGER.info(	1✔
211	String.format("Data of BloomCoarseGranDataMap %s for table %s will be written to %s",	1✔
212	this.dataMapName, getCarbonTable().getTableName() , shardName));	1✔
213	return new BloomDataMapWriter(getCarbonTable().getTablePath(), this.dataMapName,	1✔
214	this.dataMapMeta.getIndexedColumns(), segment, shardName, segmentProperties,	1✔
215	this.bloomFilterSize, this.bloomFilterFpp, bloomCompress);
216	}
217
218	@Override
219	public DataMapBuilder createBuilder(Segment segment, String shardName,
220	SegmentProperties segmentProperties) throws IOException {
221	return new BloomDataMapBuilder(getCarbonTable().getTablePath(), this.dataMapName,	1✔
222	this.dataMapMeta.getIndexedColumns(), segment, shardName, segmentProperties,	1✔
223	this.bloomFilterSize, this.bloomFilterFpp, bloomCompress);
224	}
225
226	/**
227	* returns all shard directories of bloom index files for query
228	* if bloom index files are merged we should get only one shard path
229	*/
230	public static Set<String> getAllShardPaths(String tablePath, String segmentId,
231	String dataMapName) {
232	String dataMapStorePath = CarbonTablePath.getDataMapStorePath(	1✔
233	tablePath, segmentId, dataMapName);
234	CarbonFile[] carbonFiles = FileFactory.getCarbonFile(dataMapStorePath).listFiles();	1✔
235	Set<String> shardPaths = new HashSet<>();	1✔
236	boolean mergeShardInprogress = false;	1✔
237	CarbonFile mergeShardFile = null;	1✔
238	for (CarbonFile carbonFile : carbonFiles) {	1✔
239	if (carbonFile.getName().equals(BloomIndexFileStore.MERGE_BLOOM_INDEX_SHARD_NAME)) {	1✔
240	mergeShardFile = carbonFile;	1✔
241	} else if (carbonFile.getName().equals(BloomIndexFileStore.MERGE_INPROGRESS_FILE)) {	×
242	mergeShardInprogress = true;	×
243	} else if (carbonFile.isDirectory()) {	×
244	shardPaths.add(FileFactory.getPath(carbonFile.getAbsolutePath()).toString());	×
245	}
246	}
247	if (mergeShardFile != null && !mergeShardInprogress) {	1✔
248	// should only get one shard path if mergeShard is generated successfully
249	shardPaths.clear();	1✔
250	shardPaths.add(FileFactory.getPath(mergeShardFile.getAbsolutePath()).toString());	1✔
251	}
252	return shardPaths;	1✔
253	}
254
255	@Override
256	public List<CoarseGrainDataMap> getDataMaps(Segment segment) throws IOException {
257	List<CoarseGrainDataMap> dataMaps = new ArrayList<>();	1✔
258	try {
259	Set<String> shardPaths = segmentMap.get(segment.getSegmentNo());	1✔
260	if (shardPaths == null) {	1✔
261	shardPaths =	1✔
262	getAllShardPaths(getCarbonTable().getTablePath(), segment.getSegmentNo(), dataMapName);	1✔
263	segmentMap.put(segment.getSegmentNo(), shardPaths);	1✔
264	}
265	Set<String> filteredShards = segment.getFilteredIndexShardNames();	1✔
266	for (String shard : shardPaths) {	1✔
267	if (shard.endsWith(BloomIndexFileStore.MERGE_BLOOM_INDEX_SHARD_NAME) \|\|	1✔
268	filteredShards.contains(new File(shard).getName())) {	×
269	// Filter out the tasks which are filtered through Main datamap.
270	// for merge shard, shard pruning delay to be done before pruning blocklet
271	BloomCoarseGrainDataMap bloomDM = new BloomCoarseGrainDataMap();	1✔
272	bloomDM.init(new BloomDataMapModel(shard, cache, segment.getConfiguration()));	1✔
273	bloomDM.initIndexColumnConverters(getCarbonTable(), dataMapMeta.getIndexedColumns());	1✔
274	bloomDM.setFilteredShard(filteredShards);	1✔
275	dataMaps.add(bloomDM);	1✔
276	}
277	}	1✔
278	} catch (Exception e) {	×
279	throw new IOException("Error occurs while init Bloom DataMap", e);	×
280	}	1✔
281	return dataMaps;	1✔
282	}
283
284	@Override
285	public List<CoarseGrainDataMap> getDataMaps(DataMapDistributable distributable)
286	throws IOException {
287	List<CoarseGrainDataMap> dataMaps = new ArrayList<>();	1✔
288	String indexPath = ((BloomDataMapDistributable) distributable).getIndexPath();	1✔
289	Set<String> filteredShards = ((BloomDataMapDistributable) distributable).getFilteredShards();	1✔
290	BloomCoarseGrainDataMap bloomDM = new BloomCoarseGrainDataMap();	1✔
291	bloomDM.init(new BloomDataMapModel(indexPath, cache, FileFactory.getConfiguration()));	1✔
292	bloomDM.initIndexColumnConverters(getCarbonTable(), dataMapMeta.getIndexedColumns());	1✔
293	bloomDM.setFilteredShard(filteredShards);	1✔
294	dataMaps.add(bloomDM);	1✔
295	return dataMaps;	1✔
296	}
297
298
299	@Override
300	public List<DataMapDistributable> toDistributable(Segment segment) {
301	List<DataMapDistributable> dataMapDistributableList = new ArrayList<>();	1✔
302	Set<String> shardPaths = segmentMap.get(segment.getSegmentNo());	1✔
303	if (shardPaths == null) {	1✔
304	shardPaths =	1✔
305	getAllShardPaths(getCarbonTable().getTablePath(), segment.getSegmentNo(), dataMapName);	1✔
306	segmentMap.put(segment.getSegmentNo(), shardPaths);	1✔
307	}
308	Set<String> filteredShards = segment.getFilteredIndexShardNames();	1✔
309	for (String shardPath : shardPaths) {	1✔
310	// Filter out the tasks which are filtered through Main datamap.
311	// for merge shard, shard pruning delay to be done before pruning blocklet
312	if (shardPath.endsWith(BloomIndexFileStore.MERGE_BLOOM_INDEX_SHARD_NAME) \|\|	1✔
313	filteredShards.contains(new File(shardPath).getName())) {	×
314	DataMapDistributable bloomDataMapDistributable =	1✔
315	new BloomDataMapDistributable(shardPath, filteredShards);
316	bloomDataMapDistributable.setSegment(segment);	1✔
317	bloomDataMapDistributable.setDataMapSchema(getDataMapSchema());	1✔
318	dataMapDistributableList.add(bloomDataMapDistributable);	1✔
319	}
320	}	1✔
321	return dataMapDistributableList;	1✔
322	}
323
324	@Override
325	public void fireEvent(Event event) {
326
327	}	×
328
329	@Override
330	public void clear(String segment) {
331	Set<String> shards = segmentMap.remove(segment);	1✔
332	if (shards != null) {	1✔
333	for (String shard : shards) {	1✔
334	for (CarbonColumn carbonColumn : dataMapMeta.getIndexedColumns()) {	1✔
335	cache.invalidate(new BloomCacheKeyValue.CacheKey(shard, carbonColumn.getColName()));	1✔
336	}	1✔
337	}	1✔
338	}
339	}	1✔
340
341	@Override
342	public synchronized void clear() {
343	if (segmentMap.size() > 0) {	1✔
344	List<String> segments = new ArrayList<>(segmentMap.keySet());	1✔
345	for (String segmentId : segments) {	1✔
346	clear(segmentId);	1✔
347	}	1✔
348	}
349	}	1✔
350
351	@Override
352	public void deleteDatamapData(Segment segment) throws IOException {
353	deleteSegmentDatamapData(segment.getSegmentNo());	1✔
354	}	1✔
355
356	@Override
357	public void deleteSegmentDatamapData(String segmentId) throws IOException {
358	try {
359	String datamapPath = CarbonTablePath	1✔
360	.getDataMapStorePath(getCarbonTable().getTablePath(), segmentId, dataMapName);	1✔
361	if (FileFactory.isFileExist(datamapPath)) {	1✔
362	CarbonFile file = FileFactory.getCarbonFile(datamapPath,	1✔
363	FileFactory.getFileType(datamapPath));	1✔
364	CarbonUtil.deleteFoldersAndFilesSilent(file);	1✔
365	}
366	clear(segmentId);	1✔
367	} catch (InterruptedException ex) {	×
368	throw new IOException("Failed to delete datamap for segment_" + segmentId);	×
369	}	1✔
370	}	1✔
371
372	@Override
373	public void deleteDatamapData() {
374	SegmentStatusManager ssm =	1✔
375	new SegmentStatusManager(getCarbonTable().getAbsoluteTableIdentifier());	1✔
376	try {
377	List<Segment> validSegments =	1✔
378	ssm.getValidAndInvalidSegments(getCarbonTable().isChildTable()).getValidSegments();	1✔
379	for (Segment segment : validSegments) {	1✔
380	deleteDatamapData(segment);	×
381	}	×
382	} catch (IOException e) {	×
383	LOGGER.error("drop datamap failed, failed to delete datamap directory");	×
384	}	1✔
385	}	1✔
386
387	@Override
388	public boolean willBecomeStale(TableOperation operation) {
389	switch (operation) {	1✔
390	case ALTER_RENAME:
391	return false;	1✔
392	case ALTER_DROP:
393	return true;	1✔
394	case ALTER_ADD_COLUMN:
395	return false;	1✔
396	case ALTER_CHANGE_DATATYPE:
397	return true;	1✔
398	case ALTER_COLUMN_RENAME:
399	return true;	1✔
400	case STREAMING:
401	return false;	×
402	case DELETE:
403	return true;	×
404	case UPDATE:
405	return true;	×
406	case PARTITION:
407	return true;	×
408	default:
409	return false;	×
410	}
411	}
412
413	@Override
414	public boolean isOperationBlocked(TableOperation operation, Object... targets) {
415	switch (operation) {	1✔
416	case ALTER_DROP: {
417	// alter table drop columns
418	// will be blocked if the columns in bloomfilter datamap
419	List<String> columnsToDrop = (List<String>) targets[0];	×
420	List<String> indexedColumnNames = dataMapMeta.getIndexedColumnNames();	×
421	for (String indexedcolumn : indexedColumnNames) {	×
422	for (String column : columnsToDrop) {	×
423	if (column.equalsIgnoreCase(indexedcolumn)) {	×
424	return true;	×
425	}
426	}	×
427	}	×
428	return false;	×
429	}
430	case ALTER_CHANGE_DATATYPE:
431	case ALTER_COLUMN_RENAME: {
432	// alter table change one column datatype, or rename
433	// will be blocked if the column in bloomfilter datamap
434	String columnToChangeDatatype = (String) targets[0];	×
435	List<String> indexedColumnNames = dataMapMeta.getIndexedColumnNames();	×
436	for (String indexedcolumn : indexedColumnNames) {	×
437	if (indexedcolumn.equalsIgnoreCase(columnToChangeDatatype)) {	×
438	return true;	×
439	}
440	}	×
441	return false;	×
442	}
443	default:
444	return false;	1✔
445	}
446	}
447
448	@Override
449	public DataMapMeta getMeta() {
450	return this.dataMapMeta;	1✔
451	}
452
453	@Override
454	public DataMapLevel getDataMapLevel() {
455	return DataMapLevel.CG;	1✔
456	}
457
458	@Override
459	public String getCacheSize() {
460	long sum = 0L;	1✔
461	for (Map.Entry<String, Set<String>> entry : segmentMap.entrySet()) {	1✔
462	for (String shardName : entry.getValue()) {	1✔
463	for (CarbonColumn carbonColumn : dataMapMeta.getIndexedColumns()) {	1✔
464	BloomCacheKeyValue.CacheValue cacheValue = cache	1✔
465	.getIfPresent(new BloomCacheKeyValue.CacheKey(shardName, carbonColumn.getColName()));	1✔
466	if (cacheValue != null) {	1✔
467	sum += cacheValue.getMemorySize();	1✔
468	}
469	}	1✔
470	}	1✔
471	} return 0 + ":" + sum;	1✔
472	}
473	}

apache / carbondata / 12089

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