apache / carbondata / 119

Build # 119

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/423672611cf00fa9e63d701d692aa92dc4dadfec">423672611<a href="https://github.com/apache/carbondata/commit/423672611cf00fa9e63d701d692aa92dc4dadfec">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/apache/carbondata/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/423672611cf00fa9e63d701d692aa92dc4dadfec&quot;&gt;423672611&lt;/a&gt;&lt;a href=&quot;https://github.com/apache/carbondata/commit/423672611cf00fa9e63d701d692aa92dc4dadfec&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/apache/carbondata/commit/423672611cf00fa9e63d701d692aa92d&lt;/a&gt;c4dadfec">&quot;&gt;Merge &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/c14a937f4654acc05d507c57c745f309d64dd027">c14a937f4</a><a href="https://github.com/apache/carbondata/commit/423672611cf00fa9e63d701d692aa92dc4dadfec">&amp;quot;&amp;gt;c14a937f4&amp;lt;/a&amp;gt;&amp;quot;&amp;gt;c14a937f4&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/apache/carbondata/commit/423672611cf00fa9e63d701d692aa92dc4dad&lt;/a&gt;fec&quot;&gt; into &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/c65cc125e">c65cc125e">c65cc125e</a>

Run Details

5 of 5 new or added lines in 1 file covered. (100.0%)

63962 of 82296 relevant lines covered (77.72%)

1.02 hits per line

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

37.5

/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 {

  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 (!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()
        }
      })
      CarbonProperties.getInstance().addProperty(CarbonCommonConstants
        .CARBON_ENABLE_INDEX_SERVER, "true")
      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	private val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)	2✔
