2552

Build # 2552

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/0ce78b7f2c81744610450a4d61a99bd77e9b3267">0ce78b7f2<a href="https://github.com/apache/carbondata/commit/0ce78b7f2c81744610450a4d61a99bd77e9b3267">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/0e4b26a35b5132d50b56ff1eab173ed519e2aa7a">0e4b26a35</a>">0e4b26a35</a><a href="https://github.com/apache/carbondata/commit/0ce78b7f2c81744610450a4d61a99bd77e9b3267"> into 1a2c0513e">1a2c0513e</a>

Run Details

61 of 61 new or added lines in 6 files covered. (100.0%)

62502 of 78812 relevant lines covered (79.31%)

1.05 hits per line

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

85.31

/integration/spark-common/src/main/scala/org/apache/carbondata/streaming/StreamSinkFactory.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.streaming

import java.io.IOException
import java.util

import scala.collection.JavaConverters._

import org.apache.commons.lang3.StringUtils
import org.apache.hadoop.conf.Configuration
import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.execution.streaming.{CarbonAppendableStreamSink, Sink}

import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datastore.compression.CompressorFactory
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.dictionary.server.{DictionaryServer, NonSecureDictionaryServer}
import org.apache.carbondata.core.dictionary.service.NonSecureDictionaryServiceProvider
import org.apache.carbondata.core.locks.{CarbonLockFactory, ICarbonLock, LockUsage}
import org.apache.carbondata.core.metadata.encoder.Encoding
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.core.util.path.CarbonTablePath
import org.apache.carbondata.events.{OperationContext, OperationListenerBus}
import org.apache.carbondata.processing.loading.events.LoadEvents.{LoadMetadataEvent, LoadTablePostExecutionEvent, LoadTablePreExecutionEvent, LoadTablePreStatusUpdateEvent}
import org.apache.carbondata.processing.loading.model.{CarbonLoadModel, CarbonLoadModelBuilder, LoadOption}
import org.apache.carbondata.processing.util.CarbonBadRecordUtil
import org.apache.carbondata.spark.dictionary.provider.SecureDictionaryServiceProvider
import org.apache.carbondata.spark.dictionary.server.SecureDictionaryServer
import org.apache.carbondata.streaming.segment.StreamSegment

/**
 * Stream sink factory
 */
object StreamSinkFactory {

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

  val locks = new util.concurrent.ConcurrentHashMap[String, ICarbonLock]()

  def lock(carbonTable: CarbonTable): Unit = {
    val lock = CarbonLockFactory.getCarbonLockObj(carbonTable.getAbsoluteTableIdentifier,
      LockUsage.STREAMING_LOCK)
    if (lock.lockWithRetries()) {
      locks.put(carbonTable.getTableUniqueName, lock)
      LOGGER.info("Acquired the streaming lock for stream table: " + carbonTable.getDatabaseName +
                  "." + carbonTable.getTableName)
    } else {
      LOGGER.error("Not able to acquire the streaming lock for stream table:" +
        carbonTable.getDatabaseName + "." + carbonTable.getTableName)
      throw new IOException(
        "Not able to acquire the streaming lock for stream table: " +
        carbonTable.getDatabaseName + "." + carbonTable.getTableName)
    }
  }

  def unLock(tableUniqueName: String): Unit = {
    val lock = locks.remove(tableUniqueName)
    if (lock != null) {
      lock.unlock()
    }
  }

  def createStreamTableSink(
      sparkSession: SparkSession,
      hadoopConf: Configuration,
      carbonTable: CarbonTable,
      parameters: Map[String, String]): Sink = {

    lock(carbonTable)

    validateParameters(parameters)

    // build load model
    val carbonLoadModel = buildCarbonLoadModelForStream(
      sparkSession,
      hadoopConf,
      carbonTable,
      parameters,
      "")
    // fire pre event before streamin is started
    // in case of streaming options and optionsFinal can be same
    val operationContext = new OperationContext
    val loadTablePreExecutionEvent = new LoadTablePreExecutionEvent(
      carbonTable.getCarbonTableIdentifier,
      carbonLoadModel
      )
    OperationListenerBus.getInstance().fireEvent(loadTablePreExecutionEvent, operationContext)
    // prepare the stream segment
    val segmentId = getStreamSegmentId(carbonTable)
    carbonLoadModel.setSegmentId(segmentId)

    // Used to generate load commands for child tables in case auto-handoff is fired.
    val loadMetaEvent = new LoadMetadataEvent(carbonTable, false, parameters.asJava)
    OperationListenerBus.getInstance().fireEvent(loadMetaEvent, operationContext)
    // start server if necessary
    val server = startDictionaryServer(
      sparkSession,
      carbonTable,
      carbonLoadModel)
    if (server.isDefined) {
      carbonLoadModel.setUseOnePass(true)
    } else {
      carbonLoadModel.setUseOnePass(false)
    }

    // default is carbon appended stream sink
    val carbonAppendableStreamSink = new CarbonAppendableStreamSink(
      sparkSession,
      carbonTable,
      segmentId,
      parameters,
      carbonLoadModel,
      server,
      operationContext)

    // fire post event before streamin is started
    val loadTablePostExecutionEvent = new LoadTablePostExecutionEvent(
      carbonTable.getCarbonTableIdentifier,
      carbonLoadModel
    )
    OperationListenerBus.getInstance().fireEvent(loadTablePostExecutionEvent, operationContext)
    carbonAppendableStreamSink
  }

