apache / carbondata / 18

Build # 18

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/038509b639dcde120e7446f10581ee8f0e47d955">038509b63<a href="https://github.com/apache/carbondata/commit/038509b639dcde120e7446f10581ee8f0e47d955">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/9a7f9e1eb59320b55eaa833ab9c9ff8135a585fe">9a7f9e1eb</a>">9a7f9e1eb</a><a href="https://github.com/apache/carbondata/commit/038509b639dcde120e7446f10581ee8f0e47d955"> into f81543e95">f81543e95</a>

Run Details

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

57357 of 71103 relevant lines covered (80.67%)

1.07 hits per line

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

86.24

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

import java.text.SimpleDateFormat
import java.util
import java.util.{Date, UUID}

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.mapreduce.{Job, TaskAttemptID, TaskType}
import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl
import org.apache.spark.{Partition, SerializableWritable, SparkContext, TaskContext}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.expressions.GenericInternalRow

import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.converter.SparkDataTypeConverterImpl
import org.apache.carbondata.core.datamap.Segment
import org.apache.carbondata.core.datastore.block.SegmentProperties
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.locks.{CarbonLockFactory, LockUsage}
import org.apache.carbondata.core.metadata.CarbonMetadata
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.scan.result.iterator.RawResultIterator
import org.apache.carbondata.core.statusmanager.{LoadMetadataDetails, SegmentStatus, SegmentStatusManager}
import org.apache.carbondata.core.util.{CarbonUtil, DataTypeUtil}
import org.apache.carbondata.core.util.path.CarbonTablePath
import org.apache.carbondata.events.{OperationContext, OperationListenerBus}
import org.apache.carbondata.hadoop.{CarbonInputSplit, CarbonProjection}
import org.apache.carbondata.hadoop.api.{CarbonInputFormat, CarbonTableInputFormat}
import org.apache.carbondata.processing.loading.events.LoadEvents.{LoadTablePostStatusUpdateEvent, LoadTablePreStatusUpdateEvent}
import org.apache.carbondata.processing.loading.model.CarbonLoadModel
import org.apache.carbondata.processing.merger.{CompactionResultSortProcessor, CompactionType}
import org.apache.carbondata.processing.util.CarbonLoaderUtil
import org.apache.carbondata.spark.{HandoffResult, HandoffResultImpl}
import org.apache.carbondata.spark.util.CommonUtil
import org.apache.carbondata.streaming.{CarbonStreamInputFormat, CarbonStreamRecordReader}


/**
 * partition of the handoff segment
 */
class HandoffPartition(
    val rddId: Int,
    val idx: Int,
    @transient val inputSplit: CarbonInputSplit
) extends Partition {

  val split = new SerializableWritable[CarbonInputSplit](inputSplit)

  override val index: Int = idx

  override def hashCode(): Int = 41 * (41 + rddId) + idx
}

/**
 * package the record reader of the handoff segment to RawResultIterator
 * todo: actually we should not extends rawResultIterator if we don't use any method or variable
 * from it. We only use it to reduce duplicate code for compaction and handoff
 * and we can extract it later
 */
class StreamingRawResultIterator(
    recordReader: CarbonStreamRecordReader
) extends RawResultIterator(null, null, null) {

  override def hasNext: Boolean = {
    recordReader.nextKeyValue()
  }

  override def next(): Array[Object] = {
    val rowTmp = recordReader
      .getCurrentValue
      .asInstanceOf[GenericInternalRow]
      .values
      .asInstanceOf[Array[Object]]
    val row = new Array[Object](rowTmp.length)
    System.arraycopy(rowTmp, 0, row, 0, rowTmp.length)
    row
  }
}

/**
 * execute streaming segment handoff
 */