76
77	private val serverIp: String = CarbonProperties.getInstance().getIndexServerIP	2✔
78
79	private lazy val serverPort: Int = CarbonProperties.getInstance().getIndexServerPort	×
80
81	private val numHandlers: Int = CarbonProperties.getInstance().getNumberOfHandlersForIndexServer	2✔
82
83	private val isExecutorLRUConfigured: Boolean =
84	CarbonProperties.getInstance	2✔
85	.getProperty(CarbonCommonConstants.CARBON_MAX_EXECUTOR_LRU_CACHE_SIZE) != null	2✔
86
87	/**
88	* Getting sparkSession from ActiveSession because in case of embedded mode the session would
89	* have already been created whereas in case of distributed mode the session would be
90	* created by the main method after some validations.
91	*/
92	private lazy val sparkSession: SparkSession = SparkSQLUtil.getSparkSession	2✔
93
94	private def doAs[T](f: => T): T = {
95	UserGroupInformation.getLoginUser.doAs(new PrivilegedAction[T] {	2✔
96	override def run(): T = {
97	f	2✔
98	}
99	})
100	}
101
102	def getSplits(request: DistributableDataMapFormat): ExtendedBlockletWrapperContainer = {
103	doAs {	2✔
104	if (!request.isFallbackJob) {	2✔
105	sparkSession.sparkContext.setLocalProperty("spark.jobGroup.id", request.getTaskGroupId)	×
106	sparkSession.sparkContext	×
107	.setLocalProperty("spark.job.description", request.getTaskGroupDesc)	×
108	}
109	val splits = new DistributedPruneRDD(sparkSession, request).collect()	2✔
110	if (!request.isFallbackJob) {	2✔
111	DistributedRDDUtils.updateExecutorCacheSize(splits.map(_._1).toSet)	×
112	}
113	if (request.isJobToClearDataMaps) {	2✔
114	DistributedRDDUtils.invalidateTableMapping(request.getCarbonTable.getTableUniqueName)	2✔
115	}
116	if (!request.getInvalidSegments.isEmpty) {	2✔
117	DistributedRDDUtils	2✔
118	.invalidateSegmentMapping(request.getCarbonTable.getTableUniqueName,	2✔
119	request.getInvalidSegments.asScala)	2✔
120	}
121	new ExtendedBlockletWrapperContainer(splits.map(_._2), request.isFallbackJob)	2✔
122	}
123	}
124
125	override def invalidateSegmentCache(carbonTable: CarbonTable,
126	segmentIds: Array[String], jobGroupId: String = ""): Unit = doAs {	2✔
127	val databaseName = carbonTable.getDatabaseName	2✔
128	val tableName = carbonTable.getTableName	2✔
129	val jobgroup: String = " Invalided Segment Cache for " + databaseName + "." + tableName	2✔
130	sparkSession.sparkContext.setLocalProperty("spark.job.description", jobgroup)	2✔
131	sparkSession.sparkContext.setLocalProperty("spark.jobGroup.id", jobGroupId)	2✔
132	new InvalidateSegmentCacheRDD(sparkSession, carbonTable, segmentIds.toList)	2✔
133	.collect()	2✔
134	if (segmentIds.nonEmpty) {	2✔
135	DistributedRDDUtils	2✔
136	.invalidateSegmentMapping(s"${databaseName}_$tableName",	2✔
137	segmentIds)	2✔
138	}
139	}
140
141	override def showCache(tableName: String = ""): Array[String] = doAs {	×
142	val jobgroup: String = "Show Cache for " + (tableName match {	×
143	case "" => "for all tables"	×
144	case table => s"for $table"	×
145	})
146	sparkSession.sparkContext.setLocalProperty("spark.jobGroup.id", UUID.randomUUID().toString)	×
147	sparkSession.sparkContext.setLocalProperty("spark.job.description", jobgroup)	×
148	new DistributedShowCacheRDD(sparkSession, tableName).collect()	×
149	}
150
151	def main(args: Array[String]): Unit = {
152	if (serverIp.isEmpty) {	×
153	throw new RuntimeException(s"Please set the server IP to use Index Cache Server")	×
154	} else if (!isExecutorLRUConfigured) {	×
155	throw new RuntimeException(s"Executor LRU cache size is not set. Please set using " +	×
156	s"${ CarbonCommonConstants.CARBON_MAX_EXECUTOR_LRU_CACHE_SIZE }")	×
157	} else {
158	createCarbonSession()	×
159	LOGGER.info("Starting Index Cache Server")	×
160	val conf = new Configuration()	×
161	val server: RPC.Server = new RPC.Builder(conf).setInstance(this)	×
162	.setBindAddress(serverIp)	×
163	.setPort(serverPort)	×
164	.setNumHandlers(numHandlers)	×
165	.setProtocol(classOf[ServerInterface]).build	×
166	server.start()	×
167	SecurityUtil.login(sparkSession.asInstanceOf[CarbonSession].sessionState.newHadoopConf(),	×
168	"spark.carbon.indexserver.keytab", "spark.carbon.indexserver.principal")	×
169	sparkSession.sparkContext.addSparkListener(new SparkListener {	×
170	override def onApplicationEnd(applicationEnd: SparkListenerApplicationEnd): Unit = {
171	LOGGER.info("Spark Application has ended. Stopping the Index Server")	×
172	server.stop()	×
173	}
174	})
175	CarbonProperties.getInstance().addProperty(CarbonCommonConstants	×
176	.CARBON_ENABLE_INDEX_SERVER, "true")	×
177	LOGGER.info(s"Index cache server running on ${ server.getPort } port")	×
178	}
179	}
180
181	private def createCarbonSession(): SparkSession = {
182	import org.apache.spark.sql.CarbonSession._
183	val spark = SparkSession	×
184	.builder().config(new SparkConf())	×
185	.appName("DistributedIndexServer")	×
186	.enableHiveSupport()	×
187	.getOrCreateCarbonSession(CarbonProperties.getStorePath)	×
188	SparkSession.setActiveSession(spark)	×
189	SparkSession.setDefaultSession(spark)	×
190	spark	×
191	}
192
193	/**
194	* @return Return a new Client to communicate with the Index Server.
195	*/
196	def getClient: ServerInterface = {
197	val configuration = SparkSQLUtil.sessionState(sparkSession).newHadoopConf()	×
198	import org.apache.hadoop.ipc.RPC
199	val indexServerUser = sparkSession.sparkContext.getConf	×
200	.get("spark.carbon.indexserver.principal", "")	×
201	val indexServerKeyTab = sparkSession.sparkContext.getConf	×
202	.get("spark.carbon.indexserver.keytab", "")	×
203	val ugi = UserGroupInformation.loginUserFromKeytabAndReturnUGI(indexServerUser,	×
204	indexServerKeyTab)	×
205	LOGGER.info("Login successful for user " + indexServerUser)	×
206	RPC.getProxy(classOf[ServerInterface],	×
207	RPC.getProtocolVersion(classOf[ServerInterface]),	×
208	new InetSocketAddress(serverIp, serverPort), ugi,	×
209	FileFactory.getConfiguration, NetUtils.getDefaultSocketFactory(configuration))	×
210	}
211	}

apache / carbondata / 119

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