12039

Build # 12039

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/ebe4400016d672331672bd714068c22c1c8e328d">ebe440001<a href="https://github.com/apache/carbondata/commit/ebe4400016d672331672bd714068c22c1c8e328d">">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/ebe4400016d672331672bd714068c22c1c8e328d"> into 91a799847">91a799847</a>

Run Details

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

65034 of 83574 relevant lines covered (77.82%)

1.02 hits per line

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

21.57

/integration/spark2/src/main/scala/org/apache/carbondata/indexserver/DataMapJobs.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.carbondata.indexserver

import java.util

import scala.collection.JavaConverters._

import org.apache.log4j.Logger
import org.apache.spark.sql.util.SparkSQLUtil
import org.apache.spark.util.SizeEstimator

import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.datamap.{AbstractDataMapJob, DistributableDataMapFormat}
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.indexstore.ExtendedBlocklet
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
import org.apache.carbondata.core.scan.expression.BinaryExpression
import org.apache.carbondata.core.scan.filter.FilterExpressionProcessor
import org.apache.carbondata.core.scan.filter.intf.ExpressionType
import org.apache.carbondata.core.scan.filter.resolver.{FilterResolverIntf, LogicalFilterResolverImpl, RowLevelFilterResolverImpl}
import org.apache.carbondata.core.util.CarbonUtil
import org.apache.carbondata.spark.util.CarbonScalaUtil.logTime

/**
 * Spark job to execute datamap job and prune all the datamaps distributable. This job will prune
 * and cache the appropriate datamaps in executor LRUCache.
 */
class DistributedDataMapJob extends AbstractDataMapJob {

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

  override def execute(dataMapFormat: DistributableDataMapFormat): util.List[ExtendedBlocklet] = {
    if (LOGGER.isDebugEnabled) {
      val messageSize = SizeEstimator.estimate(dataMapFormat)
      LOGGER.debug(s"Size of message sent to Index Server: $messageSize")
    }
    val splitFolderPath = CarbonUtil
      .createTempFolderForIndexServer(dataMapFormat.getCarbonTable.getTablePath,
        dataMapFormat.getQueryId)
    LOGGER
      .info("Temp folder path for Query ID: " + dataMapFormat.getQueryId + " is " + splitFolderPath)
    val (resonse, time) = logTime {
      try {
        val spark = SparkSQLUtil.getSparkSession
        dataMapFormat.setTaskGroupId(SparkSQLUtil.getTaskGroupId(spark))
        dataMapFormat.setTaskGroupDesc(SparkSQLUtil.getTaskGroupDesc(spark))
        var filterInf = dataMapFormat.getFilterResolverIntf
        val filterProcessor = new FilterExpressionProcessor
        filterInf = removeSparkUnknown(filterInf,
          dataMapFormat.getCarbonTable.getAbsoluteTableIdentifier, filterProcessor)
        dataMapFormat.setFilterResolverIntf(filterInf)
        IndexServer.getClient.getSplits(dataMapFormat)
          .getExtendedBlockets(dataMapFormat.getCarbonTable.getTablePath, dataMapFormat.getQueryId)
      } finally {
        val tmpPath = CarbonUtil
          .getIndexServerTempPath(dataMapFormat.getCarbonTable.getTablePath,
            dataMapFormat.getQueryId)
        if (null != splitFolderPath &&
            !splitFolderPath.deleteFile(tmpPath, FileFactory.getFileType(tmpPath))) {
          LOGGER.error("Problem while deleting the temp directory:" + tmpPath)
        }
      }
    }
    LOGGER.info(s"Time taken to get response from server: $time ms")
    resonse
  }

