4012

Build # 4012

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/bc02ec3274b1adead9843b1f13ca9b7f3d9beb62">bc02ec327<a href="https://github.com/apache/carbondata/commit/bc02ec3274b1adead9843b1f13ca9b7f3d9beb62">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/ad628fe8952254af1bbc8116204eaf4eba2cec61">ad628fe89</a>">ad628fe89</a><a href="https://github.com/apache/carbondata/commit/bc02ec3274b1adead9843b1f13ca9b7f3d9beb62"> into 8f0ec970e">8f0ec970e</a>

Run Details

14 of 14 new or added lines in 3 files covered. (100.0%)

65017 of 83661 relevant lines covered (77.71%)

1.02 hits per line

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

38.04

/integration/spark2/src/main/scala/org/apache/carbondata/indexserver/IndexServer.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.net.InetSocketAddress
import java.security.PrivilegedAction
import java.util.UUID

import scala.collection.JavaConverters._

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.ipc.{ProtocolInfo, RPC}
import org.apache.hadoop.net.NetUtils
import org.apache.hadoop.security.{KerberosInfo, SecurityUtil, UserGroupInformation}
import org.apache.spark.SparkConf
import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd}
import org.apache.spark.sql.{CarbonSession, SparkSession}
import org.apache.spark.sql.util.SparkSQLUtil

import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datamap.DistributableDataMapFormat
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.indexstore.ExtendedBlockletWrapperContainer
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.util.CarbonProperties

@ProtocolInfo(protocolName = "Server", protocolVersion = 1)
@KerberosInfo(serverPrincipal = "spark.carbon.indexserver.principal",
  clientPrincipal = "spark.carbon.indexserver.principal")
trait ServerInterface {
  /**
   * Used to prune and cache the datamaps for the table.
   */
  def getSplits(request: DistributableDataMapFormat): ExtendedBlockletWrapperContainer

  /**
   * Get the cache size for the specified tables.
   */
  def showCache(tableNames: String) : Array[String]

  /**
   * Invalidate the cache for the specified segments only. Used in case of compaction/Update/Delete.
   */
  def invalidateSegmentCache(carbonTable: CarbonTable,
      segmentIds: Array[String], jobGroupId: String = ""): Unit
}

/**
 * An instance of a distributed Index Server which will be used for:
 * 1. Pruning the datamaps in a distributed way by using the executors.
 * 2. Caching the pruned datamaps in executor size to be reused in the next query.
 * 3. Getting the size of the datamaps cached in the executors.
 * 4. Clearing the datamaps for a table or for the specified invalid segments.
 *
 * Start using ./bin/start-indexserver.sh
 * Stop using ./bin/stop-indexserver.sh
 */
object IndexServer extends ServerInterface {

  val isDistributedPruning: Boolean =
    CarbonProperties.getInstance().isDistributedPruningEnabled("", "")

