Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[CARBONDATA-3318] Added PreAgg & Bloom Event-Listener for ShowCacheCo…
…mmmand Decoupling of Cache Commands 1. Added PreAgg Event-Listener for ShowCacheCommmand 2. Added Bloom Event-Listener for ShowCacheCommmand 3. Added PreAgg Event-Listener for DropCacheCommand 4. Added Bloom Event-Listener for DropCacheCommmand 5. Updated doc 6. Support external table 6.1 display external table in comments/with table name 6.2 count the index files for external table This closes #3146
- Loading branch information
1 parent
66982f3
commit 2980803
Showing
13 changed files
with
581 additions
and
346 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
108 changes: 108 additions & 0 deletions
108
...ration/spark2/src/main/scala/org/apache/spark/sql/execution/command/cache/CacheUtil.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,108 @@ | ||
| /* | ||
| * 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.spark.sql.execution.command.cache | ||
|
|
||
| import org.apache.hadoop.mapred.JobConf | ||
| import scala.collection.JavaConverters._ | ||
|
|
||
| import org.apache.carbondata.core.cache.CacheType | ||
| import org.apache.carbondata.core.constants.CarbonCommonConstants | ||
| import org.apache.carbondata.core.datamap.Segment | ||
| import org.apache.carbondata.core.datastore.impl.FileFactory | ||
| import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema} | ||
| import org.apache.carbondata.core.readcommitter.LatestFilesReadCommittedScope | ||
| import org.apache.carbondata.datamap.bloom.{BloomCacheKeyValue, BloomCoarseGrainDataMapFactory} | ||
| import org.apache.carbondata.processing.merger.CarbonDataMergerUtil | ||
|
|
||
|
|
||
| object CacheUtil { | ||
|
|
||
| /** | ||
| * Given a carbonTable, returns the list of all carbonindex files | ||
| * | ||
| * @param carbonTable | ||
| * @return List of all index files | ||
| */ | ||
| def getAllIndexFiles(carbonTable: CarbonTable): List[String] = { | ||
| if (carbonTable.isTransactionalTable) { | ||
| val absoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier | ||
| CarbonDataMergerUtil.getValidSegmentList(absoluteTableIdentifier).asScala.flatMap { | ||
| segment => | ||
| segment.getCommittedIndexFile.keySet().asScala | ||
| }.toList | ||
| } else { | ||
| val tablePath = carbonTable.getTablePath | ||
| val readCommittedScope = new LatestFilesReadCommittedScope(tablePath, | ||
| FileFactory.getConfiguration) | ||
| readCommittedScope.getSegmentList.flatMap { | ||
| load => | ||
| val seg = new Segment(load.getLoadName, null, readCommittedScope) | ||
| seg.getCommittedIndexFile.keySet().asScala | ||
| }.toList | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Given a carbonTable file, returns a list of all dictionary entries which can be in cache | ||
| * | ||
| * @param carbonTable | ||
| * @return List of all dict entries which can in cache | ||
| */ | ||
| def getAllDictCacheKeys(carbonTable: CarbonTable): List[String] = { | ||
| def getDictCacheKey(columnIdentifier: String, | ||
| cacheType: CacheType[_, _]): String = { | ||
| columnIdentifier + CarbonCommonConstants.UNDERSCORE + cacheType.getCacheName | ||
| } | ||
|
|
||
| carbonTable.getAllDimensions.asScala | ||
| .collect { | ||
| case dict if dict.isGlobalDictionaryEncoding => | ||
| Seq(getDictCacheKey(dict.getColumnId, CacheType.FORWARD_DICTIONARY), | ||
| getDictCacheKey(dict.getColumnId, CacheType.REVERSE_DICTIONARY)) | ||
| }.flatten.toList | ||
| } | ||
|
|
||
| def getBloomCacheKeys(carbonTable: CarbonTable, datamap: DataMapSchema): List[String] = { | ||
| val segments = CarbonDataMergerUtil | ||
| .getValidSegmentList(carbonTable.getAbsoluteTableIdentifier).asScala | ||
|
|
||
| // Generate shard Path for the datamap | ||
| val shardPaths = segments.flatMap { | ||
| segment => | ||
| BloomCoarseGrainDataMapFactory.getAllShardPaths(carbonTable.getTablePath, | ||
| segment.getSegmentNo, datamap.getDataMapName).asScala | ||
| } | ||
|
|
||
| // get index columns | ||
| val indexColumns = carbonTable.getIndexedColumns(datamap).asScala.map { | ||
| entry => | ||
| entry.getColName | ||
| } | ||
|
|
||
| // generate cache key using shard path and index columns on which bloom was created. | ||
| val datamapKeys = shardPaths.flatMap { | ||
| shardPath => | ||
| indexColumns.map { | ||
| indexCol => | ||
| new BloomCacheKeyValue.CacheKey(shardPath, indexCol).toString | ||
| } | ||
| } | ||
| datamapKeys.toList | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.