  /**
   * Iterate over FiltersReslover,
   *   a. Change only RowLevelFilterResolverImpl because SparkUnkown is part of it
   * and others FilterReslover like ConditionalFilterResolverImpl so directly return.
   *     b. Change SparkUnkownExpression to TrueExpression so that isScanRequired
   * selects block/blocklet.
   *
   * @param filterInf       FiltersReslover to be changed
   * @param tableIdentifer  AbsoluteTableIdentifier object
   * @param filterProcessor changed FiltersReslover.
   * @return
   */
  def removeSparkUnknown(filterInf: FilterResolverIntf,
      tableIdentifer: AbsoluteTableIdentifier,
                         filterProcessor: FilterExpressionProcessor): FilterResolverIntf = {
    if (filterInf.isInstanceOf[LogicalFilterResolverImpl]) {
      return new LogicalFilterResolverImpl(
        removeSparkUnknown(filterInf.getLeft, tableIdentifer, filterProcessor),
        removeSparkUnknown(filterInf.getRight, tableIdentifer, filterProcessor),
        filterProcessor.removeUnknownExpression(filterInf.getFilterExpression).
          asInstanceOf[BinaryExpression])
    }
    if (filterInf.isInstanceOf[RowLevelFilterResolverImpl] &&
        filterInf.getFilterExpression.getFilterExpressionType == ExpressionType.UNKNOWN) {
      return filterProcessor.changeUnknownResloverToTrue(tableIdentifer)
    }
    return filterInf;
  }
}

/**
 * Spark job to execute datamap job and prune all the datamaps distributable. This job will just
 * prune the datamaps but will not cache in executors.
 */
class EmbeddedDataMapJob extends AbstractDataMapJob {

