Github user ksimar commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1320#discussion_r137186127
--- Diff:
integration/spark-common/src/main/scala/org/apache/spark/util/PartitionUtils.scala
---
@@ -80,37 +80,42 @@ object PartitionUtils {
dateFormatter: SimpleDateFormat): Unit = {
val columnDataType =
partitionInfo.getColumnSchemaList.get(0).getDataType
val index = partitionIdList.indexOf(partitionId)
- if (partitionInfo.getPartitionType == PartitionType.RANGE) {
- val rangeInfo = partitionInfo.getRangeInfo.asScala.toList
- val newRangeInfo = partitionId match {
- case 0 => rangeInfo ++ splitInfo
- case _ => rangeInfo.take(index - 1) ++ splitInfo ++
- rangeInfo.takeRight(rangeInfo.size - index)
+ if (index >= 0) {
+ if (partitionInfo.getPartitionType == PartitionType.RANGE) {
+ val rangeInfo = partitionInfo.getRangeInfo.asScala.toList
+ val newRangeInfo = partitionId match {
+ case 0 => rangeInfo ++ splitInfo
+ case _ => rangeInfo.take(index - 1) ++ splitInfo ++
+ rangeInfo.takeRight(rangeInfo.size - index)
+ }
+ CommonUtil.validateRangeInfo(newRangeInfo, columnDataType,
+ timestampFormatter, dateFormatter)
+ partitionInfo.setRangeInfo(newRangeInfo.asJava)
+ } else if (partitionInfo.getPartitionType == PartitionType.LIST) {
+ val originList =
partitionInfo.getListInfo.asScala.map(_.asScala.toList).toList
+ if (partitionId != 0) {
+ val targetListInfo = partitionInfo.getListInfo.get(index - 1)
+ CommonUtil.validateSplitListInfo(targetListInfo.asScala.toList,
splitInfo, originList)
+ } else {
+ CommonUtil.validateAddListInfo(splitInfo, originList)
+ }
+ val addListInfo =
PartitionUtils.getListInfo(splitInfo.mkString(","))
+ val newListInfo = partitionId match {
+ case 0 => originList ++ addListInfo
+ case _ => originList.take(index - 1) ++ addListInfo ++
+ originList.takeRight(originList.size - index)
+ }
+ partitionInfo.setListInfo(newListInfo.map(_.asJava).asJava)
}
- CommonUtil.validateRangeInfo(newRangeInfo, columnDataType,
- timestampFormatter, dateFormatter)
- partitionInfo.setRangeInfo(newRangeInfo.asJava)
- } else if (partitionInfo.getPartitionType == PartitionType.LIST) {
- val originList =
partitionInfo.getListInfo.asScala.map(_.asScala.toList).toList
- if (partitionId != 0) {
- val targetListInfo = partitionInfo.getListInfo.get(index - 1)
- CommonUtil.validateSplitListInfo(targetListInfo.asScala.toList,
splitInfo, originList)
+
+ if (partitionId == 0) {
+ partitionInfo.addPartition(splitInfo.size)
} else {
- CommonUtil.validateAddListInfo(splitInfo, originList)
- }
- val addListInfo = PartitionUtils.getListInfo(splitInfo.mkString(","))
- val newListInfo = partitionId match {
- case 0 => originList ++ addListInfo
- case _ => originList.take(index - 1) ++ addListInfo ++
- originList.takeRight(originList.size - index)
+ partitionInfo.splitPartition(index, splitInfo.size)
}
- partitionInfo.setListInfo(newListInfo.map(_.asJava).asJava)
}
-
- if (partitionId == 0) {
- partitionInfo.addPartition(splitInfo.size)
- } else {
- partitionInfo.splitPartition(index, splitInfo.size)
+ else {
+ throw new IllegalArgumentException("Invalid Partition Id")
--- End diff --
@jackylk please review
---
