LuciferYang commented on code in PR #12954:
URL: https://github.com/apache/gluten/pull/12954#discussion_r3950238671
##########
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:
Removed. `ScanTransformerFactory`, `IcebergScanTransformer` and
`PaimonScanTransformer` now read `batchScan.table` directly; the field is a
public case-class val on 3.4 through 4.1.
##########
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:
Both moved to
`gluten-substrait/src/main/scala/org/apache/gluten/expression/ExpressionUtils.scala`,
which already holds pure expression predicates, and the two callers
(`UnaryExpressionTransformer`, `VeloxSparkPlanExecApi`) call it there.
I argued for keeping them in the shim earlier in this thread, on the grounds
that `EvalMode` did not exist before 3.4 so the predicates had already diverged
by version once. Your point stands: they do not vary across the versions we
support now, and moving them back if `evalMode` changes again is cheap.
##########
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:
Removed the same way, with the three callers reading
`batchScan.keyGroupedPartitioning`. Worth noting for anyone reading the diff:
it is a constructor `val` on 3.4 and a `def` over `spjParams` on 3.5+, so reads
port cleanly while a `copy(keyGroupedPartitioning = ...)` would not.
`getCommonPartitionValues` stays, since 3.4 reads
`batchScan.commonPartitionValues` and 3.5+ reads it off `spjParams`.
##########
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:
Moved to a new
`gluten-substrait/src/main/scala/org/apache/gluten/utils/FileMetadataUtil.scala`,
called from `VeloxIteratorApi` and `CHIteratorApi`. The body is unchanged. I
went with a new file rather than folding it into an existing util because none
of the seven in that package is about per-file metadata: `FileIndexUtil` takes
a `FileIndex`, `PartitionsUtil` groups files into partitions.
##########
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:
Removed. `ColumnarBroadcastExchangeExec` now calls
`SparkContextUtils.broadcastInternal(sparkContext, ...)` directly, the same way
`BasicPhysicalOperatorTransformer` already calls
`SparkContextUtils.createPartitioningAwareUnionRDD`.
A correction while I am here: #12953 ยง1 said this one "cannot be lifted at
all" because `SparkContextUtils` exists once per shim module and is invisible
from `shims/common`. The first half is true, the conclusion was not, and it is
what made me leave this alone the first time. `gluten-substrait` declares
`${sparkshim.artifactId}` at compile scope, so the caller can see that class
even though `shims/common` cannot. I will fix the issue text.
##########
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:
Both removed. They shared a private `getLimit` helper that was duplicated
byte-for-byte in all four shims; since both callers are in
`OffloadSingleNodeRules`, that is now one private `limitAndOffset` there and
the four copies are gone.
--
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]
