jackylk commented on a change in pull request #4148: URL: https://github.com/apache/carbondata/pull/4148#discussion_r670181749
########## File path: integration/spark/src/main/scala/org/apache/spark/sql/execution/command/mutation/merge/CarbonMergeDataSetUtil.scala ########## @@ -0,0 +1,206 @@ +/* + * 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.sql.execution.command.mutation.merge + +import java.util + +import scala.collection.JavaConverters._ +import scala.collection.mutable + +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.optimizer.CarbonFilters + +import org.apache.carbondata.core.indexstore.PartitionSpec +import org.apache.carbondata.core.metadata.datatype.DataTypes +import org.apache.carbondata.core.metadata.schema.table.CarbonTable +import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn +import org.apache.carbondata.core.mutate.FilePathMinMaxVO +import org.apache.carbondata.core.util.{ByteUtil, CarbonUtil, DataTypeUtil} +import org.apache.carbondata.core.util.comparator.SerializableComparator + +/** + * The utility class for Merge operations + */ +object CarbonMergeDataSetUtil { + + /** + * This method reads the splits and make (blockPath, (min, max)) tuple to to min max pruning of + * the src dataset + * + * @param colTosplitsFilePathAndMinMaxMap CarbonInputSplit whose min max cached in driver or + * the index server + * @param fileMinMaxMapListOfAllJoinColumns collection to hold the filepath and min max of all the + * join columns involved + */ + def addFilePathAndMinMaxTuples( + colTosplitsFilePathAndMinMaxMap: mutable.Map[String, util.List[FilePathMinMaxVO]], + carbonTable: CarbonTable, + joinColumnsToComparatorMap: mutable.LinkedHashMap[CarbonColumn, SerializableComparator], + fileMinMaxMapListOfAllJoinColumns: mutable.ArrayBuffer[(mutable.Map[String, (AnyRef, AnyRef)], + CarbonColumn)]): Unit = { + joinColumnsToComparatorMap.foreach { case (joinColumn, comparator) => + val fileMinMaxMap: mutable.Map[String, (AnyRef, AnyRef)] = + collection.mutable.Map.empty[String, (AnyRef, AnyRef)] + val joinDataType = joinColumn.getDataType + val isDimension = joinColumn.isDimension + val isPrimitiveAndNotDate = DataTypeUtil.isPrimitiveColumn(joinDataType) && + (joinDataType != DataTypes.DATE) + colTosplitsFilePathAndMinMaxMap(joinColumn.getColName).asScala.foreach { + filePathMinMiax => + val filePath = filePathMinMiax.getFilePath + val minBytes = filePathMinMiax.getMin + val maxBytes = filePathMinMiax.getMax + val uniqBlockPath = if (carbonTable.isHivePartitionTable) { + // While data loading to SI created on Partition table, on + // partition directory, '/' will be + // replaced with '#', to support multi level partitioning. For example, BlockId will be + // look like `part1=1#part2=2/xxxxxxxxx`. During query also, blockId should be + // replaced by '#' in place of '/', to match and prune data on SI table. + CarbonUtil.getBlockId(carbonTable.getAbsoluteTableIdentifier, + filePath, + "", + true, + false, + true) + } else { + filePath.substring(filePath.lastIndexOf("/Part") + 1) + } + if (isDimension) { + if (isPrimitiveAndNotDate) { + val minValue = DataTypeUtil.getDataBasedOnDataTypeForNoDictionaryColumn(minBytes, + joinDataType) + val maxValue = DataTypeUtil.getDataBasedOnDataTypeForNoDictionaryColumn(maxBytes, + joinDataType) + // check here if present in map, if it is, compare and update min and amx + if (fileMinMaxMap.contains(uniqBlockPath)) { + val isMinLessThanMin = + comparator.compare(fileMinMaxMap(uniqBlockPath)._1, minValue) > 0 + val isMaxMoreThanMax = + comparator.compare(maxValue, fileMinMaxMap(uniqBlockPath)._2) > 0 + updateMapIfRequiredBasedOnMinMax(fileMinMaxMap, + minValue, + maxValue, + uniqBlockPath, + isMinLessThanMin, + isMaxMoreThanMax) + } else { + fileMinMaxMap += (uniqBlockPath -> (minValue, maxValue)) + } + } else { + if (fileMinMaxMap.contains(uniqBlockPath)) { + val isMinLessThanMin = ByteUtil.UnsafeComparer.INSTANCE + .compareTo(fileMinMaxMap(uniqBlockPath)._1 + .asInstanceOf[String].getBytes(), minBytes) > 0 + val isMaxMoreThanMax = ByteUtil.UnsafeComparer.INSTANCE + .compareTo(maxBytes, fileMinMaxMap(uniqBlockPath)._2 + .asInstanceOf[String].getBytes()) > 0 + updateMapIfRequiredBasedOnMinMax(fileMinMaxMap, + new String(minBytes), + new String(maxBytes), + uniqBlockPath, + isMinLessThanMin, + isMaxMoreThanMax) + } else { + fileMinMaxMap += (uniqBlockPath -> (new String(minBytes), new String(maxBytes))) + } + } + } else { + val maxValue = DataTypeUtil.getMeasureObjectFromDataType(maxBytes, joinDataType) + val minValue = DataTypeUtil.getMeasureObjectFromDataType(minBytes, joinDataType) + if (fileMinMaxMap.contains(uniqBlockPath)) { + val isMinLessThanMin = + comparator.compare(fileMinMaxMap(uniqBlockPath)._1, minValue) > 0 + val isMaxMoreThanMin = + comparator.compare(maxValue, fileMinMaxMap(uniqBlockPath)._2) > 0 + updateMapIfRequiredBasedOnMinMax(fileMinMaxMap, + minValue, + maxValue, + uniqBlockPath, + isMinLessThanMin, + isMaxMoreThanMin) + } else { + fileMinMaxMap += (uniqBlockPath -> (minValue, maxValue)) + } + } + } + fileMinMaxMapListOfAllJoinColumns += ((fileMinMaxMap, joinColumn)) + } + } + + /** + * This method updates the min max map of the block if the value is less than min or more + * than max + */ + private def updateMapIfRequiredBasedOnMinMax(fileMinMaxMap: mutable.Map[String, (AnyRef, AnyRef)], + minValue: AnyRef, + maxValue: AnyRef, + uniqBlockPath: String, + isMinLessThanMin: Boolean, + isMaxMoreThanMin: Boolean): Unit = { + (isMinLessThanMin, isMaxMoreThanMin) match { + case (true, true) => fileMinMaxMap(uniqBlockPath) = (minValue, maxValue) + case (true, false) => fileMinMaxMap(uniqBlockPath) = (minValue, + fileMinMaxMap(uniqBlockPath)._2) + case (false, true) => fileMinMaxMap(uniqBlockPath) = (fileMinMaxMap(uniqBlockPath)._1, + maxValue) + case _ => + } + } + + /** + * This method returns the partitions required to scan in the target table based on the + * partitions present in the src table or dataset + */ + def getPartitionSpecToConsiderForPruning(sparkSession: SparkSession, Review comment: move param to next line, please follow the coding convension -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
