8535

Build # 8535

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/eb3fb99f4a5a77d417915b39fd73ffd04a530a0a">eb3fb99f4<a href="https://github.com/apache/carbondata/commit/eb3fb99f4a5a77d417915b39fd73ffd04a530a0a">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/98172df2e2cd9c2e002f7f492b36b22dce0aea19">98172df2e</a>">98172df2e</a><a href="https://github.com/apache/carbondata/commit/eb3fb99f4a5a77d417915b39fd73ffd04a530a0a"> into 2fb7dc9a7">2fb7dc9a7</a>

Run Details

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

50710 of 72944 relevant lines covered (69.52%)

0.96 hits per line

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

89.2

/integration/spark2/src/main/scala/org/apache/carbondata/datamap/IndexDataMapRebuildRDD.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.datamap

import java.io.{File, IOException}
import java.text.SimpleDateFormat
import java.util

import scala.collection.JavaConverters._
import scala.collection.mutable
import scala.collection.mutable.ListBuffer

import org.apache.commons.lang3.ArrayUtils
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.mapred.JobConf
import org.apache.hadoop.mapreduce.{Job, TaskAttemptID, TaskType}
import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl
import org.apache.spark.{CarbonInputMetrics, Partition, TaskContext}
import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.Decimal

import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.converter.SparkDataTypeConverterImpl
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datamap.{DataMapStoreManager, Segment}
import org.apache.carbondata.core.datamap.dev.DataMapBuilder
import org.apache.carbondata.core.datastore.block.SegmentProperties
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.indexstore.blockletindex.BlockletDataMapFactory
import org.apache.carbondata.core.indexstore.SegmentPropertiesFetcher
import org.apache.carbondata.core.keygenerator.KeyGenerator
import org.apache.carbondata.core.keygenerator.mdkey.MultiDimKeyVarLengthGenerator
import org.apache.carbondata.core.metadata.datatype.{DataType, DataTypes}
import org.apache.carbondata.core.metadata.encoder.Encoding
import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema, TableInfo}
import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn
import org.apache.carbondata.core.scan.wrappers.ByteArrayWrapper
import org.apache.carbondata.core.statusmanager.SegmentStatusManager
import org.apache.carbondata.core.util.{CarbonUtil, TaskMetricsMap}
import org.apache.carbondata.core.util.path.CarbonTablePath
import org.apache.carbondata.datamap.bloom.DataConvertUtil
import org.apache.carbondata.events.{BuildDataMapPostExecutionEvent, BuildDataMapPreExecutionEvent, OperationContext, OperationListenerBus}
import org.apache.carbondata.hadoop.{CarbonInputSplit, CarbonMultiBlockSplit, CarbonProjection, CarbonRecordReader}
import org.apache.carbondata.hadoop.api.{CarbonInputFormat, CarbonTableInputFormat}
import org.apache.carbondata.hadoop.readsupport.CarbonReadSupport
import org.apache.carbondata.spark.{RefreshResult, RefreshResultImpl}
import org.apache.carbondata.spark.rdd.{CarbonRDDWithTableInfo, CarbonSparkPartition}


/**
 * Helper object to rebuild the index DataMap
 */
object IndexDataMapRebuildRDD {
  private val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)

  /**
   * Rebuild the datamap for all existing data in the table
   */
  def rebuildDataMap(
      sparkSession: SparkSession,
      carbonTable: CarbonTable,
      schema: DataMapSchema
  ): Unit = {
    val tableIdentifier = carbonTable.getAbsoluteTableIdentifier
    val segmentStatusManager = new SegmentStatusManager(tableIdentifier)
    val validAndInvalidSegments = segmentStatusManager.getValidAndInvalidSegments()
    val validSegments = validAndInvalidSegments.getValidSegments
    val indexedCarbonColumns = carbonTable.getIndexedColumns(schema)
    val operationContext = new OperationContext()
    val buildDataMapPreExecutionEvent = new BuildDataMapPreExecutionEvent(sparkSession,
      tableIdentifier,
      mutable.Seq[String](schema.getDataMapName))
    OperationListenerBus.getInstance().fireEvent(buildDataMapPreExecutionEvent, operationContext)

    val segments2DmStorePath = validSegments.asScala.map { segment =>
      val dataMapStorePath = CarbonTablePath.getDataMapStorePath(carbonTable.getTablePath,
        segment.getSegmentNo, schema.getDataMapName)
      segment -> dataMapStorePath
    }.filter(p => !FileFactory.isFileExist(p._2)).toMap

    segments2DmStorePath.foreach { case (_, dmPath) =>
      if (!FileFactory.mkdirs(dmPath, FileFactory.getFileType(dmPath))) {
        throw new IOException(
          s"Failed to create directory $dmPath for rebuilding datamap ${ schema.getDataMapName }")
      }
    }

    val status = new IndexDataMapRebuildRDD[String, (String, Boolean)](
      sparkSession,
      new RefreshResultImpl(),
      carbonTable.getTableInfo,
      schema.getDataMapName,
      indexedCarbonColumns.asScala.toArray,
      segments2DmStorePath.keySet
    ).collect

    // for failed segments, clean the result
    val failedSegments = status
      .find { case (taskId, (segmentId, rebuildStatus)) =>
        !rebuildStatus
      }
      .map { task =>
        val segmentId = task._2._1
        val dmPath = segments2DmStorePath.filter(p => p._1.getSegmentNo.equals(segmentId)).values
        val cleanResult = dmPath.map(p =>
          FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(p)))
        if (cleanResult.exists(!_)) {
          LOGGER.error(s"Failed to clean up datamap store for segment_$segmentId")
          false
        } else {
          true
        }
      }

    if (failedSegments.nonEmpty) {
      throw new Exception(s"Failed to refresh datamap ${ schema.getDataMapName }")
    }

    val buildDataMapPostExecutionEvent = new BuildDataMapPostExecutionEvent(sparkSession,
      tableIdentifier, validSegments.asScala.map(_.getSegmentNo), true)
    OperationListenerBus.getInstance().fireEvent(buildDataMapPostExecutionEvent, operationContext)
  }
}