class StreamHandoffRDD[K, V](
    sc: SparkContext,
    result: HandoffResult[K, V],
    carbonLoadModel: CarbonLoadModel,
    handOffSegmentId: String
) extends CarbonRDD[(K, V)](sc, Nil, sc.hadoopConfiguration) {

  private val jobTrackerId: String = {
    val formatter = new SimpleDateFormat("yyyyMMddHHmm")
    formatter.format(new Date())
  }

  override def internalCompute(
      split: Partition,
      context: TaskContext
  ): Iterator[(K, V)] = {
    carbonLoadModel.setTaskNo("" + split.index)
    val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
    DataTypeUtil.setDataTypeConverter(new SparkDataTypeConverterImpl)
    CarbonMetadata.getInstance().addCarbonTable(carbonTable)
    // the input iterator is using raw row
    val iteratorList = prepareInputIterator(split, carbonTable)

    CommonUtil.setTempStoreLocation(split.index, carbonLoadModel, true, false)
    // use CompactionResultSortProcessor to sort data dan write to columnar files
    val processor = prepareHandoffProcessor(carbonTable)
    val status = processor.execute(iteratorList)

    new Iterator[(K, V)] {
      private var finished = false

      override def hasNext: Boolean = {
        !finished
      }

      override def next(): (K, V) = {
        finished = true
        result.getKey("" + split.index, status)
      }
    }
  }

  /**
   * prepare input iterator by basing CarbonStreamRecordReader
   */
  private def prepareInputIterator(
      split: Partition,
      carbonTable: CarbonTable
  ): util.ArrayList[RawResultIterator] = {
    val inputSplit = split.asInstanceOf[HandoffPartition].split.value
    val attemptId = new TaskAttemptID(jobTrackerId, id, TaskType.MAP, split.index, 0)
    val hadoopConf = new Configuration()
    CarbonInputFormat.setDatabaseName(hadoopConf, carbonTable.getDatabaseName)
    CarbonInputFormat.setTableName(hadoopConf, carbonTable.getTableName)
    CarbonInputFormat.setTablePath(hadoopConf, carbonTable.getTablePath)
    val projection = new CarbonProjection
    val dataFields = carbonTable.getStreamStorageOrderColumn(carbonTable.getTableName)
    (0 until dataFields.size()).foreach { index =>
      projection.addColumn(dataFields.get(index).getColName)
    }
    CarbonInputFormat.setColumnProjection(hadoopConf, projection)
    CarbonInputFormat.setTableInfo(hadoopConf, carbonTable.getTableInfo)
    val attemptContext = new TaskAttemptContextImpl(hadoopConf, attemptId)
    val format = new CarbonTableInputFormat[Array[Object]]()
    val model = format.createQueryModel(inputSplit, attemptContext)
    val inputFormat = new CarbonStreamInputFormat
    val streamReader = inputFormat.createRecordReader(inputSplit, attemptContext)
      .asInstanceOf[CarbonStreamRecordReader]
    streamReader.setVectorReader(false)
    streamReader.setQueryModel(model)
    streamReader.setUseRawRow(true)
    streamReader.initialize(inputSplit, attemptContext)
    val iteratorList = new util.ArrayList[RawResultIterator](1)
    iteratorList.add(new StreamingRawResultIterator(streamReader))
    iteratorList
  }

  private def prepareHandoffProcessor(
      carbonTable: CarbonTable
  ): CompactionResultSortProcessor = {
    val wrapperColumnSchemaList = CarbonUtil.getColumnSchemaList(
      carbonTable.getDimensionByTableName(carbonTable.getTableName),
      carbonTable.getMeasureByTableName(carbonTable.getTableName))
    val dimLensWithComplex =
      (0 until wrapperColumnSchemaList.size()).map(_ => Integer.MAX_VALUE).toArray
    val dictionaryColumnCardinality =
      CarbonUtil.getFormattedCardinality(dimLensWithComplex, wrapperColumnSchemaList)
    val segmentProperties =
      new SegmentProperties(wrapperColumnSchemaList, dictionaryColumnCardinality)

    new CompactionResultSortProcessor(
      carbonLoadModel,
      carbonTable,
      segmentProperties,
      CompactionType.STREAMING,
      carbonTable.getTableName,
      null
    )
  }

  /**
   * get the partitions of the handoff segment
   */
  override protected def getPartitions: Array[Partition] = {
    val job = Job.getInstance(FileFactory.getConfiguration)
    val inputFormat = new CarbonTableInputFormat[Array[Object]]()
    val segmentList = new util.ArrayList[Segment](1)
    segmentList.add(Segment.toSegment(handOffSegmentId, null))
    val splits = inputFormat.getSplitsOfStreaming(
      job,
      carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable.getAbsoluteTableIdentifier,
      segmentList
    )

    (0 until splits.size()).map { index =>
      new HandoffPartition(id, index, splits.get(index).asInstanceOf[CarbonInputSplit])
    }.toArray[Partition]
  }
}

object StreamHandoffRDD {

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

