11532

Build # 11532

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/c440f4278533867f41a85408d74e7c4c13037c1f">c440f4278<a href="https://github.com/apache/carbondata/commit/c440f4278533867f41a85408d74e7c4c13037c1f">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/dbe8763f973ca213927c3cae56004d3c2b739b9f">dbe8763f9</a>">dbe8763f9</a><a href="https://github.com/apache/carbondata/commit/c440f4278533867f41a85408d74e7c4c13037c1f"> into 24fe230a9">24fe230a9</a>

Run Details

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

63987 of 81406 relevant lines covered (78.6%)

1.04 hits per line

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

88.89

/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/cache/CacheUtil.scala

/*
 * 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 scala.collection.JavaConverters._

import org.apache.log4j.Logger

import org.apache.carbondata.common.logging.LogServiceFactory
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.core.statusmanager.SegmentStatusManager
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.datamap.bloom.{BloomCacheKeyValue, BloomCoarseGrainDataMapFactory}
import org.apache.carbondata.indexserver.IndexServer
import org.apache.carbondata.processing.merger.CarbonDataMergerUtil


object CacheUtil {

  val LOGGER: Logger = LogServiceFactory.getLogService(this.getClass.getCanonicalName)

  /**
   * 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
      val validAndInvalidSegmentsInfo = new SegmentStatusManager(absoluteTableIdentifier)
        .getValidAndInvalidSegments()
      // Fire a job to clear the invalid segments cached in the executors.
      if (CarbonProperties.getInstance().isDistributedPruningEnabled(carbonTable.getDatabaseName,
        carbonTable.getTableName)) {
        val invalidSegmentIds = validAndInvalidSegmentsInfo.getInvalidSegments.asScala
          .map(_.getSegmentNo).toArray
        try {
          IndexServer.getClient.invalidateSegmentCache(carbonTable.getDatabaseName, carbonTable
            .getTableName, invalidSegmentIds)
        } catch {
          case e: Exception =>
            LOGGER.warn("Failed to clear cache from executors. ", e)
        }
      }
      validAndInvalidSegmentsInfo.getValidSegments.asScala.flatMap {
        segment =>
          segment.getCommittedIndexFile.keySet().asScala
      }.map { indexFile =>
        indexFile.replace(CarbonCommonConstants.WINDOWS_FILE_SEPARATOR,
          CarbonCommonConstants.FILE_SEPARATOR)
      }.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
      }.map { indexFile =>
        indexFile.replace(CarbonCommonConstants.WINDOWS_FILE_SEPARATOR,
          CarbonCommonConstants.FILE_SEPARATOR)
      }.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
  }

}

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.spark.sql.execution.command.cache
19
20	import scala.collection.JavaConverters._
21
22	import org.apache.log4j.Logger
23
24	import org.apache.carbondata.common.logging.LogServiceFactory
25	import org.apache.carbondata.core.cache.CacheType
26	import org.apache.carbondata.core.constants.CarbonCommonConstants
27	import org.apache.carbondata.core.datamap.Segment
28	import org.apache.carbondata.core.datastore.impl.FileFactory
29	import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema}
30	import org.apache.carbondata.core.readcommitter.LatestFilesReadCommittedScope
31	import org.apache.carbondata.core.statusmanager.SegmentStatusManager
32	import org.apache.carbondata.core.util.CarbonProperties
33	import org.apache.carbondata.datamap.bloom.{BloomCacheKeyValue, BloomCoarseGrainDataMapFactory}
34	import org.apache.carbondata.indexserver.IndexServer
35	import org.apache.carbondata.processing.merger.CarbonDataMergerUtil
36
37
38	object CacheUtil {
39
40	val LOGGER: Logger = LogServiceFactory.getLogService(this.getClass.getCanonicalName)	1✔
41
42	/**
43	* Given a carbonTable, returns the list of all carbonindex files
44	*
45	* @param carbonTable
46	* @return List of all index files
47	*/
48	def getAllIndexFiles(carbonTable: CarbonTable): List[String] = {
49	if (carbonTable.isTransactionalTable) {	1✔
50	val absoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier	1✔
51	val validAndInvalidSegmentsInfo = new SegmentStatusManager(absoluteTableIdentifier)	1✔
52	.getValidAndInvalidSegments()	1✔
53	// Fire a job to clear the invalid segments cached in the executors.
54	if (CarbonProperties.getInstance().isDistributedPruningEnabled(carbonTable.getDatabaseName,	1✔
55	carbonTable.getTableName)) {	1✔
56	val invalidSegmentIds = validAndInvalidSegmentsInfo.getInvalidSegments.asScala	×
57	.map(_.getSegmentNo).toArray	×
58	try {	×
59	IndexServer.getClient.invalidateSegmentCache(carbonTable.getDatabaseName, carbonTable	×
60	.getTableName, invalidSegmentIds)	×
61	} catch {
62	case e: Exception =>
63	LOGGER.warn("Failed to clear cache from executors. ", e)	×
64	}
65	}
66	validAndInvalidSegmentsInfo.getValidSegments.asScala.flatMap {	1✔
67	segment =>	1✔
68	segment.getCommittedIndexFile.keySet().asScala	1✔
69	}.map { indexFile =>	1✔
70	indexFile.replace(CarbonCommonConstants.WINDOWS_FILE_SEPARATOR,	1✔
71	CarbonCommonConstants.FILE_SEPARATOR)	1✔
72	}.toList	1✔
73	} else {
74	val tablePath = carbonTable.getTablePath	1✔
75	val readCommittedScope = new LatestFilesReadCommittedScope(tablePath,	1✔
76	FileFactory.getConfiguration)	1✔
77	readCommittedScope.getSegmentList.flatMap {	1✔
78	load =>	1✔
79	val seg = new Segment(load.getLoadName, null, readCommittedScope)	1✔
80	seg.getCommittedIndexFile.keySet().asScala	1✔
81	}.map { indexFile =>	1✔
82	indexFile.replace(CarbonCommonConstants.WINDOWS_FILE_SEPARATOR,	1✔
83	CarbonCommonConstants.FILE_SEPARATOR)	1✔
84	}.toList	1✔
85	}
86	}
87
88	/**
89	* Given a carbonTable file, returns a list of all dictionary entries which can be in cache
90	*
91	* @param carbonTable
92	* @return List of all dict entries which can in cache
93	*/
94	def getAllDictCacheKeys(carbonTable: CarbonTable): List[String] = {
95	def getDictCacheKey(columnIdentifier: String,
96	cacheType: CacheType[_, _]): String = {
97	columnIdentifier + CarbonCommonConstants.UNDERSCORE + cacheType.getCacheName	1✔
98	}
99
100	carbonTable.getAllDimensions.asScala	1✔
101	.collect {	1✔
102	case dict if dict.isGlobalDictionaryEncoding =>	1✔
103	Seq(getDictCacheKey(dict.getColumnId, CacheType.FORWARD_DICTIONARY),	1✔
104	getDictCacheKey(dict.getColumnId, CacheType.REVERSE_DICTIONARY))	1✔
105	}.flatten.toList	1✔
106	}
107
108	def getBloomCacheKeys(carbonTable: CarbonTable, datamap: DataMapSchema): List[String] = {
109	val segments = CarbonDataMergerUtil	1✔
110	.getValidSegmentList(carbonTable.getAbsoluteTableIdentifier).asScala	1✔
111
112	// Generate shard Path for the datamap
113	val shardPaths = segments.flatMap {	1✔
114	segment =>	1✔
115	BloomCoarseGrainDataMapFactory.getAllShardPaths(carbonTable.getTablePath,	1✔
116	segment.getSegmentNo, datamap.getDataMapName).asScala	1✔
117	}
118
119	// get index columns
120	val indexColumns = carbonTable.getIndexedColumns(datamap).asScala.map {	1✔
121	entry =>	1✔
122	entry.getColName	1✔
123	}
124
125	// generate cache key using shard path and index columns on which bloom was created.
126	val datamapKeys = shardPaths.flatMap {	1✔
127	shardPath =>	1✔
128	indexColumns.map {	1✔
129	indexCol =>	1✔
130	new BloomCacheKeyValue.CacheKey(shardPath, indexCol).toString	1✔
131	}
132	}
133	datamapKeys.toList	1✔
134	}
135
136	}	1✔

apache / carbondata / 11532

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