3960

Build # 3960

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/42f71293202e0212382ae00a652e16a8abc61949">42f712932<a href="https://github.com/apache/carbondata/commit/42f71293202e0212382ae00a652e16a8abc61949">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/1b568e29aab49b1578a2f07a2fcdbaaf9b9a2bf1">1b568e29a</a>">1b568e29a</a><a href="https://github.com/apache/carbondata/commit/42f71293202e0212382ae00a652e16a8abc61949"> into 785cc6cbb">785cc6cbb</a>

Run Details

26 of 26 new or added lines in 10 files covered. (100.0%)

65001 of 83646 relevant lines covered (77.71%)

1.02 hits per line

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

85.71

/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.spark.sql.SparkSession
import org.apache.spark.sql.util.SparkSQLUtil

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

apache / carbondata / 3960

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