class OriginalReadSupport(dataTypes: Array[DataType]) extends CarbonReadSupport[Array[Object]] {
  override def initialize(carbonColumns: Array[CarbonColumn],
      carbonTable: CarbonTable): Unit = {
  }

  override def readRow(data: Array[Object]): Array[Object] = {
    dataTypes.zipWithIndex.foreach { case (dataType, i) =>
      if (dataType == DataTypes.STRING && data(i) != null) {
        data(i) = data(i).toString
      }
    }
    data
  }

  override def close(): Unit = {
  }
}

/**
 * This class will generate row value which is raw bytes for the dimensions.
 */
class RawBytesReadSupport(segmentProperties: SegmentProperties, indexColumns: Array[CarbonColumn])
  extends CarbonReadSupport[Array[Object]] {
  var dimensionKeyGenerator: KeyGenerator = _
  // for the dictionary dimensions
  var indexCol2IdxInDictArray: Map[String, Int] = Map()
  // for the non dictionary dimensions
  var indexCol2IdxInNoDictArray: Map[String, Int] = Map()
  // for the measures
  var indexCol2IdxInMeasureArray: Map[String, Int] = Map()

  /**
   * rebuild process get data from query, if some columns added to table but not in this segment
   * it will be filled with default value and generate new key for dict dimension.
   * Here we use same way as `RowIdRestructureBasedRawResultCollector` to prepare
   * key generator to get surrogate value of dict column result.
   * So we do not need to make a fake mdk to split when adding row to datamap
   */
  def prepareKeyGenForDictIndexColumns(carbonTable: CarbonTable,
                                       dictIndexColumns: ListBuffer[CarbonColumn]): Unit = {

    val columnCardinality = new util.ArrayList[Integer](dictIndexColumns.length)
    val columnPartitioner = new util.ArrayList[Integer](dictIndexColumns.length)

    dictIndexColumns.foreach { col =>
      val dim = carbonTable.getDimensionByName(carbonTable.getTableName, col.getColName)
      val currentBlockDimension = segmentProperties.getDimensionFromCurrentBlock(dim)
      if (null != currentBlockDimension) {
        columnCardinality.add(segmentProperties.getDimColumnsCardinality.apply(
          currentBlockDimension.getKeyOrdinal))
        columnPartitioner.add(segmentProperties.getDimensionPartitions.apply(
          currentBlockDimension.getKeyOrdinal
        ))
      } else {
        columnPartitioner.add(1)
        if (col.hasEncoding(Encoding.DIRECT_DICTIONARY)) {
          columnCardinality.add(Integer.MAX_VALUE)
        } else {
          val defaultValue = col.getDefaultValue
          if (null != col.getDefaultValue) {
            columnCardinality.add(CarbonCommonConstants.DICTIONARY_DEFAULT_CARDINALITY + 1)
          } else {
            columnCardinality.add(CarbonCommonConstants.DICTIONARY_DEFAULT_CARDINALITY)
          }
        }
      }
    }

    if (!columnCardinality.isEmpty) {
      val latestColumnCardinality = ArrayUtils.toPrimitive(columnCardinality.toArray(
        new Array[Integer](columnCardinality.size)))
      val latestColumnPartitioner = ArrayUtils.toPrimitive(columnPartitioner.toArray(
        new Array[Integer](columnPartitioner.size)))
      val dimensionBitLength = CarbonUtil.getDimensionBitLength(
        latestColumnCardinality, latestColumnPartitioner)
      this.dimensionKeyGenerator = new MultiDimKeyVarLengthGenerator(dimensionBitLength)
    }
  }

  override def initialize(carbonColumns: Array[CarbonColumn],
      carbonTable: CarbonTable): Unit = {

    val dictIndexColumns = new ListBuffer[CarbonColumn]()

    // prepare index info to extract data from query result
    indexColumns.foreach { col =>
      if (col.isDimension) {
        val dim = carbonTable.getDimensionByName(carbonTable.getTableName, col.getColName)
        if (!dim.isGlobalDictionaryEncoding && !dim.isDirectDictionaryEncoding) {
          indexCol2IdxInNoDictArray =
            indexCol2IdxInNoDictArray + (col.getColName -> indexCol2IdxInNoDictArray.size)
        } else {
          dictIndexColumns.append(col)
          indexCol2IdxInDictArray =
            indexCol2IdxInDictArray + (col.getColName -> indexCol2IdxInDictArray.size)
        }
      } else {
        indexCol2IdxInMeasureArray =
          indexCol2IdxInMeasureArray + (col.getColName -> indexCol2IdxInMeasureArray.size)
      }
    }

    if (dictIndexColumns.size > 0) {
      prepareKeyGenForDictIndexColumns(carbonTable, dictIndexColumns)
    }
  }

  /**
   * input: all the dimensions are bundled in one ByteArrayWrapper in position 0,
   * then comes the measures one by one; last 3 elements are block/page/row id
   * output: all the dimensions and measures comes one after another
   */
  override def readRow(data: Array[Object]): Array[Object] = {

    var surrogatKeys = new Array[Long](0)
    if(null != dimensionKeyGenerator) {
        surrogatKeys = dimensionKeyGenerator.getKeyArray(
          data(0).asInstanceOf[ByteArrayWrapper].getDictionaryKey)
    }

    // fill return row from data
    val rtn = new Array[Object](indexColumns.length + 3)
    indexColumns.zipWithIndex.foreach { case (col, i) =>
      rtn(i) = if (indexCol2IdxInDictArray.contains(col.getColName)) {
        surrogatKeys(indexCol2IdxInDictArray(col.getColName)).toInt.asInstanceOf[Integer]
      } else if (indexCol2IdxInNoDictArray.contains(col.getColName)) {
        data(0).asInstanceOf[ByteArrayWrapper].getNoDictionaryKeyByIndex(
          indexCol2IdxInNoDictArray(col.getColName))
      } else {
        // measures start from 1
        val value = data(1 + indexCol2IdxInMeasureArray(col.getColName))
        if (null == value) {
          DataConvertUtil.getNullValueForMeasure(col.getDataType,
            col.getColumnSchema.getScale)
        } else if (DataTypes.isDecimal(col.getDataType)) {
          // In rebuild process, value is built for spark
          // here we need to convert it to java BigDecimal for carbon
          value.asInstanceOf[Decimal].toBigDecimal.bigDecimal
        } else {
          value
        }
      }
    }
    rtn(indexColumns.length) = data(data.length - 3)
    rtn(indexColumns.length + 1) = data(data.length - 2)
    rtn(indexColumns.length + 2) = data(data.length - 1)
    rtn
  }

  override def close(): Unit = {
  }
}

