11878

Build # 11878

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/3aa635b7d7ba6aee6eaff4c434543127db0d592b">3aa635b7d<a href="https://github.com/apache/carbondata/commit/3aa635b7d7ba6aee6eaff4c434543127db0d592b">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/a1b66f5753f69fe249f615f37186498e226db4a6">a1b66f575</a>">a1b66f575</a><a href="https://github.com/apache/carbondata/commit/3aa635b7d7ba6aee6eaff4c434543127db0d592b"> into 581591a38">581591a38</a>

Run Details

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

64102 of 82819 relevant lines covered (77.4%)

1.01 hits per line

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

83.33

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

import java.util

import scala.collection.JavaConverters._

import org.apache.spark.{Partition, TaskContext}
import org.apache.spark.sql.SparkSession

import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.metadata.SegmentFileStore
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.statusmanager.SegmentStatusManager
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.core.util.path.CarbonTablePath
import org.apache.carbondata.core.writer.CarbonIndexFileMergeWriter
import org.apache.carbondata.processing.util.CarbonLoaderUtil
import org.apache.carbondata.spark.rdd.CarbonRDD

case class CarbonMergeFilePartition(rddId: Int,
    idx: Int,
    segmentId: String,
    partitionPath: String = null) extends Partition {

  override val index: Int = idx

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

object CarbonMergeFilesRDD {

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

  /**
   * Merge the carbonindex files with in the segment to carbonindexmerge file inside same segment
   *
   * @param sparkSession carbon session
   * @param segmentIds the segments to process
   * @param segmentFileNameToSegmentIdMap a map that map the segmentFileName to segmentId
   * @param tablePath table path
   * @param carbonTable carbon table
   * @param mergeIndexProperty whether to merge the property of the carbon index, the usage
   *                           scenario is the same as that of `readFileFooterFromCarbonDataFile`
   * @param readFileFooterFromCarbonDataFile flag to read file footer information from carbondata
   *                                         file. This will used in case of upgrade from version
   *                                         which do not store the blocklet info to current
   *                                         version
   */
  def mergeIndexFiles(sparkSession: SparkSession,
      segmentIds: Seq[String],
      segmentFileNameToSegmentIdMap: java.util.Map[String, String],
      tablePath: String,
      carbonTable: CarbonTable,
      mergeIndexProperty: Boolean,
      readFileFooterFromCarbonDataFile: Boolean = false): Unit = {
    if (mergeIndexProperty) {
      new CarbonMergeFilesRDD(
        sparkSession,
        carbonTable,
        segmentIds,
        segmentFileNameToSegmentIdMap,
        carbonTable.isHivePartitionTable,
        readFileFooterFromCarbonDataFile).collect()
    } else {
      try {
        if (isPropertySet(CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT,
          CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT_DEFAULT)) {
          new CarbonMergeFilesRDD(
            sparkSession,
            carbonTable,
            segmentIds,
            segmentFileNameToSegmentIdMap,
            carbonTable.isHivePartitionTable,
            readFileFooterFromCarbonDataFile).collect()
        }
      } catch {
        case ex: Exception =>
          val message = "Merge Index files request is failed " +
                        s"for table ${ carbonTable.getTableUniqueName }. " + ex.getMessage
          LOGGER.error(message)
          if (isPropertySet(CarbonCommonConstants.CARBON_MERGE_INDEX_FAILURE_THROW_EXCEPTION,
            CarbonCommonConstants.CARBON_MERGE_INDEX_FAILURE_THROW_EXCEPTION_DEFAULT)) {
            throw new RuntimeException(message, ex)
          }
      }
    }
    if (carbonTable.isHivePartitionTable) {
      segmentIds.foreach(segmentId => {
        val readPath: String = CarbonTablePath.getSegmentFilesLocation(tablePath) +
                               CarbonCommonConstants.FILE_SEPARATOR + segmentId + "_" +
                               segmentFileNameToSegmentIdMap.get(segmentId) + ".tmp"
        // Merge all partition files into a single file.
        val segmentFileName: String = SegmentFileStore
          .genSegmentFileName(segmentId, segmentFileNameToSegmentIdMap.get(segmentId))
        SegmentFileStore
          .mergeSegmentFiles(readPath,
            segmentFileName,
            CarbonTablePath.getSegmentFilesLocation(tablePath))
      })
    }
  }

  /**
   * Check whether the Merge Index Property is set by the user.
   * If not set, take the default value of the property.
   *
   * @return
   */
  def isPropertySet(property: String, defaultValue: String): Boolean = {
    var mergeIndex: Boolean = false
    try {
      mergeIndex = CarbonProperties.getInstance().getProperty(property, defaultValue).toBoolean
    } catch {
      case _: Exception =>
        mergeIndex = defaultValue.toBoolean
    }
    mergeIndex
  }
}

/**
 * RDD to merge all carbonindex files of each segment to carbonindex file into the same segment.
 * @param ss
 * @param carbonTable
 * @param segments segments to be merged
 */
class CarbonMergeFilesRDD(
  @transient private val ss: SparkSession,
  carbonTable: CarbonTable,
  segments: Seq[String],
  segmentFileNameToSegmentIdMap: java.util.Map[String, String],
  isHivePartitionedTable: Boolean,
  readFileFooterFromCarbonDataFile: Boolean)
  extends CarbonRDD[String](ss, Nil) {

  override def internalGetPartitions: Array[Partition] = {
    if (isHivePartitionedTable) {
      val metadataDetails = SegmentStatusManager
        .readLoadMetadata(CarbonTablePath.getMetadataPath(carbonTable.getTablePath))
      // in case of partition table make rdd partitions per partition of the carbon table
      val partitionPaths: java.util.Map[String, java.util.List[String]] = new java.util.HashMap()
      segments.foreach(segment => {
        val partitionSpecs = SegmentFileStore
          .getPartitionSpecs(segment, carbonTable.getTablePath, metadataDetails)
          .asScala.map(_.getLocation.toString)
        partitionPaths.put(segment, partitionSpecs.asJava)
      })
      var index: Int = -1
      val rddPartitions: java.util.List[Partition] = new java.util.ArrayList()
      partitionPaths.asScala.foreach(partitionPath => {
        val segmentId = partitionPath._1
        partitionPath._2.asScala.map { partition =>
          index = index + 1
          rddPartitions.add(CarbonMergeFilePartition(id, index, segmentId, partition))
        }
      })
      rddPartitions.asScala.toArray
    } else {
      // in case of normal carbon table, make rdd partitions per segment
      segments.zipWithIndex.map { s =>
        CarbonMergeFilePartition(id, s._2, s._1)
      }.toArray
    }
  }

  override def internalCompute(theSplit: Partition, context: TaskContext): Iterator[String] = {
    val tablePath = carbonTable.getTablePath
    val iter = new Iterator[String] {
      val split = theSplit.asInstanceOf[CarbonMergeFilePartition]
      logInfo("Merging carbon index files of segment : " +
              CarbonTablePath.getSegmentPath(tablePath, split.segmentId))

      if (isHivePartitionedTable) {
        CarbonLoaderUtil
          .mergeIndexFilesInPartitionedSegment(carbonTable, split.segmentId,
            segmentFileNameToSegmentIdMap.get(split.segmentId), split.partitionPath)
      } else {
        new CarbonIndexFileMergeWriter(carbonTable)
          .mergeCarbonIndexFilesOfSegment(split.segmentId,
            tablePath,
            readFileFooterFromCarbonDataFile,
            segmentFileNameToSegmentIdMap.get(split.segmentId))
      }

      var havePair = false
      var finished = false

      override def hasNext: Boolean = {
        if (!finished && !havePair) {
          finished = true
          havePair = !finished
        }
        !finished
      }

      override def next(): String = {
        if (!hasNext) {
          throw new java.util.NoSuchElementException("End of stream")
        }
        havePair = false
        ""
      }

    }
    iter
  }

}


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.spark.rdd
19
20	import java.util
21
22	import scala.collection.JavaConverters._
23
24	import org.apache.spark.{Partition, TaskContext}
25	import org.apache.spark.sql.SparkSession
26
27	import org.apache.carbondata.common.logging.LogServiceFactory
28	import org.apache.carbondata.core.constants.CarbonCommonConstants
29	import org.apache.carbondata.core.metadata.SegmentFileStore
30	import org.apache.carbondata.core.metadata.schema.table.CarbonTable
31	import org.apache.carbondata.core.statusmanager.SegmentStatusManager
32	import org.apache.carbondata.core.util.CarbonProperties
33	import org.apache.carbondata.core.util.path.CarbonTablePath
34	import org.apache.carbondata.core.writer.CarbonIndexFileMergeWriter
35	import org.apache.carbondata.processing.util.CarbonLoaderUtil
36	import org.apache.carbondata.spark.rdd.CarbonRDD
37
38	case class CarbonMergeFilePartition(rddId: Int,	1✔
39	idx: Int,	1✔
40	segmentId: String,	1✔
41	partitionPath: String = null) extends Partition {	1✔
42
43	override val index: Int = idx	1✔
44
45	override def hashCode(): Int = 41 * (41 + rddId) + idx	×
46	}
47
48	object CarbonMergeFilesRDD {	×
49
50	private val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)	1✔
51
52	/**
53	* Merge the carbonindex files with in the segment to carbonindexmerge file inside same segment
54	*
55	* @param sparkSession carbon session
56	* @param segmentIds the segments to process
57	* @param segmentFileNameToSegmentIdMap a map that map the segmentFileName to segmentId
58	* @param tablePath table path
59	* @param carbonTable carbon table
60	* @param mergeIndexProperty whether to merge the property of the carbon index, the usage
61	* scenario is the same as that of `readFileFooterFromCarbonDataFile`
62	* @param readFileFooterFromCarbonDataFile flag to read file footer information from carbondata
63	* file. This will used in case of upgrade from version
64	* which do not store the blocklet info to current
65	* version
66	*/
67	def mergeIndexFiles(sparkSession: SparkSession,
68	segmentIds: Seq[String],
69	segmentFileNameToSegmentIdMap: java.util.Map[String, String],
70	tablePath: String,
71	carbonTable: CarbonTable,
72	mergeIndexProperty: Boolean,
73	readFileFooterFromCarbonDataFile: Boolean = false): Unit = {	1✔
74	if (mergeIndexProperty) {	1✔
75	new CarbonMergeFilesRDD(	1✔
76	sparkSession,	1✔
77	carbonTable,	1✔
78	segmentIds,	1✔
79	segmentFileNameToSegmentIdMap,	1✔
80	carbonTable.isHivePartitionTable,	1✔
81	readFileFooterFromCarbonDataFile).collect()	1✔
82	} else {
83	try {	×
84	if (isPropertySet(CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT,	1✔
85	CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT_DEFAULT)) {	1✔
86	new CarbonMergeFilesRDD(	1✔
87	sparkSession,	1✔
88	carbonTable,	1✔
89	segmentIds,	1✔
90	segmentFileNameToSegmentIdMap,	1✔
91	carbonTable.isHivePartitionTable,	1✔
92	readFileFooterFromCarbonDataFile).collect()	1✔
93	}
94	} catch {
95	case ex: Exception =>	×
96	val message = "Merge Index files request is failed " +	×
97	s"for table ${ carbonTable.getTableUniqueName }. " + ex.getMessage	×
98	LOGGER.error(message)	×
99	if (isPropertySet(CarbonCommonConstants.CARBON_MERGE_INDEX_FAILURE_THROW_EXCEPTION,	×
100	CarbonCommonConstants.CARBON_MERGE_INDEX_FAILURE_THROW_EXCEPTION_DEFAULT)) {	×
101	throw new RuntimeException(message, ex)	×
102	}
103	}
104	}
105	if (carbonTable.isHivePartitionTable) {	1✔
106	segmentIds.foreach(segmentId => {	1✔
107	val readPath: String = CarbonTablePath.getSegmentFilesLocation(tablePath) +	1✔
108	CarbonCommonConstants.FILE_SEPARATOR + segmentId + "_" +	1✔
109	segmentFileNameToSegmentIdMap.get(segmentId) + ".tmp"	1✔
110	// Merge all partition files into a single file.
111	val segmentFileName: String = SegmentFileStore	1✔
112	.genSegmentFileName(segmentId, segmentFileNameToSegmentIdMap.get(segmentId))	1✔
113	SegmentFileStore
114	.mergeSegmentFiles(readPath,	1✔
115	segmentFileName,	1✔
116	CarbonTablePath.getSegmentFilesLocation(tablePath))	1✔
117	})
118	}
119	}
120
121	/**
122	* Check whether the Merge Index Property is set by the user.
123	* If not set, take the default value of the property.
124	*
125	* @return
126	*/
127	def isPropertySet(property: String, defaultValue: String): Boolean = {
128	var mergeIndex: Boolean = false	1✔
129	try {	×
130	mergeIndex = CarbonProperties.getInstance().getProperty(property, defaultValue).toBoolean	1✔
131	} catch {
132	case _: Exception =>
133	mergeIndex = defaultValue.toBoolean	×
134	}
135	mergeIndex	1✔
136	}
137	}
138
139	/**
140	* RDD to merge all carbonindex files of each segment to carbonindex file into the same segment.
141	* @param ss
142	* @param carbonTable
143	* @param segments segments to be merged
144	*/
145	class CarbonMergeFilesRDD(	1✔
146	@transient private val ss: SparkSession,	1✔
147	carbonTable: CarbonTable,	1✔
148	segments: Seq[String],	1✔
149	segmentFileNameToSegmentIdMap: java.util.Map[String, String],	1✔
150	isHivePartitionedTable: Boolean,	1✔
151	readFileFooterFromCarbonDataFile: Boolean)	1✔
152	extends CarbonRDD[String](ss, Nil) {	1✔
153
154	override def internalGetPartitions: Array[Partition] = {
155	if (isHivePartitionedTable) {	1✔
156	val metadataDetails = SegmentStatusManager	1✔
157	.readLoadMetadata(CarbonTablePath.getMetadataPath(carbonTable.getTablePath))	1✔
158	// in case of partition table make rdd partitions per partition of the carbon table
159	val partitionPaths: java.util.Map[String, java.util.List[String]] = new java.util.HashMap()	1✔
160	segments.foreach(segment => {	1✔
161	val partitionSpecs = SegmentFileStore	1✔
162	.getPartitionSpecs(segment, carbonTable.getTablePath, metadataDetails)	1✔
163	.asScala.map(_.getLocation.toString)	1✔
164	partitionPaths.put(segment, partitionSpecs.asJava)	1✔
165	})
166	var index: Int = -1	1✔
167	val rddPartitions: java.util.List[Partition] = new java.util.ArrayList()	1✔
168	partitionPaths.asScala.foreach(partitionPath => {	1✔
169	val segmentId = partitionPath._1	1✔
170	partitionPath._2.asScala.map { partition =>	1✔
171	index = index + 1	1✔
172	rddPartitions.add(CarbonMergeFilePartition(id, index, segmentId, partition))	1✔
173	}
174	})
175	rddPartitions.asScala.toArray	1✔
176	} else {
177	// in case of normal carbon table, make rdd partitions per segment
178	segments.zipWithIndex.map { s =>	1✔
179	CarbonMergeFilePartition(id, s._2, s._1)	1✔
180	}.toArray	1✔
181	}
182	}
183
184	override def internalCompute(theSplit: Partition, context: TaskContext): Iterator[String] = {
185	val tablePath = carbonTable.getTablePath	1✔
186	val iter = new Iterator[String] {	1✔
187	val split = theSplit.asInstanceOf[CarbonMergeFilePartition]	1✔
188	logInfo("Merging carbon index files of segment : " +	1✔
189	CarbonTablePath.getSegmentPath(tablePath, split.segmentId))	×
190
191	if (isHivePartitionedTable) {	1✔
192	CarbonLoaderUtil
193	.mergeIndexFilesInPartitionedSegment(carbonTable, split.segmentId,	1✔
194	segmentFileNameToSegmentIdMap.get(split.segmentId), split.partitionPath)	1✔
195	} else {
196	new CarbonIndexFileMergeWriter(carbonTable)	1✔
197	.mergeCarbonIndexFilesOfSegment(split.segmentId,	1✔
198	tablePath,	1✔
199	readFileFooterFromCarbonDataFile,	1✔
200	segmentFileNameToSegmentIdMap.get(split.segmentId))	1✔
201	}
202
203	var havePair = false	1✔
204	var finished = false	1✔
205
206	override def hasNext: Boolean = {	1✔
207	if (!finished && !havePair) {	1✔
208	finished = true	1✔
209	havePair = !finished	1✔
210	}
211	!finished	1✔
212	}
213
214	override def next(): String = {
215	if (!hasNext) {	×
216	throw new java.util.NoSuchElementException("End of stream")	×
217	}
218	havePair = false	×
219	""	×
220	}
221
222	}
223	iter	1✔
224	}
225
226	}
227

apache / carbondata / 11878

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