  override def execute(dataMapFormat: DistributableDataMapFormat): util.List[ExtendedBlocklet] = {
    val spark = SparkSQLUtil.getSparkSession
    val originalJobDesc = spark.sparkContext.getLocalProperty("spark.job.description")
    dataMapFormat.setIsWriteToFile(false)
    dataMapFormat.setFallbackJob()
    val splits = IndexServer.getSplits(dataMapFormat).getExtendedBlockets(dataMapFormat
      .getCarbonTable.getTablePath, dataMapFormat.getQueryId)
    // Fire a job to clear the cache from executors as Embedded mode does not maintain the cache.
    IndexServer.invalidateSegmentCache(dataMapFormat.getCarbonTable, dataMapFormat
      .getValidSegmentIds.asScala.toArray)
    spark.sparkContext.setLocalProperty("spark.job.description", originalJobDesc)
    splits
  }

}

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.indexserver
18
19	import java.util
20
21	import scala.collection.JavaConverters._
22
23	import org.apache.log4j.Logger
24	import org.apache.spark.sql.util.SparkSQLUtil
25	import org.apache.spark.util.SizeEstimator
26
27	import org.apache.carbondata.common.logging.LogServiceFactory
28	import org.apache.carbondata.core.datamap.{AbstractDataMapJob, DistributableDataMapFormat}
29	import org.apache.carbondata.core.datastore.impl.FileFactory
30	import org.apache.carbondata.core.indexstore.ExtendedBlocklet
31	import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
32	import org.apache.carbondata.core.scan.expression.BinaryExpression
33	import org.apache.carbondata.core.scan.filter.FilterExpressionProcessor
34	import org.apache.carbondata.core.scan.filter.intf.ExpressionType
35	import org.apache.carbondata.core.scan.filter.resolver.{FilterResolverIntf, LogicalFilterResolverImpl, RowLevelFilterResolverImpl}
36	import org.apache.carbondata.core.util.CarbonUtil
37	import org.apache.carbondata.spark.util.CarbonScalaUtil.logTime
38
39	/**
40	* Spark job to execute datamap job and prune all the datamaps distributable. This job will prune
41	* and cache the appropriate datamaps in executor LRUCache.
42	*/
43	class DistributedDataMapJob extends AbstractDataMapJob {	×
44
45	val LOGGER: Logger = LogServiceFactory.getLogService(this.getClass.getCanonicalName)	×
46
47	override def execute(dataMapFormat: DistributableDataMapFormat): util.List[ExtendedBlocklet] = {
48	if (LOGGER.isDebugEnabled) {	×
49	val messageSize = SizeEstimator.estimate(dataMapFormat)	×
50	LOGGER.debug(s"Size of message sent to Index Server: $messageSize")	×
51	}
52	val splitFolderPath = CarbonUtil	×
53	.createTempFolderForIndexServer(dataMapFormat.getCarbonTable.getTablePath,	×
54	dataMapFormat.getQueryId)	×
55	LOGGER	×
56	.info("Temp folder path for Query ID: " + dataMapFormat.getQueryId + " is " + splitFolderPath)	×
57	val (resonse, time) = logTime {	×
58	try {	×
59	val spark = SparkSQLUtil.getSparkSession	×
60	dataMapFormat.setTaskGroupId(SparkSQLUtil.getTaskGroupId(spark))	×
61	dataMapFormat.setTaskGroupDesc(SparkSQLUtil.getTaskGroupDesc(spark))	×
62	var filterInf = dataMapFormat.getFilterResolverIntf	×
63	val filterProcessor = new FilterExpressionProcessor	×
64	filterInf = removeSparkUnknown(filterInf,	×
65	dataMapFormat.getCarbonTable.getAbsoluteTableIdentifier, filterProcessor)	×
66	dataMapFormat.setFilterResolverIntf(filterInf)	×
67	IndexServer.getClient.getSplits(dataMapFormat)	×
68	.getExtendedBlockets(dataMapFormat.getCarbonTable.getTablePath, dataMapFormat.getQueryId)	×
69	} finally {	×
70	val tmpPath = CarbonUtil	×
71	.getIndexServerTempPath(dataMapFormat.getCarbonTable.getTablePath,	×
72	dataMapFormat.getQueryId)	×
73	if (null != splitFolderPath &&	×
74	!splitFolderPath.deleteFile(tmpPath, FileFactory.getFileType(tmpPath))) {	×
75	LOGGER.error("Problem while deleting the temp directory:" + tmpPath)	×
76	}
77	}
78	}
79	LOGGER.info(s"Time taken to get response from server: $time ms")	×
80	resonse	×
81	}
82
83	/**
84	* Iterate over FiltersReslover,
85	* a. Change only RowLevelFilterResolverImpl because SparkUnkown is part of it
86	* and others FilterReslover like ConditionalFilterResolverImpl so directly return.
87	* b. Change SparkUnkownExpression to TrueExpression so that isScanRequired
88	* selects block/blocklet.
89	*
90	* @param filterInf FiltersReslover to be changed
91	* @param tableIdentifer AbsoluteTableIdentifier object
92	* @param filterProcessor changed FiltersReslover.
93	* @return
94	*/
95	def removeSparkUnknown(filterInf: FilterResolverIntf,
96	tableIdentifer: AbsoluteTableIdentifier,
97	filterProcessor: FilterExpressionProcessor): FilterResolverIntf = {
98	if (filterInf.isInstanceOf[LogicalFilterResolverImpl]) {	×
99	return new LogicalFilterResolverImpl(	×
100	removeSparkUnknown(filterInf.getLeft, tableIdentifer, filterProcessor),	×
101	removeSparkUnknown(filterInf.getRight, tableIdentifer, filterProcessor),	×
102	filterProcessor.removeUnknownExpression(filterInf.getFilterExpression).	×
103	asInstanceOf[BinaryExpression])
104	}
105	if (filterInf.isInstanceOf[RowLevelFilterResolverImpl] &&	×
106	filterInf.getFilterExpression.getFilterExpressionType == ExpressionType.UNKNOWN) {	×
107	return filterProcessor.changeUnknownResloverToTrue(tableIdentifer)	×
108	}
109	return filterInf;	×
110	}
111	}
112
113	/**
114	* Spark job to execute datamap job and prune all the datamaps distributable. This job will just
115	* prune the datamaps but will not cache in executors.
116	*/
117	class EmbeddedDataMapJob extends AbstractDataMapJob {	2✔
118
119	override def execute(dataMapFormat: DistributableDataMapFormat): util.List[ExtendedBlocklet] = {
120	val spark = SparkSQLUtil.getSparkSession	2✔
121	val originalJobDesc = spark.sparkContext.getLocalProperty("spark.job.description")	2✔
122	dataMapFormat.setIsWriteToFile(false)	2✔
123	dataMapFormat.setFallbackJob()	2✔
124	val splits = IndexServer.getSplits(dataMapFormat).getExtendedBlockets(dataMapFormat	2✔
125	.getCarbonTable.getTablePath, dataMapFormat.getQueryId)	2✔
126	// Fire a job to clear the cache from executors as Embedded mode does not maintain the cache.
127	IndexServer.invalidateSegmentCache(dataMapFormat.getCarbonTable, dataMapFormat	2✔
128	.getValidSegmentIds.asScala.toArray)	2✔
129	spark.sparkContext.setLocalProperty("spark.job.description", originalJobDesc)	2✔
130	splits	2✔
131	}
132
133	}

apache / carbondata / 12039

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