  private val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)

  private val serverIp: String = CarbonProperties.getInstance().getIndexServerIP

  private lazy val serverPort: Int = CarbonProperties.getInstance().getIndexServerPort

  private val numHandlers: Int = CarbonProperties.getInstance().getNumberOfHandlersForIndexServer

  private val isExecutorLRUConfigured: Boolean =
    CarbonProperties.getInstance
      .getProperty(CarbonCommonConstants.CARBON_MAX_EXECUTOR_LRU_CACHE_SIZE) != null

  /**
   * Getting sparkSession from ActiveSession because in case of embedded mode the session would
   * have already been created whereas in case of distributed mode the session would be
   * created by the main method after some validations.
   */
  private lazy val sparkSession: SparkSession = SparkSQLUtil.getSparkSession

  private def doAs[T](f: => T): T = {
    UserGroupInformation.getLoginUser.doAs(new PrivilegedAction[T] {
      override def run(): T = {
        f
      }
    })
  }

  def getSplits(request: DistributableDataMapFormat): ExtendedBlockletWrapperContainer = {
    doAs {
      if (!request.isFallbackJob) {
        sparkSession.sparkContext.setLocalProperty("spark.jobGroup.id", request.getTaskGroupId)
        sparkSession.sparkContext
          .setLocalProperty("spark.job.description", request.getTaskGroupDesc)
      }
      val splits = new DistributedPruneRDD(sparkSession, request).collect()
      if (!request.isFallbackJob) {
        DistributedRDDUtils.updateExecutorCacheSize(splits.map(_._1).toSet)
      }
      if (request.isJobToClearDataMaps) {
        DistributedRDDUtils.invalidateTableMapping(request.getCarbonTable.getTableUniqueName)
      }
      if (!request.getInvalidSegments.isEmpty) {
        DistributedRDDUtils
          .invalidateSegmentMapping(request.getCarbonTable.getTableUniqueName,
            request.getInvalidSegments.asScala)
      }
      new ExtendedBlockletWrapperContainer(splits.map(_._2), request.isFallbackJob)
    }
  }

  override def invalidateSegmentCache(carbonTable: CarbonTable,
      segmentIds: Array[String], jobGroupId: String = ""): Unit = doAs {
    val databaseName = carbonTable.getDatabaseName
    val tableName = carbonTable.getTableName
    val jobgroup: String = " Invalided Segment Cache for " + databaseName + "." + tableName
    sparkSession.sparkContext.setLocalProperty("spark.job.description", jobgroup)
    sparkSession.sparkContext.setLocalProperty("spark.jobGroup.id", jobGroupId)
    new InvalidateSegmentCacheRDD(sparkSession, carbonTable, segmentIds.toList)
      .collect()
    if (segmentIds.nonEmpty) {
      DistributedRDDUtils
        .invalidateSegmentMapping(s"${databaseName}_$tableName",
          segmentIds)
    }
  }

  override def showCache(tableName: String = ""): Array[String] = doAs {
    val jobgroup: String = "Show Cache for " + (tableName match {
      case "" => "for all tables"
      case table => s"for $table"
    })
    sparkSession.sparkContext.setLocalProperty("spark.jobGroup.id", UUID.randomUUID().toString)
    sparkSession.sparkContext.setLocalProperty("spark.job.description", jobgroup)
    new DistributedShowCacheRDD(sparkSession, tableName).collect()
  }

  def main(args: Array[String]): Unit = {
    if (serverIp.isEmpty) {
      throw new RuntimeException(s"Please set the server IP to use Index Cache Server")
    } else if (!isDistributedPruning) {
      throw new RuntimeException(
        s"Please set ${ CarbonCommonConstants.CARBON_ENABLE_INDEX_SERVER }" +
        s" as true to use index server")
    } else if (!isExecutorLRUConfigured) {
      throw new RuntimeException(s"Executor LRU cache size is not set. Please set using " +
                                 s"${ CarbonCommonConstants.CARBON_MAX_EXECUTOR_LRU_CACHE_SIZE }")
    } else {
      createCarbonSession()
      LOGGER.info("Starting Index Cache Server")
      val conf = new Configuration()
      val server: RPC.Server = new RPC.Builder(conf).setInstance(this)
        .setBindAddress(serverIp)
        .setPort(serverPort)
        .setNumHandlers(numHandlers)
        .setProtocol(classOf[ServerInterface]).build
      server.start()
      SecurityUtil.login(sparkSession.asInstanceOf[CarbonSession].sessionState.newHadoopConf(),
        "spark.carbon.indexserver.keytab", "spark.carbon.indexserver.principal")
      sparkSession.sparkContext.addSparkListener(new SparkListener {
        override def onApplicationEnd(applicationEnd: SparkListenerApplicationEnd): Unit = {
          LOGGER.info("Spark Application has ended. Stopping the Index Server")
          server.stop()
        }
      })
      LOGGER.info(s"Index cache server running on ${ server.getPort } port")
    }
  }

  private def createCarbonSession(): SparkSession = {
    import org.apache.spark.sql.CarbonSession._
    val spark = SparkSession
      .builder().config(new SparkConf())
      .appName("DistributedIndexServer")
      .enableHiveSupport()
      .getOrCreateCarbonSession(CarbonProperties.getStorePath)
    SparkSession.setActiveSession(spark)
    SparkSession.setDefaultSession(spark)
    spark
  }

  /**
   * @return Return a new Client to communicate with the Index Server.
   */
  def getClient: ServerInterface = {
    val configuration = SparkSQLUtil.sessionState(sparkSession).newHadoopConf()
    import org.apache.hadoop.ipc.RPC
    val indexServerUser = sparkSession.sparkContext.getConf
      .get("spark.carbon.indexserver.principal", "")
    val indexServerKeyTab = sparkSession.sparkContext.getConf
      .get("spark.carbon.indexserver.keytab", "")
    val ugi = UserGroupInformation.loginUserFromKeytabAndReturnUGI(indexServerUser,
      indexServerKeyTab)
    LOGGER.info("Login successful for user " + indexServerUser)
    RPC.getProxy(classOf[ServerInterface],
      RPC.getProtocolVersion(classOf[ServerInterface]),
      new InetSocketAddress(serverIp, serverPort), ugi,
      FileFactory.getConfiguration, NetUtils.getDefaultSocketFactory(configuration))
  }
}

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.net.InetSocketAddress
20	import java.security.PrivilegedAction
21	import java.util.UUID
22
23	import scala.collection.JavaConverters._
24
25	import org.apache.hadoop.conf.Configuration
26	import org.apache.hadoop.ipc.{ProtocolInfo, RPC}
27	import org.apache.hadoop.net.NetUtils
28	import org.apache.hadoop.security.{KerberosInfo, SecurityUtil, UserGroupInformation}
29	import org.apache.spark.SparkConf
30	import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd}
31	import org.apache.spark.sql.{CarbonSession, SparkSession}
32	import org.apache.spark.sql.util.SparkSQLUtil
33
34	import org.apache.carbondata.common.logging.LogServiceFactory
35	import org.apache.carbondata.core.constants.CarbonCommonConstants
36	import org.apache.carbondata.core.datamap.DistributableDataMapFormat
37	import org.apache.carbondata.core.datastore.impl.FileFactory
38	import org.apache.carbondata.core.indexstore.ExtendedBlockletWrapperContainer
39	import org.apache.carbondata.core.metadata.schema.table.CarbonTable
40	import org.apache.carbondata.core.util.CarbonProperties
41
42	@ProtocolInfo(protocolName = "Server", protocolVersion = 1)
43	@KerberosInfo(serverPrincipal = "spark.carbon.indexserver.principal",
44	clientPrincipal = "spark.carbon.indexserver.principal")
45	trait ServerInterface {	2✔
46	/**
47	* Used to prune and cache the datamaps for the table.
48	*/
49	def getSplits(request: DistributableDataMapFormat): ExtendedBlockletWrapperContainer
50
51	/**
52	* Get the cache size for the specified tables.
53	*/
54	def showCache(tableNames: String) : Array[String]
55
56	/**
57	* Invalidate the cache for the specified segments only. Used in case of compaction/Update/Delete.
58	*/
59	def invalidateSegmentCache(carbonTable: CarbonTable,
60	segmentIds: Array[String], jobGroupId: String = ""): Unit	×
61	}
62
63	/**
64	* An instance of a distributed Index Server which will be used for:
65	* 1. Pruning the datamaps in a distributed way by using the executors.
66	* 2. Caching the pruned datamaps in executor size to be reused in the next query.
67	* 3. Getting the size of the datamaps cached in the executors.
68	* 4. Clearing the datamaps for a table or for the specified invalid segments.
69	*
70	* Start using ./bin/start-indexserver.sh
71	* Stop using ./bin/stop-indexserver.sh
72	*/
73	object IndexServer extends ServerInterface {	2✔
74
75	val isDistributedPruning: Boolean =	2✔
76	CarbonProperties.getInstance().isDistributedPruningEnabled("", "")	2✔
77
78	private val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)	2✔
79
80	private val serverIp: String = CarbonProperties.getInstance().getIndexServerIP	2✔
81
82	private lazy val serverPort: Int = CarbonProperties.getInstance().getIndexServerPort	×
83
84	private val numHandlers: Int = CarbonProperties.getInstance().getNumberOfHandlersForIndexServer	2✔
85
86	private val isExecutorLRUConfigured: Boolean =
87	CarbonProperties.getInstance	2✔
88	.getProperty(CarbonCommonConstants.CARBON_MAX_EXECUTOR_LRU_CACHE_SIZE) != null	2✔
89
90	/**
91	* Getting sparkSession from ActiveSession because in case of embedded mode the session would
92	* have already been created whereas in case of distributed mode the session would be
93	* created by the main method after some validations.
94	*/
95	private lazy val sparkSession: SparkSession = SparkSQLUtil.getSparkSession	2✔
96
97	private def doAs[T](f: => T): T = {
98	UserGroupInformation.getLoginUser.doAs(new PrivilegedAction[T] {	2✔
99	override def run(): T = {
100	f	2✔
101	}
102	})
103	}
104
105	def getSplits(request: DistributableDataMapFormat): ExtendedBlockletWrapperContainer = {
106	doAs {	2✔
107	if (!request.isFallbackJob) {	2✔
108	sparkSession.sparkContext.setLocalProperty("spark.jobGroup.id", request.getTaskGroupId)	×
109	sparkSession.sparkContext	×
110	.setLocalProperty("spark.job.description", request.getTaskGroupDesc)	×
111	}
112	val splits = new DistributedPruneRDD(sparkSession, request).collect()	2✔
113	if (!request.isFallbackJob) {	2✔
114	DistributedRDDUtils.updateExecutorCacheSize(splits.map(_._1).toSet)	×
115	}
116	if (request.isJobToClearDataMaps) {	2✔
117	DistributedRDDUtils.invalidateTableMapping(request.getCarbonTable.getTableUniqueName)	2✔
118	}
119	if (!request.getInvalidSegments.isEmpty) {	2✔
120	DistributedRDDUtils	2✔
121	.invalidateSegmentMapping(request.getCarbonTable.getTableUniqueName,	2✔
122	request.getInvalidSegments.asScala)	2✔
123	}
124	new ExtendedBlockletWrapperContainer(splits.map(_._2), request.isFallbackJob)	2✔
125	}
126	}
127
128	override def invalidateSegmentCache(carbonTable: CarbonTable,
129	segmentIds: Array[String], jobGroupId: String = ""): Unit = doAs {	2✔
130	val databaseName = carbonTable.getDatabaseName	2✔
131	val tableName = carbonTable.getTableName	2✔
132	val jobgroup: String = " Invalided Segment Cache for " + databaseName + "." + tableName	2✔
133	sparkSession.sparkContext.setLocalProperty("spark.job.description", jobgroup)	2✔
134	sparkSession.sparkContext.setLocalProperty("spark.jobGroup.id", jobGroupId)	2✔
135	new InvalidateSegmentCacheRDD(sparkSession, carbonTable, segmentIds.toList)	2✔
136	.collect()	2✔
137	if (segmentIds.nonEmpty) {	2✔
138	DistributedRDDUtils	2✔
139	.invalidateSegmentMapping(s"${databaseName}_$tableName",	2✔
140	segmentIds)	2✔
141	}
142	}
143
144	override def showCache(tableName: String = ""): Array[String] = doAs {	×
145	val jobgroup: String = "Show Cache for " + (tableName match {	×
146	case "" => "for all tables"	×
147	case table => s"for $table"	×
148	})
149	sparkSession.sparkContext.setLocalProperty("spark.jobGroup.id", UUID.randomUUID().toString)	×
150	sparkSession.sparkContext.setLocalProperty("spark.job.description", jobgroup)	×
151	new DistributedShowCacheRDD(sparkSession, tableName).collect()	×
152	}
153
154	def main(args: Array[String]): Unit = {
155	if (serverIp.isEmpty) {	×
156	throw new RuntimeException(s"Please set the server IP to use Index Cache Server")	×
157	} else if (!isDistributedPruning) {	×
158	throw new RuntimeException(	×
159	s"Please set ${ CarbonCommonConstants.CARBON_ENABLE_INDEX_SERVER }" +	×
160	s" as true to use index server")	×
161	} else if (!isExecutorLRUConfigured) {	×
162	throw new RuntimeException(s"Executor LRU cache size is not set. Please set using " +	×
163	s"${ CarbonCommonConstants.CARBON_MAX_EXECUTOR_LRU_CACHE_SIZE }")	×
164	} else {
165	createCarbonSession()	×
166	LOGGER.info("Starting Index Cache Server")	×
167	val conf = new Configuration()	×
168	val server: RPC.Server = new RPC.Builder(conf).setInstance(this)	×
169	.setBindAddress(serverIp)	×
170	.setPort(serverPort)	×
171	.setNumHandlers(numHandlers)	×
172	.setProtocol(classOf[ServerInterface]).build	×
173	server.start()	×
174	SecurityUtil.login(sparkSession.asInstanceOf[CarbonSession].sessionState.newHadoopConf(),	×
175	"spark.carbon.indexserver.keytab", "spark.carbon.indexserver.principal")	×
176	sparkSession.sparkContext.addSparkListener(new SparkListener {	×
177	override def onApplicationEnd(applicationEnd: SparkListenerApplicationEnd): Unit = {
178	LOGGER.info("Spark Application has ended. Stopping the Index Server")	×
179	server.stop()	×
180	}
181	})
182	LOGGER.info(s"Index cache server running on ${ server.getPort } port")	×
183	}
184	}
185
186	private def createCarbonSession(): SparkSession = {
187	import org.apache.spark.sql.CarbonSession._
188	val spark = SparkSession	×
189	.builder().config(new SparkConf())	×
190	.appName("DistributedIndexServer")	×
191	.enableHiveSupport()	×
192	.getOrCreateCarbonSession(CarbonProperties.getStorePath)	×
193	SparkSession.setActiveSession(spark)	×
194	SparkSession.setDefaultSession(spark)	×
195	spark	×
196	}
197
198	/**
199	* @return Return a new Client to communicate with the Index Server.
200	*/
201	def getClient: ServerInterface = {
202	val configuration = SparkSQLUtil.sessionState(sparkSession).newHadoopConf()	×
203	import org.apache.hadoop.ipc.RPC
204	val indexServerUser = sparkSession.sparkContext.getConf	×
205	.get("spark.carbon.indexserver.principal", "")	×
206	val indexServerKeyTab = sparkSession.sparkContext.getConf	×
207	.get("spark.carbon.indexserver.keytab", "")	×
208	val ugi = UserGroupInformation.loginUserFromKeytabAndReturnUGI(indexServerUser,	×
209	indexServerKeyTab)	×
210	LOGGER.info("Login successful for user " + indexServerUser)	×
211	RPC.getProxy(classOf[ServerInterface],	×
212	RPC.getProtocolVersion(classOf[ServerInterface]),	×
213	new InetSocketAddress(serverIp, serverPort), ugi,	×
214	FileFactory.getConfiguration, NetUtils.getDefaultSocketFactory(configuration))	×
215	}
216	}

apache / carbondata / 4012

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