cloud-fan commented on code in PR #58317:
URL: https://github.com/apache/spark/pull/58317#discussion_r4009045020
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SaveIntoDataSourceCommand.scala:
##########
@@ -71,7 +71,7 @@ case class SaveIntoDataSourceCommand(
try {
val logicalRelation = LogicalRelation(relation,
toAttributes(relation.schema), None,
- false, None)
+ false, None, None)
Review Comment:
Confirmed: both V1 mutation callers now recache by BaseRelation without
weakening ordinary mode-sensitive cache identity. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3916512564","thread_id":"inline:3916512564","verdict_sha256":"9e8299073fe3a3fd317d35d14ddb819ddbbaf0c580a01f233a538f0836de3d9e"}
-->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala:
##########
@@ -752,15 +757,28 @@ case class FileSourceScanExec(
lazy val inputRDD: RDD[InternalRow] = {
val options = relation.options +
(FileFormat.OPTION_RETURNING_BATCH -> supportsColumnar.toString)
- val readFile: (PartitionedFile) => Iterator[InternalRow] =
- relation.fileFormat.buildReaderWithPartitionValues(
- sparkSession = relation.sparkSession,
- dataSchema = relation.dataSchema,
- partitionSchema = relation.partitionSchema,
- requiredSchema = requiredSchema,
- filters = pushedDownFilters,
- options = options,
- hadoopConf = getHadoopConf(relation.sparkSession, relation.options))
+ val hadoopConf = getHadoopConf(relation.sparkSession, relation.options)
+ val readFile: (PartitionedFile) => Iterator[InternalRow] =
relation.fileFormat match {
+ case format: OrcFileFormat if charVarcharStandardSemantics.isDefined =>
+ format.buildReaderWithPartitionValues(
Review Comment:
Confirmed in the final design: the generic FileFormat bridge preserves
existing public virtual dispatch for external ORC subclasses. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3916512573","thread_id":"inline:3916512573","verdict_sha256":"9e8299073fe3a3fd317d35d14ddb819ddbbaf0c580a01f233a538f0836de3d9e"}
-->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala:
##########
@@ -535,11 +548,14 @@ object OrcUtils extends Logging {
dataSchema: StructType,
resultSchema: StructType,
partitionSchema: StructType,
- conf: Configuration): String = {
+ conf: Configuration,
+ charVarcharStandardSemantics: Boolean): String = {
Review Comment:
Confirmed: the parameter documentation now explains both physical ORC schema
branches. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3916512583","thread_id":"inline:3916512583","verdict_sha256":"9e8299073fe3a3fd317d35d14ddb819ddbbaf0c580a01f233a538f0836de3d9e"}
-->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala:
##########
@@ -752,15 +757,30 @@ case class FileSourceScanExec(
lazy val inputRDD: RDD[InternalRow] = {
val options = relation.options +
(FileFormat.OPTION_RETURNING_BATCH -> supportsColumnar.toString)
- val readFile: (PartitionedFile) => Iterator[InternalRow] =
- relation.fileFormat.buildReaderWithPartitionValues(
- sparkSession = relation.sparkSession,
- dataSchema = relation.dataSchema,
- partitionSchema = relation.partitionSchema,
- requiredSchema = requiredSchema,
- filters = pushedDownFilters,
- options = options,
- hadoopConf = getHadoopConf(relation.sparkSession, relation.options))
+ val hadoopConf = getHadoopConf(relation.sparkSession, relation.options)
+ val readFile: (PartitionedFile) => Iterator[InternalRow] =
relation.fileFormat match {
+ case format: OrcFileFormat
Review Comment:
Confirmed: the typed mode and generic FileFormat bridge preserve the
analyzed mode through delegating subclasses, with the requested row and vector
coverage. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3928083269","thread_id":"inline:3928083269","verdict_sha256":"9e8299073fe3a3fd317d35d14ddb819ddbbaf0c580a01f233a538f0836de3d9e"}
-->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ApplyCharTypePadding.scala:
##########
@@ -51,34 +51,71 @@ object ApplyCharTypePadding extends Rule[LogicalPlan] {
}
override def apply(plan: LogicalPlan): LogicalPlan = {
+ val standardSemantics = conf.charVarcharStandardSemantics
+
+ // Bind into case-class state, not a TreeNodeTag: tags are dropped by
canonicalization
Review Comment:
Confirmed: the comment now states the correct structural equality and
sameResult reason. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3928083280","thread_id":"inline:3928083280","verdict_sha256":"9e8299073fe3a3fd317d35d14ddb819ddbbaf0c580a01f233a538f0836de3d9e"}
-->
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -116,7 +116,10 @@ case class DataSourceV2Relation(
catalog: Option[CatalogPlugin],
identifier: Option[Identifier],
options: CaseInsensitiveStringMap,
- timeTravelSpec: Option[TimeTravelSpec] = None)
+ timeTravelSpec: Option[TimeTravelSpec] = None,
+ // Bound at analysis so sameResult / cache reuse distinguish preserve-only
vs standard
+ // CHAR/VARCHAR scans. None means the relation was not analyzed under
first-class types.
+ charVarcharScanMode: Option[CharVarcharScanMode] = None)
Review Comment:
Confirmed: catalog-less V2 writes now use the mutation-specific matcher and
retain ordinary mode-sensitive cache identity. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3960197735","thread_id":"inline:3960197735","verdict_sha256":"9e8299073fe3a3fd317d35d14ddb819ddbbaf0c580a01f233a538f0836de3d9e"}
-->
--
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]
