dongjoon-hyun commented on code in PR #56928:
URL: https://github.com/apache/spark/pull/56928#discussion_r3510582444
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala:
##########
@@ -320,6 +320,33 @@ trait FileSourceScanLike extends DataSourceScanExec with
SessionStateHelper {
def requiredSchema: StructType
// Identifier for the table in the metastore.
def tableIdentifier: Option[TableIdentifier]
+ // When true, the `MarkSingleTaskExecution` optimizer rule has marked this
scan's plan shape as a
+ // candidate for single-task execution. The scan is only actually executed
in a single task when
+ // it additionally passes the file count and size thresholds (see
`useSingleTaskExecution`).
+ def markedForSingleTaskExecution: Boolean
+
+ /**
+ * Whether this file scan should run in a single task, reporting a
`SinglePartition` output
+ * partitioning so that a following shuffle can be elided. This is true when
the plan shape was
+ * marked eligible by the optimizer and the statically-selected files fall
within the configured
+ * count and size bounds. It relies on `selectedPartitions`, so it must not
be evaluated before
+ * the scan's file listing is available.
+ */
+ lazy val useSingleTaskExecution: Boolean = {
+ if (!markedForSingleTaskExecution) {
+ false
+ } else {
+ val sqlConf = getSqlConf(relation.sparkSession)
+ val minNumFiles =
sqlConf.getConf(SQLConf.SINGLE_TASK_EXECUTION_MIN_NUM_FILES)
+ val maxNumFiles =
sqlConf.getConf(SQLConf.SINGLE_TASK_EXECUTION_MAX_NUM_FILES)
+ val minNumBytes =
sqlConf.getConf(SQLConf.SINGLE_TASK_EXECUTION_MIN_NUM_BYTES)
+ val maxPartitionBytes =
sqlConf.getConf(SQLConf.FILES_MAX_PARTITION_BYTES)
+ val numFiles = selectedPartitions.totalNumberOfFiles
+ val numBytes = selectedPartitions.totalFileSize
+ numFiles >= minNumFiles && numFiles <= maxNumFiles &&
+ numBytes >= minNumBytes && numBytes <= maxPartitionBytes
Review Comment:
In `isLocalRelationEligible`, `get(SQLConf.LEAF_NODE_DEFAULT_PARALLELISM)`
is considered in a way. Do we need to consider
SQLConf.LEAF_NODE_DEFAULT_PARALLELISM here too?
--
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]
