11607

Build # 11607

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/72dc4eb9fb029eca9bf650704daef0cf85ced07f">72dc4eb9f<a href="https://github.com/apache/carbondata/commit/72dc4eb9fb029eca9bf650704daef0cf85ced07f">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/be6b238d4cd6c0e93531f894252cc4a6139b9537">be6b238d4</a>">be6b238d4</a><a href="https://github.com/apache/carbondata/commit/72dc4eb9fb029eca9bf650704daef0cf85ced07f"> into 7541ef25f">7541ef25f</a>

Run Details

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

64128 of 81950 relevant lines covered (78.25%)

1.04 hits per line

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

20.63

/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 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.ExtendedBlocklet
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): Array[ExtendedBlocklet]

  /**
   * Invalidate the cache for the provided table.
   */
  def invalidateCache(request: DistributableDataMapFormat): Unit

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

  /**
   * Invalidate the cache for the specified segments only. Used in case of compaction/Update/Delete.
   */
  def invalidateSegmentCache(databaseName: String,
      tableName: String,
      segmentIds: Array[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

  /**
   * 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): Array[ExtendedBlocklet] = doAs {
    val splits = new DistributedPruneRDD(sparkSession, request).collect()
    DistributedRDDUtils.updateExecutorCacheSize(splits.map(_._1).toSet)
    splits.map(_._2)
  }

  override def invalidateCache(request: DistributableDataMapFormat): Unit = doAs {
    val splits = new DistributedPruneRDD(sparkSession, request).collect()
    DistributedRDDUtils.updateExecutorCacheSize(splits.map(_._1).toSet)
    if (request.isJobToClearDataMaps) {
      DistributedRDDUtils.invalidateCache(request.getCarbonTable.getTableUniqueName)
    }
  }

  override def invalidateSegmentCache(databaseName: String, tableName: String,
      segmentIds: Array[String]): Unit = doAs {
    new InvalidateSegmentCacheRDD(sparkSession, databaseName, tableName, segmentIds.toList)
      .collect()
  }

  override def showCache(tableName: String = ""): Array[String] = doAs {
    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 {
      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 = {
    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)
    RPC.getProxy(classOf[ServerInterface],
      RPC.getProtocolVersion(classOf[ServerInterface]),
      new InetSocketAddress(serverIp, serverPort), ugi,
      FileFactory.getConfiguration, NetUtils.getDefaultSocketFactory(FileFactory.getConfiguration))
  }
}

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

apache / carbondata / 11607

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