  def iterateStreamingHandoff(
      carbonLoadModel: CarbonLoadModel,
      operationContext: OperationContext,
      sparkSession: SparkSession): Unit = {
    val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
    val identifier = carbonTable.getAbsoluteTableIdentifier
    var continueHandoff = false
    // require handoff lock on table
    val lock = CarbonLockFactory.getCarbonLockObj(identifier, LockUsage.HANDOFF_LOCK)
    try {
      if (lock.lockWithRetries()) {
        LOGGER.info("Acquired the handoff lock for table" +
                    s" ${ carbonTable.getDatabaseName }.${ carbonTable.getTableName }")
        // handoff streaming segment one by one
        do {
          val segmentStatusManager = new SegmentStatusManager(identifier)
          var loadMetadataDetails: Array[LoadMetadataDetails] = null
          // lock table to read table status file
          val statusLock = segmentStatusManager.getTableStatusLock
          try {
            if (statusLock.lockWithRetries()) {
              loadMetadataDetails = SegmentStatusManager.readLoadMetadata(
                CarbonTablePath.getMetadataPath(identifier.getTablePath))
            }
          } finally {
            if (null != statusLock) {
              statusLock.unlock()
            }
          }
          if (null != loadMetadataDetails) {
            val streamSegments =
              loadMetadataDetails.filter(_.getSegmentStatus == SegmentStatus.STREAMING_FINISH)

            continueHandoff = streamSegments.length > 0
            if (continueHandoff) {
              // handoff a streaming segment
              val loadMetadataDetail = streamSegments(0)
              carbonLoadModel.setSegmentId(loadMetadataDetail.getLoadName)
              executeStreamingHandoff(
                carbonLoadModel,
                sparkSession,
                operationContext,
                loadMetadataDetail.getLoadName
              )
            }
          } else {
            continueHandoff = false
          }
        } while (continueHandoff)
      }
    } finally {
      if (null != lock) {
        lock.unlock()
      }
    }
  }

  /**
   * start new thread to execute stream segment handoff
   */
  def startStreamingHandoffThread(
      carbonLoadModel: CarbonLoadModel,
      operationContext: OperationContext,
      sparkSession: SparkSession,
      isDDL: Boolean
  ): Unit = {
    if (isDDL) {
      iterateStreamingHandoff(carbonLoadModel, operationContext, sparkSession)
    } else {
      // start a new thread to execute streaming segment handoff
      val handoffThread = new Thread() {
        override def run(): Unit = {
          iterateStreamingHandoff(carbonLoadModel, operationContext, sparkSession)
        }
      }
      handoffThread.start()
    }
  }

  /**
   * invoke StreamHandoffRDD to handoff a streaming segment to a columnar segment
   */
  def executeStreamingHandoff(
      carbonLoadModel: CarbonLoadModel,
      sparkSession: SparkSession,
      operationContext: OperationContext,
      handoffSegmenId: String): Unit = {
    var loadStatus = SegmentStatus.SUCCESS
    var errorMessage: String = "Handoff failure"
    try {
      // generate new columnar segment
      val newMetaEntry = new LoadMetadataDetails
      carbonLoadModel.setFactTimeStamp(System.currentTimeMillis())
      CarbonLoaderUtil.populateNewLoadMetaEntry(
        newMetaEntry,
        SegmentStatus.INSERT_IN_PROGRESS,
        carbonLoadModel.getFactTimeStamp,
        false)
      CarbonLoaderUtil.recordNewLoadMetadata(newMetaEntry, carbonLoadModel, true, false)
      // convert a streaming segment to columnar segment

      val status = new StreamHandoffRDD(
        sparkSession.sparkContext,
        new HandoffResultImpl(),
        carbonLoadModel,
        handoffSegmenId).collect()

      status.foreach { x =>
        if (!x._2) {
          loadStatus = SegmentStatus.LOAD_FAILURE
        }
      }
    } catch {
      case ex: Exception =>
        loadStatus = SegmentStatus.LOAD_FAILURE
        errorMessage = errorMessage + ": " + ex.getCause.getMessage
        LOGGER.error(errorMessage)
        LOGGER.error(ex, s"Handoff failed on streaming segment $handoffSegmenId")
    }

    if (loadStatus == SegmentStatus.LOAD_FAILURE) {
      CarbonLoaderUtil.updateTableStatusForFailure(carbonLoadModel)
      LOGGER.info("********starting clean up**********")
      CarbonLoaderUtil.deleteSegment(carbonLoadModel, carbonLoadModel.getSegmentId.toInt)
      LOGGER.info("********clean up done**********")
      LOGGER.audit(s"Handoff is failed for " +
                   s"${ carbonLoadModel.getDatabaseName }.${ carbonLoadModel.getTableName }")
      LOGGER.error("Cannot write load metadata file as handoff failed")
      throw new Exception(errorMessage)
    }

    if (loadStatus == SegmentStatus.SUCCESS) {
      operationContext.setProperty("uuid", UUID.randomUUID().toString)
      val loadTablePreStatusUpdateEvent: LoadTablePreStatusUpdateEvent =
        new LoadTablePreStatusUpdateEvent(
          carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable.getCarbonTableIdentifier,
          carbonLoadModel)
      OperationListenerBus.getInstance().fireEvent(loadTablePreStatusUpdateEvent, operationContext)

      val done = updateLoadMetadata(handoffSegmenId, carbonLoadModel)

      val loadTablePostStatusUpdateEvent: LoadTablePostStatusUpdateEvent =
        new LoadTablePostStatusUpdateEvent(carbonLoadModel)
      OperationListenerBus.getInstance()
        .fireEvent(loadTablePostStatusUpdateEvent, operationContext)
      if (!done) {
        val errorMessage = "Handoff failed due to failure in table status updation."
        LOGGER.audit("Handoff is failed for " +
                     s"${ carbonLoadModel.getDatabaseName }.${ carbonLoadModel.getTableName }")
        LOGGER.error("Handoff failed due to failure in table status updation.")
        throw new Exception(errorMessage)
      }
      done
    }

  }