  private def validateParameters(parameters: Map[String, String]): Unit = {
    val segmentSize = parameters.get(CarbonCommonConstants.HANDOFF_SIZE)
    if (segmentSize.isDefined) {
      try {
        val value = java.lang.Long.parseLong(segmentSize.get)
        if (value < CarbonCommonConstants.HANDOFF_SIZE_MIN) {
          new CarbonStreamException(CarbonCommonConstants.HANDOFF_SIZE +
                                    "should be bigger than or equal " +
                                    CarbonCommonConstants.HANDOFF_SIZE_MIN)
        }
      } catch {
        case _: NumberFormatException =>
          new CarbonStreamException(CarbonCommonConstants.HANDOFF_SIZE +
                                    s" $segmentSize is an illegal number")
      }
    }
  }

  /**
   * get current stream segment id
   * @return
   */
  private def getStreamSegmentId(carbonTable: CarbonTable): String = {
    val segmentId = StreamSegment.open(carbonTable)
    val segmentDir = CarbonTablePath.getSegmentPath(carbonTable.getTablePath, segmentId)
    val fileType = FileFactory.getFileType(segmentDir)
    val metadataPath = CarbonTablePath.getMetadataPath(carbonTable.getTablePath)
    if (!FileFactory.isFileExist(metadataPath, fileType)) {
      // Create table directory path, in case of enabling hive metastore first load may not have
      // table folder created.
      FileFactory.mkdirs(metadataPath, fileType)
    }
    if (FileFactory.isFileExist(segmentDir, fileType)) {
      // recover fault
      StreamSegment.recoverSegmentIfRequired(segmentDir)
    } else {
      FileFactory.mkdirs(segmentDir, fileType)
    }
    segmentId
  }

  def startDictionaryServer(
      sparkSession: SparkSession,
      carbonTable: CarbonTable,
      carbonLoadModel: CarbonLoadModel): Option[DictionaryServer] = {
    // start dictionary server when use one pass load and dimension with DICTIONARY
    // encoding is present.
    val allDimensions = carbonTable.getAllDimensions.asScala.toList
    val createDictionary = allDimensions.exists {
      carbonDimension => carbonDimension.hasEncoding(Encoding.DICTIONARY) &&
                         !carbonDimension.hasEncoding(Encoding.DIRECT_DICTIONARY)
    }
    val carbonSecureModeDictServer = CarbonProperties.getInstance.
      getProperty(CarbonCommonConstants.CARBON_SECURE_DICTIONARY_SERVER,
        CarbonCommonConstants.CARBON_SECURE_DICTIONARY_SERVER_DEFAULT)

    val sparkConf = sparkSession.sqlContext.sparkContext.getConf
    val sparkDriverHost = sparkSession.sqlContext.sparkContext.
      getConf.get("spark.driver.host")

    val server: Option[DictionaryServer] = if (createDictionary) {
      if (sparkConf.get("spark.authenticate", "false").equalsIgnoreCase("true") &&
          carbonSecureModeDictServer.toBoolean) {
        val dictionaryServer = SecureDictionaryServer.getInstance(sparkConf,
          sparkDriverHost.toString, carbonLoadModel.getDictionaryServerPort, carbonTable)
        carbonLoadModel.setDictionaryServerPort(dictionaryServer.getPort)
        carbonLoadModel.setDictionaryServerHost(dictionaryServer.getHost)
        carbonLoadModel.setDictionaryServerSecretKey(dictionaryServer.getSecretKey)
        carbonLoadModel.setDictionaryEncryptServerSecure(dictionaryServer.isEncryptSecureServer)
        carbonLoadModel.setDictionaryServiceProvider(new SecureDictionaryServiceProvider())
        sparkSession.sparkContext.addSparkListener(new SparkListener() {
          override def onApplicationEnd(applicationEnd: SparkListenerApplicationEnd) {
            dictionaryServer.shutdown()
          }
        })
        Some(dictionaryServer)
      } else {
        val dictionaryServer = NonSecureDictionaryServer
          .getInstance(carbonLoadModel.getDictionaryServerPort, carbonTable)
        carbonLoadModel.setDictionaryServerPort(dictionaryServer.getPort)
        carbonLoadModel.setDictionaryServerHost(dictionaryServer.getHost)
        carbonLoadModel.setDictionaryEncryptServerSecure(false)
        carbonLoadModel
          .setDictionaryServiceProvider(new NonSecureDictionaryServiceProvider(dictionaryServer
            .getPort))
        sparkSession.sparkContext.addSparkListener(new SparkListener() {
          override def onApplicationEnd(applicationEnd: SparkListenerApplicationEnd) {
            dictionaryServer.shutdown()
          }
        })
        Some(dictionaryServer)
      }
    } else {
      None
    }
    server
  }

