You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2025/05/08 13:34:56 UTC
(spark) branch master updated: [SPARK-52036][SQL] Add SQLConfHelper trait to v2.FileScan
This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 0fb544583f0c [SPARK-52036][SQL] Add SQLConfHelper trait to v2.FileScan
0fb544583f0c is described below
commit 0fb544583f0c4bfd802bec8eca1fab32cd9031f8
Author: Kent Yao <ya...@apache.org>
AuthorDate: Thu May 8 06:34:44 2025 -0700
[SPARK-52036][SQL] Add SQLConfHelper trait to v2.FileScan
### What changes were proposed in this pull request?
Add SQLConfHelper trait to v2.FileScan
### Why are the changes needed?
Make concrete classes easy to get configs and help the IDE bypass type inference failure
### Does this PR introduce _any_ user-facing change?
no
### How was this patch tested?
Passing GA
### Was this patch authored or co-authored using generative AI tooling?
NO
Closes #50826 from yaooqinn/SPARK-52036.
Authored-by: Kent Yao <ya...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../org/apache/spark/sql/v2/avro/AvroScan.scala | 2 +-
.../sql/execution/datasources/v2/FileScan.scala | 21 ++++++++++++++++-----
.../sql/execution/datasources/v2/csv/CSVScan.scala | 8 ++++----
.../execution/datasources/v2/json/JsonScan.scala | 6 +++---
.../sql/execution/datasources/v2/orc/OrcScan.scala | 6 +++---
.../datasources/v2/parquet/ParquetScan.scala | 19 +++++++++----------
.../execution/datasources/v2/text/TextScan.scala | 2 +-
7 files changed, 37 insertions(+), 27 deletions(-)
diff --git a/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroScan.scala b/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroScan.scala
index cf2cc1f0aefc..237f5060f506 100644
--- a/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroScan.scala
+++ b/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroScan.scala
@@ -54,7 +54,7 @@ case class AvroScan(
// The partition values are already truncated in `FileScan.partitions`.
// We should use `readPartitionSchema` as the partition schema here.
AvroPartitionReaderFactory(
- sparkSession.sessionState.conf,
+ conf,
broadcastedConf,
dataSchema,
readDataSchema,
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScan.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScan.scala
index bb005cd7b697..9ace0540ec24 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScan.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScan.scala
@@ -25,6 +25,7 @@ import org.apache.spark.internal.{Logging, MDC}
import org.apache.spark.internal.LogKeys.{PATH, REASON}
import org.apache.spark.internal.config.IO_WARNING_LARGEFILETHRESHOLD
import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.SQLConfHelper
import org.apache.spark.sql.catalyst.expressions.{AttributeSet, Expression, ExpressionSet}
import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
import org.apache.spark.sql.catalyst.plans.QueryPlan
@@ -33,13 +34,18 @@ import org.apache.spark.sql.connector.read.{Batch, InputPartition, Scan, Statist
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.execution.PartitionedFileUtil
import org.apache.spark.sql.execution.datasources._
+import org.apache.spark.sql.internal.{SessionState, SQLConf}
import org.apache.spark.sql.internal.connector.SupportsMetadata
import org.apache.spark.sql.sources.Filter
import org.apache.spark.sql.types.StructType
import org.apache.spark.util.Utils
trait FileScan extends Scan
- with Batch with SupportsReportStatistics with SupportsMetadata with Logging {
+ with Batch
+ with SupportsReportStatistics
+ with SupportsMetadata
+ with SQLConfHelper
+ with Logging {
/**
* Returns whether a file with `path` could be split or not.
*/
@@ -107,13 +113,18 @@ trait FileScan extends Scan
override def hashCode(): Int = getClass.hashCode()
- val maxMetadataValueLength = sparkSession.sessionState.conf.maxMetadataStringLength
+ override def conf: SQLConf = {
+ val sessionState: SessionState = sparkSession.sessionState
+ sessionState.conf
+ }
+
+ val maxMetadataValueLength = conf.maxMetadataStringLength
override def description(): String = {
val metadataStr = getMetaData().toSeq.sorted.map {
case (key, value) =>
val redactedValue =
- Utils.redact(sparkSession.sessionState.conf.stringRedactionPattern, value)
+ Utils.redact(conf.stringRedactionPattern, value)
key + ": " + StringUtils.abbreviate(redactedValue, maxMetadataValueLength)
}.mkString(", ")
s"${this.getClass.getSimpleName} $metadataStr"
@@ -182,7 +193,7 @@ trait FileScan extends Scan
override def estimateStatistics(): Statistics = {
new Statistics {
override def sizeInBytes(): OptionalLong = {
- val compressionFactor = sparkSession.sessionState.conf.fileCompressionFactor
+ val compressionFactor = conf.fileCompressionFactor
val size = (compressionFactor * fileIndex.sizeInBytes /
(dataSchema.defaultSize + fileIndex.partitionSchema.defaultSize) *
(readDataSchema.defaultSize + readPartitionSchema.defaultSize)).toLong
@@ -204,7 +215,7 @@ trait FileScan extends Scan
a.sortBy(_.hashCode()).sameElements(b.sortBy(_.hashCode()))
}
- private val isCaseSensitive = sparkSession.sessionState.conf.caseSensitiveAnalysis
+ private val isCaseSensitive = conf.caseSensitiveAnalysis
private def normalizeName(name: String): String = {
if (isCaseSensitive) {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/csv/CSVScan.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/csv/CSVScan.scala
index 48c4d4ff9518..6aee89d1abce 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/csv/CSVScan.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/csv/CSVScan.scala
@@ -46,12 +46,12 @@ case class CSVScan(
dataFilters: Seq[Expression] = Seq.empty)
extends TextBasedFileScan(sparkSession, options) {
- val columnPruning = sparkSession.sessionState.conf.csvColumnPruning
+ val columnPruning = conf.csvColumnPruning
private lazy val parsedOptions: CSVOptions = new CSVOptions(
options.asScala.toMap,
columnPruning = columnPruning,
- sparkSession.sessionState.conf.sessionLocalTimeZone,
- sparkSession.sessionState.conf.columnNameOfCorruptRecord)
+ conf.sessionLocalTimeZone,
+ conf.columnNameOfCorruptRecord)
override def isSplitable(path: Path): Boolean = {
CSVDataSource(parsedOptions).isSplitable && super.isSplitable(path)
@@ -87,7 +87,7 @@ case class CSVScan(
new SerializableConfiguration(hadoopConf))
// The partition values are already truncated in `FileScan.partitions`.
// We should use `readPartitionSchema` as the partition schema here.
- CSVPartitionReaderFactory(sparkSession.sessionState.conf, broadcastedConf,
+ CSVPartitionReaderFactory(conf, broadcastedConf,
dataSchema, readDataSchema, readPartitionSchema, parsedOptions,
actualFilters.toImmutableArraySeq)
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/json/JsonScan.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/json/JsonScan.scala
index db92f645eedb..981b5e360798 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/json/JsonScan.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/json/JsonScan.scala
@@ -49,8 +49,8 @@ case class JsonScan(
private val parsedOptions = new JSONOptionsInRead(
CaseInsensitiveMap(options.asScala.toMap),
- sparkSession.sessionState.conf.sessionLocalTimeZone,
- sparkSession.sessionState.conf.columnNameOfCorruptRecord)
+ conf.sessionLocalTimeZone,
+ conf.columnNameOfCorruptRecord)
override def isSplitable(path: Path): Boolean = {
JsonDataSource(parsedOptions).isSplitable && super.isSplitable(path)
@@ -80,7 +80,7 @@ case class JsonScan(
new SerializableConfiguration(hadoopConf))
// The partition values are already truncated in `FileScan.partitions`.
// We should use `readPartitionSchema` as the partition schema here.
- JsonPartitionReaderFactory(sparkSession.sessionState.conf, broadcastedConf,
+ JsonPartitionReaderFactory(conf, broadcastedConf,
dataSchema, readDataSchema, readPartitionSchema, parsedOptions,
pushedFilters.toImmutableArraySeq)
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcScan.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcScan.scala
index 6bf36c72e035..8d2a668d1989 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcScan.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcScan.scala
@@ -66,16 +66,16 @@ case class OrcScan(
override def createReaderFactory(): PartitionReaderFactory = {
val broadcastedConf = sparkSession.sparkContext.broadcast(
new SerializableConfiguration(hadoopConf))
- val memoryMode = if (sparkSession.sessionState.conf.offHeapColumnVectorEnabled) {
+ val memoryMode = if (conf.offHeapColumnVectorEnabled) {
MemoryMode.OFF_HEAP
} else {
MemoryMode.ON_HEAP
}
// The partition values are already truncated in `FileScan.partitions`.
// We should use `readPartitionSchema` as the partition schema here.
- OrcPartitionReaderFactory(sparkSession.sessionState.conf, broadcastedConf,
+ OrcPartitionReaderFactory(conf, broadcastedConf,
dataSchema, readDataSchema, readPartitionSchema, pushedFilters, pushedAggregate,
- new OrcOptions(options.asScala.toMap, sparkSession.sessionState.conf), memoryMode)
+ new OrcOptions(options.asScala.toMap, conf), memoryMode)
}
override def equals(obj: Any): Boolean = obj match {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScan.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScan.scala
index 6b0552551547..9f1dff525707 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScan.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScan.scala
@@ -71,40 +71,39 @@ case class ParquetScan(
readDataSchemaAsJson)
hadoopConf.set(
SQLConf.SESSION_LOCAL_TIMEZONE.key,
- sparkSession.sessionState.conf.sessionLocalTimeZone)
+ conf.sessionLocalTimeZone)
hadoopConf.setBoolean(
SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key,
- sparkSession.sessionState.conf.nestedSchemaPruningEnabled)
+ conf.nestedSchemaPruningEnabled)
hadoopConf.setBoolean(
SQLConf.CASE_SENSITIVE.key,
- sparkSession.sessionState.conf.caseSensitiveAnalysis)
+ conf.caseSensitiveAnalysis)
// Sets flags for `ParquetToSparkSchemaConverter`
hadoopConf.setBoolean(
SQLConf.PARQUET_BINARY_AS_STRING.key,
- sparkSession.sessionState.conf.isParquetBinaryAsString)
+ conf.isParquetBinaryAsString)
hadoopConf.setBoolean(
SQLConf.PARQUET_INT96_AS_TIMESTAMP.key,
- sparkSession.sessionState.conf.isParquetINT96AsTimestamp)
+ conf.isParquetINT96AsTimestamp)
hadoopConf.setBoolean(
SQLConf.PARQUET_INFER_TIMESTAMP_NTZ_ENABLED.key,
- sparkSession.sessionState.conf.parquetInferTimestampNTZEnabled)
+ conf.parquetInferTimestampNTZEnabled)
hadoopConf.setBoolean(
SQLConf.LEGACY_PARQUET_NANOS_AS_LONG.key,
- sparkSession.sessionState.conf.legacyParquetNanosAsLong)
+ conf.legacyParquetNanosAsLong)
val broadcastedConf = sparkSession.sparkContext.broadcast(
new SerializableConfiguration(hadoopConf))
- val sqlConf = sparkSession.sessionState.conf
ParquetPartitionReaderFactory(
- sqlConf,
+ conf,
broadcastedConf,
dataSchema,
readDataSchema,
readPartitionSchema,
pushedFilters,
pushedAggregate,
- new ParquetOptions(options.asCaseSensitiveMap.asScala.toMap, sqlConf))
+ new ParquetOptions(options.asCaseSensitiveMap.asScala.toMap, conf))
}
override def equals(obj: Any): Boolean = obj match {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/text/TextScan.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/text/TextScan.scala
index d19a8adb6ba2..6cf961369b60 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/text/TextScan.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/text/TextScan.scala
@@ -73,7 +73,7 @@ case class TextScan(
}
val broadcastedConf = sparkSession.sparkContext.broadcast(
new SerializableConfiguration(hadoopConf))
- TextPartitionReaderFactory(sparkSession.sessionState.conf, broadcastedConf, readDataSchema,
+ TextPartitionReaderFactory(conf, broadcastedConf, readDataSchema,
readPartitionSchema, textOptions)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org