philo-he commented on code in PR #12954:
URL: https://github.com/apache/gluten/pull/12954#discussion_r3939739304
##########
shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala:
##########
@@ -212,18 +233,32 @@ trait SparkShims {
commonPartitionValues: Option[Seq[(InternalRow, Int)]],
applyPartialClustering: Boolean,
replicatePartitions: Boolean,
- joinKeyPositions: Option[Seq[Int]] = None): Seq[Seq[InputPartition]] =
- filteredPartitions
-
- def extractExpressionTimestampAddUnit(timestampAdd: Expression):
Option[Seq[String]] =
- Option.empty
-
- def extractExpressionTimestampDiffUnit(timestampDiff: Expression):
Option[String] =
- Option.empty
-
- def withTryEvalMode(expr: Expression): Boolean = false
+ joinKeyPositions: Option[Seq[Int]] = None): Seq[Seq[InputPartition]]
+
+ def extractExpressionTimestampAddUnit(timestampAdd: Expression):
Option[Seq[String]]
+
+ def withTryEvalMode(expr: Expression): Boolean = {
+ expr match {
+ case a: Add => a.evalMode == EvalMode.TRY
+ case s: Subtract => s.evalMode == EvalMode.TRY
+ case d: Divide => d.evalMode == EvalMode.TRY
+ case m: Multiply => m.evalMode == EvalMode.TRY
+ case c: Cast => c.evalMode == EvalMode.TRY
+ case _ => false
+ }
+ }
- def withAnsiEvalMode(expr: Expression): Boolean = false
+ def withAnsiEvalMode(expr: Expression): Boolean = {
Review Comment:
It seems both `withAnsiEvalMode` and `withTryEvalMode` can be moved from
this shim class, since no divergence among the supported Spark versions with
Spark 3.3 removed. Perhaps, it would be better to move them to an existing or
new util class.
##########
shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala:
##########
@@ -124,28 +133,18 @@ trait SparkShims {
sparkPartitionId: Int,
sparkAttemptNumber: Int,
committer: FileCommitProtocol,
- iterator: Iterator[InternalRow]): WriteTaskResult = {
- throw new UnsupportedOperationException()
- }
+ iterator: Iterator[InternalRow]): WriteTaskResult
- def enableNativeWriteFilesByDefault(): Boolean = false
+ def enableNativeWriteFilesByDefault(): Boolean
- // Planned V1 writes were introduced in Spark 3.4. Older versions do not
expose a required
- // ordering utility and keep the default empty ordering.
- // TODO: Remove this shim after dropping Spark 3.3 support.
def getV1WriteRequiredOrdering(
outputColumns: Seq[Attribute],
partitionColumns: Seq[Attribute],
bucketSpec: Option[BucketSpec],
options: Map[String, String],
- numStaticPartitionCols: Int): Seq[SortOrder] = Seq.empty
+ numStaticPartitionCols: Int): Seq[SortOrder]
- def broadcastInternal[T: ClassTag](sc: SparkContext, value: T): Broadcast[T]
= {
- // Since Spark 3.4, the `sc.broadcast` has been optimized to use
`sc.broadcastInternal`.
- // More details see SPARK-39983.
- // TODO, remove this shim once we drop Spark3.3 and previous
- sc.broadcast(value)
- }
+ def broadcastInternal[T: ClassTag](sc: SparkContext, value: T): Broadcast[T]
Review Comment:
Seems we can remove this shim API also.
##########
shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala:
##########
@@ -89,33 +91,40 @@ trait SparkShims {
sparkSession: SparkSession,
readFunction: PartitionedFile => Iterator[InternalRow],
filePartitions: Seq[FilePartition],
- fileSourceScanExec: FileSourceScanExec): FileScanRDD
+ fileSourceScanExec: FileSourceScanExec): FileScanRDD = {
+ new FileScanRDD(
+ sparkSession,
+ readFunction,
+ filePartitions,
+ new StructType(
+ fileSourceScanExec.requiredSchema.fields ++
+ fileSourceScanExec.relation.partitionSchema.fields),
+ fileSourceScanExec.fileConstantMetadataColumns
+ )
+ }
def filesGroupedToBuckets(
selectedPartitions: Array[PartitionDirectory]): Map[Int,
Array[PartitionedFile]]
- // Spark3.4 new add table parameter in BatchScanExec.
- def getBatchScanExecTable(batchScan: BatchScanExec): Table
+ def getBatchScanExecTable(batchScan: BatchScanExec): Table = batchScan.table
- // The PartitionedFile API changed in spark 3.4
def generatePartitionedFile(
partitionValues: InternalRow,
filePath: String,
start: Long,
length: Long,
- @transient locations: Array[String] = Array.empty): PartitionedFile
+ @transient locations: Array[String] = Array.empty): PartitionedFile =
+ PartitionedFile(partitionValues, SparkPath.fromPathString(filePath),
start, length, locations)
Review Comment:
Ditto to directly call the implementation on the call side and remove this
shim API.
##########
shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala:
##########
@@ -184,20 +183,42 @@ trait SparkShims {
file: PartitionedFile,
metadataColumnNames: Seq[String] = Seq.empty): Map[String, String] = {
val requested = metadataColumnNames.toSet
- Seq(
+ val originMetadataColumn = Seq(
InputFileName().prettyName -> file.filePath.toString,
InputFileBlockStart().prettyName -> file.start.toString,
InputFileBlockLength().prettyName -> file.length.toString
).collect { case (name, value) if requested.contains(name) => name ->
value }.toMap
+ val metadataColumn: mutable.Map[String, String] =
mutable.Map(originMetadataColumn.toSeq: _*)
+ val path = new Path(file.filePath.toString)
+ for (columnName <- metadataColumnNames) {
+ columnName match {
+ case FileFormat.FILE_PATH => metadataColumn += (FileFormat.FILE_PATH
-> path.toString)
+ case FileFormat.FILE_NAME => metadataColumn += (FileFormat.FILE_NAME
-> path.getName)
+ case FileFormat.FILE_SIZE =>
+ metadataColumn += (FileFormat.FILE_SIZE -> file.fileSize.toString)
+ case FileFormat.FILE_MODIFICATION_TIME =>
+ val fileModifyTime = TimestampFormatter
+ .getFractionFormatter(ZoneOffset.UTC)
+ .format(file.modificationTime * 1000L)
+ metadataColumn += (FileFormat.FILE_MODIFICATION_TIME ->
fileModifyTime)
+ case FileFormat.FILE_BLOCK_START =>
+ metadataColumn += (FileFormat.FILE_BLOCK_START ->
file.start.toString)
+ case FileFormat.FILE_BLOCK_LENGTH =>
+ metadataColumn += (FileFormat.FILE_BLOCK_LENGTH ->
file.length.toString)
+ case _ =>
+ }
+ }
+ metadataColumn.toMap
}
// For compatibility with Spark-3.5.
def getAnalysisExceptionPlan(ae: AnalysisException): Option[LogicalPlan]
- def getKeyGroupedPartitioning(batchScan: BatchScanExec):
Option[Seq[Expression]] = Option(Seq())
+ def getKeyGroupedPartitioning(batchScan: BatchScanExec):
Option[Seq[Expression]] = {
+ batchScan.keyGroupedPartitioning
Review Comment:
It looks like this method can be removed, and let
`batchScan.keyGroupedPartitioning` be directly called on the caller side.
##########
shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala:
##########
Review Comment:
I posted some comments before being aware of this comment. I would like to
do the further refactor to remove those shim APIs if their implementations are
consistent across the supported Spark versions. We can move them to the caller
side or a proper module (if the shim API implementation is a bit complex and
they are called from two or more places) for meeting the dependency requirement.
##########
shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala:
##########
@@ -89,33 +91,40 @@ trait SparkShims {
sparkSession: SparkSession,
readFunction: PartitionedFile => Iterator[InternalRow],
filePartitions: Seq[FilePartition],
- fileSourceScanExec: FileSourceScanExec): FileScanRDD
+ fileSourceScanExec: FileSourceScanExec): FileScanRDD = {
+ new FileScanRDD(
+ sparkSession,
+ readFunction,
+ filePartitions,
+ new StructType(
+ fileSourceScanExec.requiredSchema.fields ++
+ fileSourceScanExec.relation.partitionSchema.fields),
+ fileSourceScanExec.fileConstantMetadataColumns
+ )
+ }
def filesGroupedToBuckets(
selectedPartitions: Array[PartitionDirectory]): Map[Int,
Array[PartitionedFile]]
- // Spark3.4 new add table parameter in BatchScanExec.
- def getBatchScanExecTable(batchScan: BatchScanExec): Table
+ def getBatchScanExecTable(batchScan: BatchScanExec): Table = batchScan.table
Review Comment:
Should this shim API be removed? Then, directly call batchScan.table on the
caller side.
##########
shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala:
##########
@@ -184,20 +183,42 @@ trait SparkShims {
file: PartitionedFile,
metadataColumnNames: Seq[String] = Seq.empty): Map[String, String] = {
val requested = metadataColumnNames.toSet
- Seq(
+ val originMetadataColumn = Seq(
InputFileName().prettyName -> file.filePath.toString,
InputFileBlockStart().prettyName -> file.start.toString,
InputFileBlockLength().prettyName -> file.length.toString
).collect { case (name, value) if requested.contains(name) => name ->
value }.toMap
+ val metadataColumn: mutable.Map[String, String] =
mutable.Map(originMetadataColumn.toSeq: _*)
+ val path = new Path(file.filePath.toString)
+ for (columnName <- metadataColumnNames) {
+ columnName match {
+ case FileFormat.FILE_PATH => metadataColumn += (FileFormat.FILE_PATH
-> path.toString)
+ case FileFormat.FILE_NAME => metadataColumn += (FileFormat.FILE_NAME
-> path.getName)
+ case FileFormat.FILE_SIZE =>
+ metadataColumn += (FileFormat.FILE_SIZE -> file.fileSize.toString)
+ case FileFormat.FILE_MODIFICATION_TIME =>
+ val fileModifyTime = TimestampFormatter
+ .getFractionFormatter(ZoneOffset.UTC)
+ .format(file.modificationTime * 1000L)
+ metadataColumn += (FileFormat.FILE_MODIFICATION_TIME ->
fileModifyTime)
+ case FileFormat.FILE_BLOCK_START =>
+ metadataColumn += (FileFormat.FILE_BLOCK_START ->
file.start.toString)
+ case FileFormat.FILE_BLOCK_LENGTH =>
+ metadataColumn += (FileFormat.FILE_BLOCK_LENGTH ->
file.length.toString)
+ case _ =>
+ }
+ }
+ metadataColumn.toMap
Review Comment:
Perhaps, it would be better to move this shim API to other existing or new
util class.
##########
shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala:
##########
@@ -89,33 +91,40 @@ trait SparkShims {
sparkSession: SparkSession,
readFunction: PartitionedFile => Iterator[InternalRow],
filePartitions: Seq[FilePartition],
- fileSourceScanExec: FileSourceScanExec): FileScanRDD
+ fileSourceScanExec: FileSourceScanExec): FileScanRDD = {
+ new FileScanRDD(
+ sparkSession,
+ readFunction,
+ filePartitions,
+ new StructType(
+ fileSourceScanExec.requiredSchema.fields ++
+ fileSourceScanExec.relation.partitionSchema.fields),
+ fileSourceScanExec.fileConstantMetadataColumns
+ )
+ }
def filesGroupedToBuckets(
selectedPartitions: Array[PartitionDirectory]): Map[Int,
Array[PartitionedFile]]
- // Spark3.4 new add table parameter in BatchScanExec.
- def getBatchScanExecTable(batchScan: BatchScanExec): Table
+ def getBatchScanExecTable(batchScan: BatchScanExec): Table = batchScan.table
- // The PartitionedFile API changed in spark 3.4
def generatePartitionedFile(
partitionValues: InternalRow,
filePath: String,
start: Long,
length: Long,
- @transient locations: Array[String] = Array.empty): PartitionedFile
+ @transient locations: Array[String] = Array.empty): PartitionedFile =
+ PartitionedFile(partitionValues, SparkPath.fromPathString(filePath),
start, length, locations)
def isWindowGroupLimitExec(plan: SparkPlan): Boolean = false
def getWindowGroupLimitExecShim(plan: SparkPlan): WindowGroupLimitExecShim =
null
def getWindowGroupLimitExec(windowGroupLimitExecShim:
WindowGroupLimitExecShim): SparkPlan = null
- def getLimitAndOffsetFromGlobalLimit(plan: GlobalLimitExec): (Int, Int) =
(plan.limit, 0)
-
- def getLimitAndOffsetFromTopK(plan: TakeOrderedAndProjectExec): (Int, Int) =
(plan.limit, 0)
+ def getLimitAndOffsetFromGlobalLimit(plan: GlobalLimitExec): (Int, Int)
- def getExtendedColumnarPostRules(): List[SparkSession => Rule[SparkPlan]]
+ def getLimitAndOffsetFromTopK(plan: TakeOrderedAndProjectExec): (Int, Int)
Review Comment:
It seems that the above two shims APIs can be removed now. And the
implementation for them are consistent for Spark 3.4 and later versions now.
Consider to use their implementations on the caller side.
##########
shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala:
##########
@@ -89,33 +91,40 @@ trait SparkShims {
sparkSession: SparkSession,
readFunction: PartitionedFile => Iterator[InternalRow],
filePartitions: Seq[FilePartition],
- fileSourceScanExec: FileSourceScanExec): FileScanRDD
+ fileSourceScanExec: FileSourceScanExec): FileScanRDD = {
+ new FileScanRDD(
+ sparkSession,
+ readFunction,
+ filePartitions,
+ new StructType(
+ fileSourceScanExec.requiredSchema.fields ++
+ fileSourceScanExec.relation.partitionSchema.fields),
+ fileSourceScanExec.fileConstantMetadataColumns
Review Comment:
Ditto to directly use the implementation on the caller side or move to a
util class or method if it is called from two or more places.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