  private def buildCarbonLoadModelForStream(
      sparkSession: SparkSession,
      hadoopConf: Configuration,
      carbonTable: CarbonTable,
      parameters: Map[String, String],
      segmentId: String): CarbonLoadModel = {
    val carbonProperty: CarbonProperties = CarbonProperties.getInstance()
    carbonProperty.addProperty("zookeeper.enable.lock", "false")
    val optionsFinal = LoadOption.fillOptionWithDefaultValue(parameters.asJava)
    optionsFinal.put("sort_scope", "no_sort")
    if (parameters.get("fileheader").isEmpty) {
      optionsFinal.put("fileheader", carbonTable.getCreateOrderColumn(carbonTable.getTableName)
        .asScala.map(_.getColName).mkString(","))
    }
    optionsFinal
      .put("bad_record_path", CarbonBadRecordUtil.getBadRecordsPath(parameters.asJava, carbonTable))
    val carbonLoadModel = new CarbonLoadModel()
    new CarbonLoadModelBuilder(carbonTable).build(
      parameters.asJava,
      optionsFinal,
      carbonLoadModel,
      hadoopConf)
    carbonLoadModel.setSegmentId(segmentId)
    // stream should use one pass
    val dictionaryServerPort = parameters.getOrElse(
      CarbonCommonConstants.DICTIONARY_SERVER_PORT,
      carbonProperty.getProperty(
        CarbonCommonConstants.DICTIONARY_SERVER_PORT,
        CarbonCommonConstants.DICTIONARY_SERVER_PORT_DEFAULT))
    val sparkDriverHost = sparkSession.sqlContext.sparkContext.
      getConf.get("spark.driver.host")
    carbonLoadModel.setDictionaryServerHost(sparkDriverHost)
    carbonLoadModel.setDictionaryServerPort(dictionaryServerPort.toInt)
    val columnCompressor = carbonTable.getTableInfo.getFactTable.getTableProperties.asScala
      .getOrElse(CarbonCommonConstants.COMPRESSOR,
        CompressorFactory.getInstance().getCompressor.getName)
    carbonLoadModel.setColumnCompressor(columnCompressor)
    carbonLoadModel
  }
}

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.carbondata.streaming
19
20	import java.io.IOException
21	import java.util
22
23	import scala.collection.JavaConverters._
24
25	import org.apache.commons.lang3.StringUtils
26	import org.apache.hadoop.conf.Configuration
27	import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd}
28	import org.apache.spark.sql.SparkSession
29	import org.apache.spark.sql.execution.streaming.{CarbonAppendableStreamSink, Sink}
30
31	import org.apache.carbondata.common.logging.LogServiceFactory
32	import org.apache.carbondata.core.constants.CarbonCommonConstants
33	import org.apache.carbondata.core.datastore.compression.CompressorFactory
34	import org.apache.carbondata.core.datastore.impl.FileFactory
35	import org.apache.carbondata.core.dictionary.server.{DictionaryServer, NonSecureDictionaryServer}
36	import org.apache.carbondata.core.dictionary.service.NonSecureDictionaryServiceProvider
37	import org.apache.carbondata.core.locks.{CarbonLockFactory, ICarbonLock, LockUsage}
38	import org.apache.carbondata.core.metadata.encoder.Encoding
39	import org.apache.carbondata.core.metadata.schema.table.CarbonTable
40	import org.apache.carbondata.core.util.CarbonProperties
41	import org.apache.carbondata.core.util.path.CarbonTablePath
42	import org.apache.carbondata.events.{OperationContext, OperationListenerBus}
43	import org.apache.carbondata.processing.loading.events.LoadEvents.{LoadMetadataEvent, LoadTablePostExecutionEvent, LoadTablePreExecutionEvent, LoadTablePreStatusUpdateEvent}
44	import org.apache.carbondata.processing.loading.model.{CarbonLoadModel, CarbonLoadModelBuilder, LoadOption}
45	import org.apache.carbondata.processing.util.CarbonBadRecordUtil
46	import org.apache.carbondata.spark.dictionary.provider.SecureDictionaryServiceProvider
47	import org.apache.carbondata.spark.dictionary.server.SecureDictionaryServer
48	import org.apache.carbondata.streaming.segment.StreamSegment
49
50	/**
51	* Stream sink factory
52	*/
53	object StreamSinkFactory {
54
55	val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)	1✔
56
57	val locks = new util.concurrent.ConcurrentHashMap[String, ICarbonLock]()	1✔
58
59	def lock(carbonTable: CarbonTable): Unit = {	1✔
60	val lock = CarbonLockFactory.getCarbonLockObj(carbonTable.getAbsoluteTableIdentifier,	1✔
61	LockUsage.STREAMING_LOCK)	1✔
62	if (lock.lockWithRetries()) {	1✔
63	locks.put(carbonTable.getTableUniqueName, lock)	1✔
64	LOGGER.info("Acquired the streaming lock for stream table: " + carbonTable.getDatabaseName +	1✔
65	"." + carbonTable.getTableName)	1✔
66	} else {
67	LOGGER.error("Not able to acquire the streaming lock for stream table:" +	×
68	carbonTable.getDatabaseName + "." + carbonTable.getTableName)	×
69	throw new IOException(	×
70	"Not able to acquire the streaming lock for stream table: " +	×
71	carbonTable.getDatabaseName + "." + carbonTable.getTableName)	×
72	}
73	}
74
75	def unLock(tableUniqueName: String): Unit = {	1✔
76	val lock = locks.remove(tableUniqueName)	1✔
77	if (lock != null) {	1✔
78	lock.unlock()	1✔
79	}
80	}
81
82	def createStreamTableSink(
83	sparkSession: SparkSession,
84	hadoopConf: Configuration,
85	carbonTable: CarbonTable,
86	parameters: Map[String, String]): Sink = {
87
88	lock(carbonTable)	1✔
89
90	validateParameters(parameters)	1✔
91
92	// build load model
93	val carbonLoadModel = buildCarbonLoadModelForStream(	1✔
94	sparkSession,	1✔
95	hadoopConf,	1✔
96	carbonTable,	1✔
97	parameters,	1✔
98	"")	1✔
99	// fire pre event before streamin is started
100	// in case of streaming options and optionsFinal can be same
101	val operationContext = new OperationContext	1✔
102	val loadTablePreExecutionEvent = new LoadTablePreExecutionEvent(	1✔
103	carbonTable.getCarbonTableIdentifier,	1✔
104	carbonLoadModel	1✔
105	)
106	OperationListenerBus.getInstance().fireEvent(loadTablePreExecutionEvent, operationContext)	1✔
107	// prepare the stream segment
108	val segmentId = getStreamSegmentId(carbonTable)	1✔
109	carbonLoadModel.setSegmentId(segmentId)	1✔
110
111	// Used to generate load commands for child tables in case auto-handoff is fired.
112	val loadMetaEvent = new LoadMetadataEvent(carbonTable, false, parameters.asJava)	1✔
113	OperationListenerBus.getInstance().fireEvent(loadMetaEvent, operationContext)	1✔
114	// start server if necessary
115	val server = startDictionaryServer(	1✔
116	sparkSession,	1✔
117	carbonTable,	1✔
118	carbonLoadModel)	1✔
119	if (server.isDefined) {	1✔
120	carbonLoadModel.setUseOnePass(true)	1✔
121	} else {
122	carbonLoadModel.setUseOnePass(false)	1✔
123	}
124
125	// default is carbon appended stream sink
126	val carbonAppendableStreamSink = new CarbonAppendableStreamSink(	1✔
127	sparkSession,	1✔
128	carbonTable,	1✔
129	segmentId,	1✔
130	parameters,	1✔
131	carbonLoadModel,	1✔
132	server,	1✔
133	operationContext)	1✔
134
135	// fire post event before streamin is started
136	val loadTablePostExecutionEvent = new LoadTablePostExecutionEvent(	1✔
137	carbonTable.getCarbonTableIdentifier,	1✔
138	carbonLoadModel	1✔
139	)
140	OperationListenerBus.getInstance().fireEvent(loadTablePostExecutionEvent, operationContext)	1✔
141	carbonAppendableStreamSink	1✔
142	}
143
144	private def validateParameters(parameters: Map[String, String]): Unit = {	1✔
145	val segmentSize = parameters.get(CarbonCommonConstants.HANDOFF_SIZE)	1✔
146	if (segmentSize.isDefined) {	1✔
147	try {	×
148	val value = java.lang.Long.parseLong(segmentSize.get)	1✔
149	if (value < CarbonCommonConstants.HANDOFF_SIZE_MIN) {	1✔
150	new CarbonStreamException(CarbonCommonConstants.HANDOFF_SIZE +	1✔
151	"should be bigger than or equal " +	1✔
152	CarbonCommonConstants.HANDOFF_SIZE_MIN)	1✔
153	}
154	} catch {
155	case _: NumberFormatException =>
156	new CarbonStreamException(CarbonCommonConstants.HANDOFF_SIZE +	×
157	s" $segmentSize is an illegal number")	×
158	}
159	}
160	}
161
162	/**
163	* get current stream segment id
164	* @return
165	*/
166	private def getStreamSegmentId(carbonTable: CarbonTable): String = {
167	val segmentId = StreamSegment.open(carbonTable)	1✔
168	val segmentDir = CarbonTablePath.getSegmentPath(carbonTable.getTablePath, segmentId)	1✔
169	val fileType = FileFactory.getFileType(segmentDir)	1✔
170	val metadataPath = CarbonTablePath.getMetadataPath(carbonTable.getTablePath)	1✔
171	if (!FileFactory.isFileExist(metadataPath, fileType)) {	1✔
172	// Create table directory path, in case of enabling hive metastore first load may not have
173	// table folder created.
174	FileFactory.mkdirs(metadataPath, fileType)	×
175	}
176	if (FileFactory.isFileExist(segmentDir, fileType)) {	1✔
177	// recover fault
178	StreamSegment.recoverSegmentIfRequired(segmentDir)	1✔
179	} else {
180	FileFactory.mkdirs(segmentDir, fileType)	1✔
181	}
182	segmentId	1✔
183	}
184
185	def startDictionaryServer(
186	sparkSession: SparkSession,
187	carbonTable: CarbonTable,
188	carbonLoadModel: CarbonLoadModel): Option[DictionaryServer] = {
189	// start dictionary server when use one pass load and dimension with DICTIONARY
190	// encoding is present.
191	val allDimensions = carbonTable.getAllDimensions.asScala.toList	1✔
192	val createDictionary = allDimensions.exists {	1✔
193	carbonDimension => carbonDimension.hasEncoding(Encoding.DICTIONARY) &&	1✔
194	!carbonDimension.hasEncoding(Encoding.DIRECT_DICTIONARY)	1✔
195	}
196	val carbonSecureModeDictServer = CarbonProperties.getInstance.	1✔
197	getProperty(CarbonCommonConstants.CARBON_SECURE_DICTIONARY_SERVER,	1✔
198	CarbonCommonConstants.CARBON_SECURE_DICTIONARY_SERVER_DEFAULT)	1✔
199
200	val sparkConf = sparkSession.sqlContext.sparkContext.getConf	1✔
201	val sparkDriverHost = sparkSession.sqlContext.sparkContext.	1✔
202	getConf.get("spark.driver.host")	1✔
203
204	val server: Option[DictionaryServer] = if (createDictionary) {	1✔
205	if (sparkConf.get("spark.authenticate", "false").equalsIgnoreCase("true") &&	1✔
206	carbonSecureModeDictServer.toBoolean) {	×
207	val dictionaryServer = SecureDictionaryServer.getInstance(sparkConf,	×
208	sparkDriverHost.toString, carbonLoadModel.getDictionaryServerPort, carbonTable)	×
209	carbonLoadModel.setDictionaryServerPort(dictionaryServer.getPort)	×
210	carbonLoadModel.setDictionaryServerHost(dictionaryServer.getHost)	×
211	carbonLoadModel.setDictionaryServerSecretKey(dictionaryServer.getSecretKey)	×
212	carbonLoadModel.setDictionaryEncryptServerSecure(dictionaryServer.isEncryptSecureServer)	×
213	carbonLoadModel.setDictionaryServiceProvider(new SecureDictionaryServiceProvider())	×
214	sparkSession.sparkContext.addSparkListener(new SparkListener() {	×
215	override def onApplicationEnd(applicationEnd: SparkListenerApplicationEnd) {
216	dictionaryServer.shutdown()	×
217	}
218	})
219	Some(dictionaryServer)	×
220	} else {
221	val dictionaryServer = NonSecureDictionaryServer	1✔
222	.getInstance(carbonLoadModel.getDictionaryServerPort, carbonTable)	1✔
223	carbonLoadModel.setDictionaryServerPort(dictionaryServer.getPort)	1✔
224	carbonLoadModel.setDictionaryServerHost(dictionaryServer.getHost)	1✔
225	carbonLoadModel.setDictionaryEncryptServerSecure(false)	1✔
226	carbonLoadModel	1✔
227	.setDictionaryServiceProvider(new NonSecureDictionaryServiceProvider(dictionaryServer	1✔
228	.getPort))	1✔
229	sparkSession.sparkContext.addSparkListener(new SparkListener() {	1✔
230	override def onApplicationEnd(applicationEnd: SparkListenerApplicationEnd) {
231	dictionaryServer.shutdown()	×
232	}
233	})
234	Some(dictionaryServer)	1✔
235	}
236	} else {
237	None	1✔
238	}
239	server	1✔
240	}
241
242	private def buildCarbonLoadModelForStream(
243	sparkSession: SparkSession,
244	hadoopConf: Configuration,
245	carbonTable: CarbonTable,
246	parameters: Map[String, String],
247	segmentId: String): CarbonLoadModel = {
248	val carbonProperty: CarbonProperties = CarbonProperties.getInstance()	1✔
249	carbonProperty.addProperty("zookeeper.enable.lock", "false")	1✔
250	val optionsFinal = LoadOption.fillOptionWithDefaultValue(parameters.asJava)	1✔
251	optionsFinal.put("sort_scope", "no_sort")	1✔
252	if (parameters.get("fileheader").isEmpty) {	1✔
253	optionsFinal.put("fileheader", carbonTable.getCreateOrderColumn(carbonTable.getTableName)	1✔
254	.asScala.map(_.getColName).mkString(","))	1✔
255	}
256	optionsFinal	1✔
257	.put("bad_record_path", CarbonBadRecordUtil.getBadRecordsPath(parameters.asJava, carbonTable))	1✔
258	val carbonLoadModel = new CarbonLoadModel()	1✔
259	new CarbonLoadModelBuilder(carbonTable).build(	1✔
260	parameters.asJava,	1✔
261	optionsFinal,	1✔
262	carbonLoadModel,	1✔
263	hadoopConf)	1✔
264	carbonLoadModel.setSegmentId(segmentId)	1✔
265	// stream should use one pass
266	val dictionaryServerPort = parameters.getOrElse(	1✔
267	CarbonCommonConstants.DICTIONARY_SERVER_PORT,	1✔
268	carbonProperty.getProperty(	1✔
269	CarbonCommonConstants.DICTIONARY_SERVER_PORT,	1✔
270	CarbonCommonConstants.DICTIONARY_SERVER_PORT_DEFAULT))	1✔
271	val sparkDriverHost = sparkSession.sqlContext.sparkContext.	1✔
272	getConf.get("spark.driver.host")	1✔
273	carbonLoadModel.setDictionaryServerHost(sparkDriverHost)	1✔
274	carbonLoadModel.setDictionaryServerPort(dictionaryServerPort.toInt)	1✔
275	val columnCompressor = carbonTable.getTableInfo.getFactTable.getTableProperties.asScala	1✔
276	.getOrElse(CarbonCommonConstants.COMPRESSOR,	1✔
277	CompressorFactory.getInstance().getCompressor.getName)	1✔
278	carbonLoadModel.setColumnCompressor(columnCompressor)	1✔
279	carbonLoadModel	1✔
280	}
281	}	1✔

apache / carbondata / 2552

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