  /**
   * update streaming segment and new columnar segment
   */
  private def updateLoadMetadata(
      handoffSegmentId: String,
      loadModel: CarbonLoadModel
  ): Boolean = {
    var status = false
    val metaDataFilepath = loadModel.getCarbonDataLoadSchema.getCarbonTable.getMetadataPath
    val identifier = loadModel.getCarbonDataLoadSchema.getCarbonTable.getAbsoluteTableIdentifier
    val metadataPath = CarbonTablePath.getMetadataPath(identifier.getTablePath)
    val fileType = FileFactory.getFileType(metadataPath)
    if (!FileFactory.isFileExist(metadataPath, fileType)) {
      FileFactory.mkdirs(metadataPath, fileType)
    }
    val tableStatusPath = CarbonTablePath.getTableStatusFilePath(identifier.getTablePath)
    val segmentStatusManager = new SegmentStatusManager(identifier)
    val carbonLock = segmentStatusManager.getTableStatusLock
    try {
      if (carbonLock.lockWithRetries()) {
        LOGGER.info(
          "Acquired lock for table" + loadModel.getDatabaseName() + "." + loadModel.getTableName()
          + " for table status updation")
        val listOfLoadFolderDetailsArray =
          SegmentStatusManager.readLoadMetadata(metaDataFilepath)

        // update new columnar segment to success status
        val newSegment =
          listOfLoadFolderDetailsArray.find(_.getLoadName.equals(loadModel.getSegmentId))
        if (newSegment.isEmpty) {
          throw new Exception("Failed to update table status for new segment")
        } else {
          newSegment.get.setSegmentStatus(SegmentStatus.SUCCESS)
          newSegment.get.setLoadEndTime(System.currentTimeMillis())
        }

        // update streaming segment to compacted status
        val streamSegment =
          listOfLoadFolderDetailsArray.find(_.getLoadName.equals(handoffSegmentId))
        if (streamSegment.isEmpty) {
          throw new Exception("Failed to update table status for streaming segment")
        } else {
          streamSegment.get.setSegmentStatus(SegmentStatus.COMPACTED)
          streamSegment.get.setMergedLoadName(loadModel.getSegmentId)
        }

        // refresh table status file
        SegmentStatusManager.writeLoadDetailsIntoFile(tableStatusPath, listOfLoadFolderDetailsArray)
        status = true
      } else {
        LOGGER.error("Not able to acquire the lock for Table status updation for table " + loadModel
          .getDatabaseName() + "." + loadModel.getTableName())
      }
    } finally {
      if (carbonLock.unlock()) {
        LOGGER.info("Table unlocked successfully after table status updation" +
                    loadModel.getDatabaseName() + "." + loadModel.getTableName())
      } else {
        LOGGER.error("Unable to unlock Table lock for table" + loadModel.getDatabaseName() +
                     "." + loadModel.getTableName() + " during table status updation")
      }
    }
    status
  }
}

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.spark.rdd
19
20	import java.text.SimpleDateFormat
21	import java.util
22	import java.util.{Date, UUID}
23
24	import org.apache.hadoop.conf.Configuration
25	import org.apache.hadoop.mapreduce.{Job, TaskAttemptID, TaskType}
26	import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl
27	import org.apache.spark.{Partition, SerializableWritable, SparkContext, TaskContext}
28	import org.apache.spark.sql.SparkSession
29	import org.apache.spark.sql.catalyst.expressions.GenericInternalRow
30
31	import org.apache.carbondata.common.logging.LogServiceFactory
32	import org.apache.carbondata.converter.SparkDataTypeConverterImpl
33	import org.apache.carbondata.core.datamap.Segment
34	import org.apache.carbondata.core.datastore.block.SegmentProperties
35	import org.apache.carbondata.core.datastore.impl.FileFactory
36	import org.apache.carbondata.core.locks.{CarbonLockFactory, LockUsage}
37	import org.apache.carbondata.core.metadata.CarbonMetadata
38	import org.apache.carbondata.core.metadata.schema.table.CarbonTable
39	import org.apache.carbondata.core.scan.result.iterator.RawResultIterator
40	import org.apache.carbondata.core.statusmanager.{LoadMetadataDetails, SegmentStatus, SegmentStatusManager}
41	import org.apache.carbondata.core.util.{CarbonUtil, DataTypeUtil}
42	import org.apache.carbondata.core.util.path.CarbonTablePath
43	import org.apache.carbondata.events.{OperationContext, OperationListenerBus}
44	import org.apache.carbondata.hadoop.{CarbonInputSplit, CarbonProjection}
45	import org.apache.carbondata.hadoop.api.{CarbonInputFormat, CarbonTableInputFormat}
46	import org.apache.carbondata.processing.loading.events.LoadEvents.{LoadTablePostStatusUpdateEvent, LoadTablePreStatusUpdateEvent}
47	import org.apache.carbondata.processing.loading.model.CarbonLoadModel
48	import org.apache.carbondata.processing.merger.{CompactionResultSortProcessor, CompactionType}
49	import org.apache.carbondata.processing.util.CarbonLoaderUtil
50	import org.apache.carbondata.spark.{HandoffResult, HandoffResultImpl}
51	import org.apache.carbondata.spark.util.CommonUtil
52	import org.apache.carbondata.streaming.{CarbonStreamInputFormat, CarbonStreamRecordReader}
53
54
55	/**
56	* partition of the handoff segment
57	*/
58	class HandoffPartition(	×
59	val rddId: Int,	1✔
60	val idx: Int,	1✔
61	@transient val inputSplit: CarbonInputSplit	1✔
62	) extends Partition {	1✔
63
64	val split = new SerializableWritable[CarbonInputSplit](inputSplit)	1✔
65
66	override val index: Int = idx	1✔
67
68	override def hashCode(): Int = 41 * (41 + rddId) + idx	×
69	}
70
71	/**
72	* package the record reader of the handoff segment to RawResultIterator
73	* todo: actually we should not extends rawResultIterator if we don't use any method or variable
74	* from it. We only use it to reduce duplicate code for compaction and handoff
75	* and we can extract it later
76	*/
77	class StreamingRawResultIterator(
78	recordReader: CarbonStreamRecordReader	1✔
79	) extends RawResultIterator(null, null, null) {	1✔
80
81	override def hasNext: Boolean = {
82	recordReader.nextKeyValue()	1✔
83	}
84
85	override def next(): Array[Object] = {
86	val rowTmp = recordReader	1✔
87	.getCurrentValue	1✔
88	.asInstanceOf[GenericInternalRow]
89	.values	1✔
90	.asInstanceOf[Array[Object]]
91	val row = new Array[Object](rowTmp.length)	1✔
92	System.arraycopy(rowTmp, 0, row, 0, rowTmp.length)	1✔
93	row	1✔
94	}
95	}
96
97	/**
98	* execute streaming segment handoff
99	*/
100	class StreamHandoffRDD[K, V](
101	sc: SparkContext,
102	result: HandoffResult[K, V],	1✔
103	carbonLoadModel: CarbonLoadModel,	1✔
104	handOffSegmentId: String	1✔
105	) extends CarbonRDD[(K, V)](sc, Nil, sc.hadoopConfiguration) {	1✔
106
107	private val jobTrackerId: String = {	1✔
108	val formatter = new SimpleDateFormat("yyyyMMddHHmm")	1✔
109	formatter.format(new Date())	1✔
110	}
111
112	override def internalCompute(
113	split: Partition,
114	context: TaskContext
115	): Iterator[(K, V)] = {
116	carbonLoadModel.setTaskNo("" + split.index)	1✔
117	val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable	1✔
118	DataTypeUtil.setDataTypeConverter(new SparkDataTypeConverterImpl)	1✔
119	CarbonMetadata.getInstance().addCarbonTable(carbonTable)	1✔
120	// the input iterator is using raw row
121	val iteratorList = prepareInputIterator(split, carbonTable)	1✔
122
123	CommonUtil.setTempStoreLocation(split.index, carbonLoadModel, true, false)	1✔
124	// use CompactionResultSortProcessor to sort data dan write to columnar files
125	val processor = prepareHandoffProcessor(carbonTable)	1✔
126	val status = processor.execute(iteratorList)	1✔
127
128	new Iterator[(K, V)] {	1✔
129	private var finished = false	1✔
130
131	override def hasNext: Boolean = {
132	!finished	1✔
133	}
134
135	override def next(): (K, V) = {
136	finished = true	1✔
137	result.getKey("" + split.index, status)	1✔
138	}
139	}
140	}
141
142	/**
143	* prepare input iterator by basing CarbonStreamRecordReader
144	*/
145	private def prepareInputIterator(
146	split: Partition,
147	carbonTable: CarbonTable
148	): util.ArrayList[RawResultIterator] = {
149	val inputSplit = split.asInstanceOf[HandoffPartition].split.value	1✔
150	val attemptId = new TaskAttemptID(jobTrackerId, id, TaskType.MAP, split.index, 0)	1✔
151	val hadoopConf = new Configuration()	1✔
152	CarbonInputFormat.setDatabaseName(hadoopConf, carbonTable.getDatabaseName)	1✔
153	CarbonInputFormat.setTableName(hadoopConf, carbonTable.getTableName)	1✔
154	CarbonInputFormat.setTablePath(hadoopConf, carbonTable.getTablePath)	1✔
155	val projection = new CarbonProjection	1✔
156	val dataFields = carbonTable.getStreamStorageOrderColumn(carbonTable.getTableName)	1✔
157	(0 until dataFields.size()).foreach { index =>	1✔
158	projection.addColumn(dataFields.get(index).getColName)	1✔
159	}
160	CarbonInputFormat.setColumnProjection(hadoopConf, projection)	1✔
161	CarbonInputFormat.setTableInfo(hadoopConf, carbonTable.getTableInfo)	1✔
162	val attemptContext = new TaskAttemptContextImpl(hadoopConf, attemptId)	1✔
163	val format = new CarbonTableInputFormat[Array[Object]]()	1✔
164	val model = format.createQueryModel(inputSplit, attemptContext)	1✔
165	val inputFormat = new CarbonStreamInputFormat	1✔
166	val streamReader = inputFormat.createRecordReader(inputSplit, attemptContext)	1✔
167	.asInstanceOf[CarbonStreamRecordReader]
168	streamReader.setVectorReader(false)	1✔
169	streamReader.setQueryModel(model)	1✔
170	streamReader.setUseRawRow(true)	1✔
171	streamReader.initialize(inputSplit, attemptContext)	1✔
172	val iteratorList = new util.ArrayList[RawResultIterator](1)	1✔
173	iteratorList.add(new StreamingRawResultIterator(streamReader))	1✔
174	iteratorList	1✔
175	}
176
177	private def prepareHandoffProcessor(
178	carbonTable: CarbonTable
179	): CompactionResultSortProcessor = {
180	val wrapperColumnSchemaList = CarbonUtil.getColumnSchemaList(	1✔
181	carbonTable.getDimensionByTableName(carbonTable.getTableName),	1✔
182	carbonTable.getMeasureByTableName(carbonTable.getTableName))	1✔
183	val dimLensWithComplex =	1✔
184	(0 until wrapperColumnSchemaList.size()).map(_ => Integer.MAX_VALUE).toArray	1✔
185	val dictionaryColumnCardinality =	1✔
186	CarbonUtil.getFormattedCardinality(dimLensWithComplex, wrapperColumnSchemaList)	1✔
187	val segmentProperties =	1✔
188	new SegmentProperties(wrapperColumnSchemaList, dictionaryColumnCardinality)	1✔
189
190	new CompactionResultSortProcessor(	1✔
191	carbonLoadModel,	1✔
192	carbonTable,	1✔
193	segmentProperties,	1✔
194	CompactionType.STREAMING,	1✔
195	carbonTable.getTableName,	1✔
196	null	1✔
197	)
198	}
199
200	/**
201	* get the partitions of the handoff segment
202	*/
203	override protected def getPartitions: Array[Partition] = {
204	val job = Job.getInstance(FileFactory.getConfiguration)	1✔
205	val inputFormat = new CarbonTableInputFormat[Array[Object]]()	1✔
206	val segmentList = new util.ArrayList[Segment](1)	1✔
207	segmentList.add(Segment.toSegment(handOffSegmentId, null))	1✔
208	val splits = inputFormat.getSplitsOfStreaming(	1✔
209	job,	1✔
210	carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable.getAbsoluteTableIdentifier,	1✔
211	segmentList	1✔
212	)
213
214	(0 until splits.size()).map { index =>	1✔
215	new HandoffPartition(id, index, splits.get(index).asInstanceOf[CarbonInputSplit])	1✔
216	}.toArray[Partition]	1✔
217	}
218	}
219
220	object StreamHandoffRDD {	×
221
222	private val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)	1✔
223
224	def iterateStreamingHandoff(
225	carbonLoadModel: CarbonLoadModel,
226	operationContext: OperationContext,
227	sparkSession: SparkSession): Unit = {	1✔
228	val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable	1✔
229	val identifier = carbonTable.getAbsoluteTableIdentifier	1✔
230	var continueHandoff = false	1✔
231	// require handoff lock on table
232	val lock = CarbonLockFactory.getCarbonLockObj(identifier, LockUsage.HANDOFF_LOCK)	1✔
233	try {
234	if (lock.lockWithRetries()) {	1✔
235	LOGGER.info("Acquired the handoff lock for table" +	1✔
236	s" ${ carbonTable.getDatabaseName }.${ carbonTable.getTableName }")	1✔
237	// handoff streaming segment one by one
238	do {	1✔
239	val segmentStatusManager = new SegmentStatusManager(identifier)
240	var loadMetadataDetails: Array[LoadMetadataDetails] = null	1✔
241	// lock table to read table status file
242	val statusLock = segmentStatusManager.getTableStatusLock	1✔
243	try {
244	if (statusLock.lockWithRetries()) {	1✔
245	loadMetadataDetails = SegmentStatusManager.readLoadMetadata(	1✔
246	CarbonTablePath.getMetadataPath(identifier.getTablePath))	1✔
247	}
248	} finally {
249	if (null != statusLock) {	1✔
250	statusLock.unlock()	1✔
251	}
252	}
253	if (null != loadMetadataDetails) {	1✔
254	val streamSegments =	1✔
255	loadMetadataDetails.filter(_.getSegmentStatus == SegmentStatus.STREAMING_FINISH)	1✔
256
257	continueHandoff = streamSegments.length > 0	1✔
258	if (continueHandoff) {	1✔
259	// handoff a streaming segment
260	val loadMetadataDetail = streamSegments(0)	1✔
261	carbonLoadModel.setSegmentId(loadMetadataDetail.getLoadName)	1✔
262	executeStreamingHandoff(	1✔
263	carbonLoadModel,	1✔
264	sparkSession,	1✔
265	operationContext,	1✔
266	loadMetadataDetail.getLoadName	1✔
267	)
268	}
269	} else {
270	continueHandoff = false	×
271	}
272	} while (continueHandoff)	1✔
273	}
274	} finally {
275	if (null != lock) {	1✔
276	lock.unlock()	1✔
277	}
278	}
279	}
280
281	/**
282	* start new thread to execute stream segment handoff
283	*/
284	def startStreamingHandoffThread(
285	carbonLoadModel: CarbonLoadModel,
286	operationContext: OperationContext,
287	sparkSession: SparkSession,
288	isDDL: Boolean
289	): Unit = {
290	if (isDDL) {	1✔
291	iterateStreamingHandoff(carbonLoadModel, operationContext, sparkSession)	1✔
292	} else {
293	// start a new thread to execute streaming segment handoff
294	val handoffThread = new Thread() {	1✔
295	override def run(): Unit = {
296	iterateStreamingHandoff(carbonLoadModel, operationContext, sparkSession)	1✔
297	}
298	}
299	handoffThread.start()	1✔
300	}
301	}
302
303	/**
304	* invoke StreamHandoffRDD to handoff a streaming segment to a columnar segment
305	*/
306	def executeStreamingHandoff(
307	carbonLoadModel: CarbonLoadModel,
308	sparkSession: SparkSession,
309	operationContext: OperationContext,
310	handoffSegmenId: String): Unit = {	1✔
311	var loadStatus = SegmentStatus.SUCCESS	1✔
312	var errorMessage: String = "Handoff failure"	1✔
313	try {	×
314	// generate new columnar segment
315	val newMetaEntry = new LoadMetadataDetails	1✔
316	carbonLoadModel.setFactTimeStamp(System.currentTimeMillis())	1✔
317	CarbonLoaderUtil.populateNewLoadMetaEntry(	1✔
318	newMetaEntry,	1✔
319	SegmentStatus.INSERT_IN_PROGRESS,	1✔
320	carbonLoadModel.getFactTimeStamp,	1✔
321	false)	1✔
322	CarbonLoaderUtil.recordNewLoadMetadata(newMetaEntry, carbonLoadModel, true, false)	1✔
323	// convert a streaming segment to columnar segment
324
325	val status = new StreamHandoffRDD(	1✔
326	sparkSession.sparkContext,	1✔
327	new HandoffResultImpl(),	1✔
328	carbonLoadModel,	1✔
329	handoffSegmenId).collect()	1✔
330
331	status.foreach { x =>	1✔
332	if (!x._2) {	1✔
333	loadStatus = SegmentStatus.LOAD_FAILURE	×
334	}
335	}
336	} catch {
337	case ex: Exception =>
338	loadStatus = SegmentStatus.LOAD_FAILURE	×
339	errorMessage = errorMessage + ": " + ex.getCause.getMessage	×
340	LOGGER.error(errorMessage)	×
341	LOGGER.error(ex, s"Handoff failed on streaming segment $handoffSegmenId")	×
342	}
343
344	if (loadStatus == SegmentStatus.LOAD_FAILURE) {	1✔
345	CarbonLoaderUtil.updateTableStatusForFailure(carbonLoadModel)	×
346	LOGGER.info("******starting clean up********")	×
347	CarbonLoaderUtil.deleteSegment(carbonLoadModel, carbonLoadModel.getSegmentId.toInt)	×
348	LOGGER.info("******clean up done********")	×
349	LOGGER.audit(s"Handoff is failed for " +	×
350	s"${ carbonLoadModel.getDatabaseName }.${ carbonLoadModel.getTableName }")	×
351	LOGGER.error("Cannot write load metadata file as handoff failed")	×
352	throw new Exception(errorMessage)	×
353	}
354
355	if (loadStatus == SegmentStatus.SUCCESS) {	1✔
356	operationContext.setProperty("uuid", UUID.randomUUID().toString)	1✔
357	val loadTablePreStatusUpdateEvent: LoadTablePreStatusUpdateEvent =	1✔
358	new LoadTablePreStatusUpdateEvent(	1✔
359	carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable.getCarbonTableIdentifier,	1✔
360	carbonLoadModel)	1✔
361	OperationListenerBus.getInstance().fireEvent(loadTablePreStatusUpdateEvent, operationContext)	1✔
362
363	val done = updateLoadMetadata(handoffSegmenId, carbonLoadModel)	1✔
364
365	val loadTablePostStatusUpdateEvent: LoadTablePostStatusUpdateEvent =	1✔
366	new LoadTablePostStatusUpdateEvent(carbonLoadModel)	1✔
367	OperationListenerBus.getInstance()	1✔
368	.fireEvent(loadTablePostStatusUpdateEvent, operationContext)	1✔
369	if (!done) {	1✔
370	val errorMessage = "Handoff failed due to failure in table status updation."	×
371	LOGGER.audit("Handoff is failed for " +	×
372	s"${ carbonLoadModel.getDatabaseName }.${ carbonLoadModel.getTableName }")	×
373	LOGGER.error("Handoff failed due to failure in table status updation.")	×
374	throw new Exception(errorMessage)	×
375	}
376	done	1✔
377	}
378
379	}
380
381	/**
382	* update streaming segment and new columnar segment
383	*/
384	private def updateLoadMetadata(
385	handoffSegmentId: String,
386	loadModel: CarbonLoadModel
387	): Boolean = {
388	var status = false	1✔
389	val metaDataFilepath = loadModel.getCarbonDataLoadSchema.getCarbonTable.getMetadataPath	1✔
390	val identifier = loadModel.getCarbonDataLoadSchema.getCarbonTable.getAbsoluteTableIdentifier	1✔
391	val metadataPath = CarbonTablePath.getMetadataPath(identifier.getTablePath)	1✔
392	val fileType = FileFactory.getFileType(metadataPath)	1✔
393	if (!FileFactory.isFileExist(metadataPath, fileType)) {	1✔
394	FileFactory.mkdirs(metadataPath, fileType)	×
395	}
396	val tableStatusPath = CarbonTablePath.getTableStatusFilePath(identifier.getTablePath)	1✔
397	val segmentStatusManager = new SegmentStatusManager(identifier)	1✔
398	val carbonLock = segmentStatusManager.getTableStatusLock	1✔
399	try {
400	if (carbonLock.lockWithRetries()) {	1✔
401	LOGGER.info(	1✔
402	"Acquired lock for table" + loadModel.getDatabaseName() + "." + loadModel.getTableName()	1✔
403	+ " for table status updation")	1✔
404	val listOfLoadFolderDetailsArray =	1✔
405	SegmentStatusManager.readLoadMetadata(metaDataFilepath)	1✔
406
407	// update new columnar segment to success status
408	val newSegment =	1✔
409	listOfLoadFolderDetailsArray.find(_.getLoadName.equals(loadModel.getSegmentId))	1✔
410	if (newSegment.isEmpty) {	1✔
411	throw new Exception("Failed to update table status for new segment")	×
412	} else {
413	newSegment.get.setSegmentStatus(SegmentStatus.SUCCESS)	1✔
414	newSegment.get.setLoadEndTime(System.currentTimeMillis())	1✔
415	}
416
417	// update streaming segment to compacted status
418	val streamSegment =	1✔
419	listOfLoadFolderDetailsArray.find(_.getLoadName.equals(handoffSegmentId))	1✔
420	if (streamSegment.isEmpty) {	1✔
421	throw new Exception("Failed to update table status for streaming segment")	×
422	} else {
423	streamSegment.get.setSegmentStatus(SegmentStatus.COMPACTED)	1✔
424	streamSegment.get.setMergedLoadName(loadModel.getSegmentId)	1✔
425	}
426
427	// refresh table status file
428	SegmentStatusManager.writeLoadDetailsIntoFile(tableStatusPath, listOfLoadFolderDetailsArray)	1✔
429	status = true	1✔
430	} else {
431	LOGGER.error("Not able to acquire the lock for Table status updation for table " + loadModel	×
432	.getDatabaseName() + "." + loadModel.getTableName())	×
433	}
434	} finally {
435	if (carbonLock.unlock()) {	1✔
436	LOGGER.info("Table unlocked successfully after table status updation" +	1✔
437	loadModel.getDatabaseName() + "." + loadModel.getTableName())	1✔
438	} else {
439	LOGGER.error("Unable to unlock Table lock for table" + loadModel.getDatabaseName() +	×
440	"." + loadModel.getTableName() + " during table status updation")	×
441	}
442	}
443	status	1✔
444	}
445	}	1✔

apache / carbondata / 18

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