class IndexDataMapRebuildRDD[K, V](
    @transient private val session: SparkSession,
    result: RefreshResult[K, V],
    @transient private val tableInfo: TableInfo,
    dataMapName: String,
    indexColumns: Array[CarbonColumn],
    segments: Set[Segment])
  extends CarbonRDDWithTableInfo[(K, V)](session, Nil, tableInfo.serialize()) {

  private val dataMapSchema = DataMapStoreManager.getInstance().getDataMapSchema(dataMapName)
  private val queryId = sparkContext.getConf.get("queryId", System.nanoTime() + "")
  private val jobTrackerId: String = {
    val formatter = new SimpleDateFormat("yyyyMMddHHmm")
    formatter.format(new util.Date())
  }

  override def internalCompute(split: Partition, context: TaskContext): Iterator[(K, V)] = {
    val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)
    val carbonTable = CarbonTable.buildFromTableInfo(getTableInfo)
    val dataMapFactory = DataMapManager.get().getDataMapProvider(
      carbonTable, dataMapSchema, session).getDataMapFactory
    var status = false
    val inputMetrics = new CarbonInputMetrics
    TaskMetricsMap.getInstance().registerThreadCallback()
    val inputSplit = split.asInstanceOf[CarbonSparkPartition].split.value
    val segmentId = inputSplit.getAllSplits.get(0).getSegment.getSegmentNo
    val segment = segments.find(p => p.getSegmentNo.equals(segmentId))
    if (segment.isDefined) {
      inputMetrics.initBytesReadCallback(context, inputSplit)

      val attemptId = new TaskAttemptID(jobTrackerId, id, TaskType.MAP, split.index, 0)
      val attemptContext = new TaskAttemptContextImpl(FileFactory.getConfiguration, attemptId)
      val format = createInputFormat(segment.get, attemptContext)

      val model = format.createQueryModel(inputSplit, attemptContext)
      // one query id per table
      model.setQueryId(queryId)
      model.setVectorReader(false)
      model.setRequiredRowId(true)

      var reader: CarbonRecordReader[Array[Object]] = null
      var refresher: DataMapBuilder = null
      try {
        val segmentPropertiesFetcher = DataMapStoreManager.getInstance().getDataMap(carbonTable,
          BlockletDataMapFactory.DATA_MAP_SCHEMA).getDataMapFactory
          .asInstanceOf[SegmentPropertiesFetcher]
        val segmentProperties = segmentPropertiesFetcher.getSegmentProperties(segment.get)

        // we use task name as shard name to create the folder for this datamap
        val shardName = CarbonTablePath.getShardName(inputSplit.getAllSplits.get(0).getBlockPath)
        refresher = dataMapFactory.createBuilder(segment.get, shardName, segmentProperties)
        refresher.initialize()

        model.setForcedDetailRawQuery(refresher.isIndexForCarbonRawBytes)
        val readSupport = if (refresher.isIndexForCarbonRawBytes) {
          new RawBytesReadSupport(segmentProperties, indexColumns)
        } else {
          new OriginalReadSupport(indexColumns.map(_.getDataType))
        }
        reader = new CarbonRecordReader[Array[Object]](model, readSupport, inputMetrics,
          attemptContext.getConfiguration)
        reader.initialize(inputSplit, attemptContext)
        // skip clear datamap and we will do this adter rebuild
        reader.setSkipClearDataMapAtClose(true)

        // Note that blockletId in rowWithPosition does not work properly,
        // here we use another way to generate it.
        var blockletId = 0
        var firstRow = true
        while (reader.nextKeyValue()) {
          val rowWithPosition = reader.getCurrentValue
          val size = rowWithPosition.length
          val pageId = rowWithPosition(size - 2).asInstanceOf[Int]
          val rowId = rowWithPosition(size - 1).asInstanceOf[Int]

          if (!firstRow && pageId == 0 && rowId == 0) {
            blockletId = blockletId + 1
          } else {
            firstRow = false
          }
          refresher.addRow(blockletId, pageId, rowId, rowWithPosition)
        }

        refresher.finish()

        status = true
      } finally {
        if (reader != null) {
          try {
            reader.close()
          } catch {
            case ex: Throwable =>
              LOGGER.error(ex, "Failed to close reader")
          }
        }

        if (refresher != null) {
          try {
            refresher.close()
          } catch {
            case ex: Throwable =>
              LOGGER.error(ex, "Failed to close index writer")
          }
        }
      }
    }

    new Iterator[(K, V)] {

      var finished = false

      override def hasNext: Boolean = {
        !finished
      }

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


  private def createInputFormat(segment: Segment,
      attemptContext: TaskAttemptContextImpl) = {
    val format = new CarbonTableInputFormat[Object]
    val tableInfo1 = getTableInfo
    val conf = attemptContext.getConfiguration
    CarbonInputFormat.setTableInfo(conf, tableInfo1)
    CarbonInputFormat.setDatabaseName(conf, tableInfo1.getDatabaseName)
    CarbonInputFormat.setTableName(conf, tableInfo1.getFactTable.getTableName)
    CarbonInputFormat.setDataTypeConverter(conf, classOf[SparkDataTypeConverterImpl])

    val identifier = tableInfo1.getOrCreateAbsoluteTableIdentifier()
    CarbonInputFormat.setTablePath(
      conf,
      identifier.appendWithLocalPrefix(identifier.getTablePath))

    CarbonInputFormat.setSegmentsToAccess(
      conf,
      List(segment).asJava)

    CarbonInputFormat.setColumnProjection(
      conf,
      new CarbonProjection(indexColumns.map(_.getColName)))
    format
  }

  override protected def internalGetPartitions = {
    if (!dataMapSchema.isIndexDataMap) {
      throw new UnsupportedOperationException
    }
    val conf = FileFactory.getConfiguration
    val jobConf = new JobConf(conf)
    SparkHadoopUtil.get.addCredentials(jobConf)
    val job = Job.getInstance(jobConf)
    job.getConfiguration.set("query.id", queryId)

    val format = new CarbonTableInputFormat[Object]

    CarbonInputFormat.setSegmentsToAccess(
      job.getConfiguration,
      segments.toList.asJava)

    CarbonInputFormat.setTableInfo(
      job.getConfiguration,
      tableInfo)
    CarbonInputFormat.setTablePath(
      job.getConfiguration,
      tableInfo.getOrCreateAbsoluteTableIdentifier().getTablePath)
    CarbonInputFormat.setDatabaseName(
      job.getConfiguration,
      tableInfo.getDatabaseName)
    CarbonInputFormat.setTableName(
      job.getConfiguration,
      tableInfo.getFactTable.getTableName)

    format
      .getSplits(job)
      .asScala
      .map(_.asInstanceOf[CarbonInputSplit])
      .groupBy(p => (p.getSegmentId, p.taskId))
      .map { group =>
        new CarbonMultiBlockSplit(
          group._2.asJava,
          group._2.flatMap(_.getLocations).toArray)
      }
      .zipWithIndex
      .map { split =>
        new CarbonSparkPartition(id, split._2, split._1)
      }
      .toArray
  }
}

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.datamap
19
20	import java.io.{File, IOException}
21	import java.text.SimpleDateFormat
22	import java.util
23
24	import scala.collection.JavaConverters._
25	import scala.collection.mutable
26	import scala.collection.mutable.ListBuffer
27
28	import org.apache.commons.lang3.ArrayUtils
29	import org.apache.hadoop.conf.Configuration
30	import org.apache.hadoop.mapred.JobConf
31	import org.apache.hadoop.mapreduce.{Job, TaskAttemptID, TaskType}
32	import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl
33	import org.apache.spark.{CarbonInputMetrics, Partition, TaskContext}
34	import org.apache.spark.deploy.SparkHadoopUtil
35	import org.apache.spark.sql.SparkSession
36	import org.apache.spark.sql.types.Decimal
37
38	import org.apache.carbondata.common.logging.LogServiceFactory
39	import org.apache.carbondata.converter.SparkDataTypeConverterImpl
40	import org.apache.carbondata.core.constants.CarbonCommonConstants
41	import org.apache.carbondata.core.datamap.{DataMapStoreManager, Segment}
42	import org.apache.carbondata.core.datamap.dev.DataMapBuilder
43	import org.apache.carbondata.core.datastore.block.SegmentProperties
44	import org.apache.carbondata.core.datastore.impl.FileFactory
45	import org.apache.carbondata.core.indexstore.blockletindex.BlockletDataMapFactory
46	import org.apache.carbondata.core.indexstore.SegmentPropertiesFetcher
47	import org.apache.carbondata.core.keygenerator.KeyGenerator
48	import org.apache.carbondata.core.keygenerator.mdkey.MultiDimKeyVarLengthGenerator
49	import org.apache.carbondata.core.metadata.datatype.{DataType, DataTypes}
50	import org.apache.carbondata.core.metadata.encoder.Encoding
51	import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema, TableInfo}
52	import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn
53	import org.apache.carbondata.core.scan.wrappers.ByteArrayWrapper
54	import org.apache.carbondata.core.statusmanager.SegmentStatusManager
55	import org.apache.carbondata.core.util.{CarbonUtil, TaskMetricsMap}
56	import org.apache.carbondata.core.util.path.CarbonTablePath
57	import org.apache.carbondata.datamap.bloom.DataConvertUtil
58	import org.apache.carbondata.events.{BuildDataMapPostExecutionEvent, BuildDataMapPreExecutionEvent, OperationContext, OperationListenerBus}
59	import org.apache.carbondata.hadoop.{CarbonInputSplit, CarbonMultiBlockSplit, CarbonProjection, CarbonRecordReader}
60	import org.apache.carbondata.hadoop.api.{CarbonInputFormat, CarbonTableInputFormat}
61	import org.apache.carbondata.hadoop.readsupport.CarbonReadSupport
62	import org.apache.carbondata.spark.{RefreshResult, RefreshResultImpl}
63	import org.apache.carbondata.spark.rdd.{CarbonRDDWithTableInfo, CarbonSparkPartition}
64
65
66	/**
67	* Helper object to rebuild the index DataMap
68	*/
69	object IndexDataMapRebuildRDD {	×
70	private val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)	2✔
71
72	/**
73	* Rebuild the datamap for all existing data in the table
74	*/
75	def rebuildDataMap(
76	sparkSession: SparkSession,
77	carbonTable: CarbonTable,
78	schema: DataMapSchema
79	): Unit = {
80	val tableIdentifier = carbonTable.getAbsoluteTableIdentifier	2✔
81	val segmentStatusManager = new SegmentStatusManager(tableIdentifier)	2✔
82	val validAndInvalidSegments = segmentStatusManager.getValidAndInvalidSegments()	2✔
83	val validSegments = validAndInvalidSegments.getValidSegments	2✔
84	val indexedCarbonColumns = carbonTable.getIndexedColumns(schema)	2✔
85	val operationContext = new OperationContext()	2✔
86	val buildDataMapPreExecutionEvent = new BuildDataMapPreExecutionEvent(sparkSession,	2✔
87	tableIdentifier,	2✔
88	mutable.Seq[String](schema.getDataMapName))	2✔
89	OperationListenerBus.getInstance().fireEvent(buildDataMapPreExecutionEvent, operationContext)	2✔
90
91	val segments2DmStorePath = validSegments.asScala.map { segment =>	2✔
92	val dataMapStorePath = CarbonTablePath.getDataMapStorePath(carbonTable.getTablePath,	2✔
93	segment.getSegmentNo, schema.getDataMapName)	2✔
94	segment -> dataMapStorePath	2✔
95	}.filter(p => !FileFactory.isFileExist(p._2)).toMap	2✔
96
97	segments2DmStorePath.foreach { case (_, dmPath) =>	2✔
98	if (!FileFactory.mkdirs(dmPath, FileFactory.getFileType(dmPath))) {	2✔
99	throw new IOException(	×
100	s"Failed to create directory $dmPath for rebuilding datamap ${ schema.getDataMapName }")	×
101	}
102	}
103
104	val status = new IndexDataMapRebuildRDD[String, (String, Boolean)](	2✔
105	sparkSession,	2✔
106	new RefreshResultImpl(),	2✔
107	carbonTable.getTableInfo,	2✔
108	schema.getDataMapName,	2✔
109	indexedCarbonColumns.asScala.toArray,	2✔
110	segments2DmStorePath.keySet	2✔
111	).collect	2✔
112
113	// for failed segments, clean the result
114	val failedSegments = status	2✔
115	.find { case (taskId, (segmentId, rebuildStatus)) =>	2✔
116	!rebuildStatus	2✔
117	}
118	.map { task =>	2✔
119	val segmentId = task._2._1	×
120	val dmPath = segments2DmStorePath.filter(p => p._1.getSegmentNo.equals(segmentId)).values	×
121	val cleanResult = dmPath.map(p =>	×
122	FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(p)))	×
123	if (cleanResult.exists(!_)) {	×
124	LOGGER.error(s"Failed to clean up datamap store for segment_$segmentId")	×
125	false	×
126	} else {
127	true	×
128	}
129	}
130
131	if (failedSegments.nonEmpty) {	2✔
132	throw new Exception(s"Failed to refresh datamap ${ schema.getDataMapName }")	×
133	}
134
135	val buildDataMapPostExecutionEvent = new BuildDataMapPostExecutionEvent(sparkSession,	2✔
136	tableIdentifier, validSegments.asScala.map(_.getSegmentNo), true)	2✔
137	OperationListenerBus.getInstance().fireEvent(buildDataMapPostExecutionEvent, operationContext)	2✔
138	}
139	}
140
141	class OriginalReadSupport(dataTypes: Array[DataType]) extends CarbonReadSupport[Array[Object]] {	2✔
142	override def initialize(carbonColumns: Array[CarbonColumn],
143	carbonTable: CarbonTable): Unit = {	×
144	}
145
146	override def readRow(data: Array[Object]): Array[Object] = {
147	dataTypes.zipWithIndex.foreach { case (dataType, i) =>	×
148	if (dataType == DataTypes.STRING && data(i) != null) {	×
149	data(i) = data(i).toString	×
150	}
151	}
152	data	×
153	}
154
155	override def close(): Unit = {	×
156	}
157	}
158
159	/**
160	* This class will generate row value which is raw bytes for the dimensions.
161	*/
162	class RawBytesReadSupport(segmentProperties: SegmentProperties, indexColumns: Array[CarbonColumn])	2✔
163	extends CarbonReadSupport[Array[Object]] {	2✔
164	var dimensionKeyGenerator: KeyGenerator = _	2✔
165	// for the dictionary dimensions
166	var indexCol2IdxInDictArray: Map[String, Int] = Map()	2✔
167	// for the non dictionary dimensions
168	var indexCol2IdxInNoDictArray: Map[String, Int] = Map()	2✔
169	// for the measures
170	var indexCol2IdxInMeasureArray: Map[String, Int] = Map()	2✔
171
172	/**
173	* rebuild process get data from query, if some columns added to table but not in this segment
174	* it will be filled with default value and generate new key for dict dimension.
175	* Here we use same way as `RowIdRestructureBasedRawResultCollector` to prepare
176	* key generator to get surrogate value of dict column result.
177	* So we do not need to make a fake mdk to split when adding row to datamap
178	*/
179	def prepareKeyGenForDictIndexColumns(carbonTable: CarbonTable,
180	dictIndexColumns: ListBuffer[CarbonColumn]): Unit = {	2✔
181
182	val columnCardinality = new util.ArrayList[Integer](dictIndexColumns.length)	2✔
183	val columnPartitioner = new util.ArrayList[Integer](dictIndexColumns.length)	2✔
184
185	dictIndexColumns.foreach { col =>	2✔
186	val dim = carbonTable.getDimensionByName(carbonTable.getTableName, col.getColName)	2✔
187	val currentBlockDimension = segmentProperties.getDimensionFromCurrentBlock(dim)	2✔
188	if (null != currentBlockDimension) {	2✔
189	columnCardinality.add(segmentProperties.getDimColumnsCardinality.apply(	2✔
190	currentBlockDimension.getKeyOrdinal))	2✔
191	columnPartitioner.add(segmentProperties.getDimensionPartitions.apply(	2✔
192	currentBlockDimension.getKeyOrdinal	2✔
193	))
194	} else {
195	columnPartitioner.add(1)	2✔
196	if (col.hasEncoding(Encoding.DIRECT_DICTIONARY)) {	2✔
197	columnCardinality.add(Integer.MAX_VALUE)	×
198	} else {
199	val defaultValue = col.getDefaultValue	2✔
200	if (null != col.getDefaultValue) {	2✔
201	columnCardinality.add(CarbonCommonConstants.DICTIONARY_DEFAULT_CARDINALITY + 1)	2✔
202	} else {
203	columnCardinality.add(CarbonCommonConstants.DICTIONARY_DEFAULT_CARDINALITY)	×
204	}
205	}
206	}
207	}
208
209	if (!columnCardinality.isEmpty) {	2✔
210	val latestColumnCardinality = ArrayUtils.toPrimitive(columnCardinality.toArray(	2✔
211	new Array[Integer](columnCardinality.size)))	2✔
212	val latestColumnPartitioner = ArrayUtils.toPrimitive(columnPartitioner.toArray(	2✔
213	new Array[Integer](columnPartitioner.size)))	2✔
214	val dimensionBitLength = CarbonUtil.getDimensionBitLength(	2✔
215	latestColumnCardinality, latestColumnPartitioner)	2✔
216	this.dimensionKeyGenerator = new MultiDimKeyVarLengthGenerator(dimensionBitLength)	2✔
217	}
218	}
219
220	override def initialize(carbonColumns: Array[CarbonColumn],
221	carbonTable: CarbonTable): Unit = {	2✔
222
223	val dictIndexColumns = new ListBuffer[CarbonColumn]()	2✔
224
225	// prepare index info to extract data from query result
226	indexColumns.foreach { col =>	2✔
227	if (col.isDimension) {	2✔
228	val dim = carbonTable.getDimensionByName(carbonTable.getTableName, col.getColName)	2✔
229	if (!dim.isGlobalDictionaryEncoding && !dim.isDirectDictionaryEncoding) {	2✔
230	indexCol2IdxInNoDictArray =	2✔
231	indexCol2IdxInNoDictArray + (col.getColName -> indexCol2IdxInNoDictArray.size)	2✔
232	} else {
233	dictIndexColumns.append(col)	2✔
234	indexCol2IdxInDictArray =	2✔
235	indexCol2IdxInDictArray + (col.getColName -> indexCol2IdxInDictArray.size)	2✔
236	}
237	} else {
238	indexCol2IdxInMeasureArray =	2✔
239	indexCol2IdxInMeasureArray + (col.getColName -> indexCol2IdxInMeasureArray.size)	2✔
240	}
241	}
242
243	if (dictIndexColumns.size > 0) {	2✔
244	prepareKeyGenForDictIndexColumns(carbonTable, dictIndexColumns)	2✔
245	}
246	}
247
248	/**
249	* input: all the dimensions are bundled in one ByteArrayWrapper in position 0,
250	* then comes the measures one by one; last 3 elements are block/page/row id
251	* output: all the dimensions and measures comes one after another
252	*/
253	override def readRow(data: Array[Object]): Array[Object] = {
254
255	var surrogatKeys = new Array[Long](0)	2✔
256	if(null != dimensionKeyGenerator) {	2✔
257	surrogatKeys = dimensionKeyGenerator.getKeyArray(	2✔
258	data(0).asInstanceOf[ByteArrayWrapper].getDictionaryKey)	2✔
259	}
260
261	// fill return row from data
262	val rtn = new Array[Object](indexColumns.length + 3)	2✔
263	indexColumns.zipWithIndex.foreach { case (col, i) =>	2✔
264	rtn(i) = if (indexCol2IdxInDictArray.contains(col.getColName)) {	2✔
265	surrogatKeys(indexCol2IdxInDictArray(col.getColName)).toInt.asInstanceOf[Integer]	2✔
266	} else if (indexCol2IdxInNoDictArray.contains(col.getColName)) {	2✔
267	data(0).asInstanceOf[ByteArrayWrapper].getNoDictionaryKeyByIndex(	2✔
268	indexCol2IdxInNoDictArray(col.getColName))	2✔
269	} else {
270	// measures start from 1
271	val value = data(1 + indexCol2IdxInMeasureArray(col.getColName))	2✔
272	if (null == value) {	2✔
273	DataConvertUtil.getNullValueForMeasure(col.getDataType,	2✔
274	col.getColumnSchema.getScale)	2✔
275	} else if (DataTypes.isDecimal(col.getDataType)) {	2✔
276	// In rebuild process, value is built for spark
277	// here we need to convert it to java BigDecimal for carbon
278	value.asInstanceOf[Decimal].toBigDecimal.bigDecimal	2✔
279	} else {
280	value	2✔
281	}
282	}
283	}
284	rtn(indexColumns.length) = data(data.length - 3)	2✔
285	rtn(indexColumns.length + 1) = data(data.length - 2)	2✔
286	rtn(indexColumns.length + 2) = data(data.length - 1)	2✔
287	rtn	2✔
288	}
289
290	override def close(): Unit = {	2✔
291	}
292	}
293
294	class IndexDataMapRebuildRDD[K, V](
295	@transient private val session: SparkSession,	2✔
296	result: RefreshResult[K, V],	2✔
297	@transient private val tableInfo: TableInfo,	2✔
298	dataMapName: String,
299	indexColumns: Array[CarbonColumn],	2✔
300	segments: Set[Segment])	2✔
301	extends CarbonRDDWithTableInfo[(K, V)](session, Nil, tableInfo.serialize()) {	2✔
302
303	private val dataMapSchema = DataMapStoreManager.getInstance().getDataMapSchema(dataMapName)	2✔
304	private val queryId = sparkContext.getConf.get("queryId", System.nanoTime() + "")	2✔
305	private val jobTrackerId: String = {	2✔
306	val formatter = new SimpleDateFormat("yyyyMMddHHmm")	2✔
307	formatter.format(new util.Date())	2✔
308	}
309
310	override def internalCompute(split: Partition, context: TaskContext): Iterator[(K, V)] = {
311	val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)	2✔
312	val carbonTable = CarbonTable.buildFromTableInfo(getTableInfo)	2✔
313	val dataMapFactory = DataMapManager.get().getDataMapProvider(	2✔
314	carbonTable, dataMapSchema, session).getDataMapFactory	2✔
315	var status = false	2✔
316	val inputMetrics = new CarbonInputMetrics	2✔
317	TaskMetricsMap.getInstance().registerThreadCallback()	2✔
318	val inputSplit = split.asInstanceOf[CarbonSparkPartition].split.value	2✔
319	val segmentId = inputSplit.getAllSplits.get(0).getSegment.getSegmentNo	2✔
320	val segment = segments.find(p => p.getSegmentNo.equals(segmentId))	2✔
321	if (segment.isDefined) {	2✔
322	inputMetrics.initBytesReadCallback(context, inputSplit)	2✔
323
324	val attemptId = new TaskAttemptID(jobTrackerId, id, TaskType.MAP, split.index, 0)	2✔
325	val attemptContext = new TaskAttemptContextImpl(FileFactory.getConfiguration, attemptId)	2✔
326	val format = createInputFormat(segment.get, attemptContext)	2✔
327
328	val model = format.createQueryModel(inputSplit, attemptContext)	2✔
329	// one query id per table
330	model.setQueryId(queryId)	2✔
331	model.setVectorReader(false)	2✔
332	model.setRequiredRowId(true)	2✔
333
334	var reader: CarbonRecordReader[Array[Object]] = null	2✔
335	var refresher: DataMapBuilder = null	2✔
336	try {
337	val segmentPropertiesFetcher = DataMapStoreManager.getInstance().getDataMap(carbonTable,	2✔
338	BlockletDataMapFactory.DATA_MAP_SCHEMA).getDataMapFactory	2✔
339	.asInstanceOf[SegmentPropertiesFetcher]
340	val segmentProperties = segmentPropertiesFetcher.getSegmentProperties(segment.get)	2✔
341
342	// we use task name as shard name to create the folder for this datamap
343	val shardName = CarbonTablePath.getShardName(inputSplit.getAllSplits.get(0).getBlockPath)	2✔
344	refresher = dataMapFactory.createBuilder(segment.get, shardName, segmentProperties)	2✔
345	refresher.initialize()	2✔
346
347	model.setForcedDetailRawQuery(refresher.isIndexForCarbonRawBytes)	2✔
348	val readSupport = if (refresher.isIndexForCarbonRawBytes) {	2✔
349	new RawBytesReadSupport(segmentProperties, indexColumns)	2✔
350	} else {
351	new OriginalReadSupport(indexColumns.map(_.getDataType))	×
352	}
353	reader = new CarbonRecordReader[Array[Object]](model, readSupport, inputMetrics,	2✔
354	attemptContext.getConfiguration)	2✔
355	reader.initialize(inputSplit, attemptContext)	2✔
356	// skip clear datamap and we will do this adter rebuild
357	reader.setSkipClearDataMapAtClose(true)	2✔
358
359	// Note that blockletId in rowWithPosition does not work properly,
360	// here we use another way to generate it.
361	var blockletId = 0	2✔
362	var firstRow = true	2✔
363	while (reader.nextKeyValue()) {	2✔
364	val rowWithPosition = reader.getCurrentValue	2✔
365	val size = rowWithPosition.length	2✔
366	val pageId = rowWithPosition(size - 2).asInstanceOf[Int]	2✔
367	val rowId = rowWithPosition(size - 1).asInstanceOf[Int]	2✔
368
369	if (!firstRow && pageId == 0 && rowId == 0) {	2✔
370	blockletId = blockletId + 1	2✔
371	} else {
372	firstRow = false	2✔
373	}
374	refresher.addRow(blockletId, pageId, rowId, rowWithPosition)	2✔
375	}
376
377	refresher.finish()	2✔
378
379	status = true	2✔
380	} finally {	×
381	if (reader != null) {	2✔
382	try {	×
383	reader.close()	2✔
384	} catch {
385	case ex: Throwable =>
386	LOGGER.error(ex, "Failed to close reader")	×
387	}
388	}
389
390	if (refresher != null) {	2✔
391	try {	×
392	refresher.close()	2✔
393	} catch {
394	case ex: Throwable =>
395	LOGGER.error(ex, "Failed to close index writer")	×
396	}
397	}
398	}
399	}
400
401	new Iterator[(K, V)] {	2✔
402
403	var finished = false	2✔
404
405	override def hasNext: Boolean = {
406	!finished	2✔
407	}
408
409	override def next(): (K, V) = {
410	finished = true	2✔
411	result.getKey(split.index.toString, (segmentId, status))	2✔
412	}
413	}
414	}
415
416
417	private def createInputFormat(segment: Segment,
418	attemptContext: TaskAttemptContextImpl) = {
419	val format = new CarbonTableInputFormat[Object]	2✔
420	val tableInfo1 = getTableInfo	2✔
421	val conf = attemptContext.getConfiguration	2✔
422	CarbonInputFormat.setTableInfo(conf, tableInfo1)	2✔
423	CarbonInputFormat.setDatabaseName(conf, tableInfo1.getDatabaseName)	2✔
424	CarbonInputFormat.setTableName(conf, tableInfo1.getFactTable.getTableName)	2✔
425	CarbonInputFormat.setDataTypeConverter(conf, classOf[SparkDataTypeConverterImpl])	2✔
426
427	val identifier = tableInfo1.getOrCreateAbsoluteTableIdentifier()	2✔
428	CarbonInputFormat.setTablePath(	2✔
429	conf,	2✔
430	identifier.appendWithLocalPrefix(identifier.getTablePath))	2✔
431
432	CarbonInputFormat.setSegmentsToAccess(	2✔
433	conf,	2✔
434	List(segment).asJava)	2✔
435
436	CarbonInputFormat.setColumnProjection(	2✔
437	conf,	2✔
438	new CarbonProjection(indexColumns.map(_.getColName)))	2✔
439	format	2✔
440	}
441
442	override protected def internalGetPartitions = {
443	if (!dataMapSchema.isIndexDataMap) {	2✔
444	throw new UnsupportedOperationException	×
445	}
446	val conf = FileFactory.getConfiguration	2✔
447	val jobConf = new JobConf(conf)	2✔
448	SparkHadoopUtil.get.addCredentials(jobConf)	2✔
449	val job = Job.getInstance(jobConf)	2✔
450	job.getConfiguration.set("query.id", queryId)	2✔
451
452	val format = new CarbonTableInputFormat[Object]	2✔
453
454	CarbonInputFormat.setSegmentsToAccess(	2✔
455	job.getConfiguration,	2✔
456	segments.toList.asJava)	2✔
457
458	CarbonInputFormat.setTableInfo(	2✔
459	job.getConfiguration,	2✔
460	tableInfo)	2✔
461	CarbonInputFormat.setTablePath(	2✔
462	job.getConfiguration,	2✔
463	tableInfo.getOrCreateAbsoluteTableIdentifier().getTablePath)	2✔
464	CarbonInputFormat.setDatabaseName(	2✔
465	job.getConfiguration,	2✔
466	tableInfo.getDatabaseName)	2✔
467	CarbonInputFormat.setTableName(	2✔
468	job.getConfiguration,	2✔
469	tableInfo.getFactTable.getTableName)	2✔
470
471	format	2✔
472	.getSplits(job)	2✔
473	.asScala	2✔
474	.map(_.asInstanceOf[CarbonInputSplit])	2✔
475	.groupBy(p => (p.getSegmentId, p.taskId))	2✔
476	.map { group =>	2✔
477	new CarbonMultiBlockSplit(	2✔
478	group._2.asJava,	2✔
479	group._2.flatMap(_.getLocations).toArray)	2✔
480	}
481	.zipWithIndex	2✔
482	.map { split =>	2✔
483	new CarbonSparkPartition(id, split._2, split._1)	2✔
484	}
485	.toArray	2✔
486	}
487	}

apache / carbondata / 8535

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