srielau commented on code in PR #59468:
URL: https://github.com/apache/spark/pull/59468#discussion_r4287234173
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroDataToCatalyst.scala:
##########
@@ -25,24 +25,46 @@ import org.apache.avro.io.{BinaryDecoder, DecoderFactory}
import org.apache.spark.SparkException
import org.apache.spark.sql.catalyst.expressions.{ExpectsInputTypes,
Expression, SpecificInternalRow, UnaryExpression}
import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext,
CodeGenerator, ExprCode}
-import org.apache.spark.sql.catalyst.util.{FailFastMode, ParseMode,
PermissiveMode}
+import org.apache.spark.sql.catalyst.util.{CharVarcharUtils, FailFastMode,
ParseMode, PermissiveMode}
import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
case class AvroDataToCatalyst(
child: Expression,
jsonFormatSchema: String,
- options: Map[String, String])
+ options: Map[String, String],
+ // PERSISTED: capture at analysis (like JsonToStructs) so copy()/views keep
+ // creation-time CHAR/VARCHAR map-key checks, not the caller's session
setting.
+ // No default args: TreeNode.makeCopy would otherwise re-read live SQLConf.
+ charVarcharStandardSemantics: Boolean,
+ charVarcharFirstClassTypes: Boolean)
extends UnaryExpression with ExpectsInputTypes {
+ // Tests and 3-arg apply capture the current session flags.
+ def this(child: Expression, jsonFormatSchema: String, options: Map[String,
String]) =
+ this(
+ child,
+ jsonFormatSchema,
+ options,
+ SQLConf.get.charVarcharStandardSemantics,
+ SQLConf.get.charVarcharFirstClassTypes)
+
override def inputTypes: Seq[AbstractDataType] = Seq(BinaryType)
override lazy val dataType: DataType = {
- val dt = SchemaConverters.toSqlType(
+ val converted = SchemaConverters.toSqlType(
expectedSchema,
avroOptions.useStableIdForUnionType,
avroOptions.stableIdPrefixForUnionType,
avroOptions.recursiveFieldMaxDepth).dataType
+ // SchemaConverters restores CHAR/VARCHAR stamps from Avro even when
first-class types
+ // are off. Use the analysis-time snapshot so copy() cannot drop CHAR map
keys.
+ val dt = if (charVarcharFirstClassTypes) {
+ converted
+ } else {
+ CharVarcharUtils.replaceCharVarcharWithString(converted)
Review Comment:
Fixed in 882633ae4e9: `dataType` now uses
`CharVarcharUtils.replaceCharVarcharWithStringAlways` when the persisted
snapshot is false, so a copy in a first-class session cannot revive
CHAR/VARCHAR. Added a temp-view test that creates from_avro with both flags off
and queries it after SET standardSemantics true; the column stays STRING and
the over-long value is unchanged.
##########
connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:
##########
@@ -4012,11 +4012,18 @@ abstract class AvroSuite
withTempPath { dir =>
Seq("abcdef").toDF("c").write.format("avro").save(dir.getCanonicalPath)
Seq("CHAR", "VARCHAR").foreach { typ =>
+ // Avro file reads wrap the row-level SparkRuntimeException in
SparkException
Review Comment:
Restored the original `intercept[SparkRuntimeException]` and dropped the
FAILED_READ_FILE comment in 882633ae4e9. Scan overflow for this CHAR/VARCHAR
column still comes from ApplyCharTypePadding, not a wrapped file-read error.
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroDataToCatalyst.scala:
##########
@@ -25,24 +25,46 @@ import org.apache.avro.io.{BinaryDecoder, DecoderFactory}
import org.apache.spark.SparkException
import org.apache.spark.sql.catalyst.expressions.{ExpectsInputTypes,
Expression, SpecificInternalRow, UnaryExpression}
import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext,
CodeGenerator, ExprCode}
-import org.apache.spark.sql.catalyst.util.{FailFastMode, ParseMode,
PermissiveMode}
+import org.apache.spark.sql.catalyst.util.{CharVarcharUtils, FailFastMode,
ParseMode, PermissiveMode}
import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
case class AvroDataToCatalyst(
child: Expression,
jsonFormatSchema: String,
- options: Map[String, String])
+ options: Map[String, String],
+ // PERSISTED: capture at analysis (like JsonToStructs) so copy()/views keep
+ // creation-time CHAR/VARCHAR map-key checks, not the caller's session
setting.
+ // No default args: TreeNode.makeCopy would otherwise re-read live SQLConf.
Review Comment:
Dropped that sentence in 882633ae4e9. `copy()` / `makeCopy` keep the product
fields; JsonToStructs already persists the same kind of flag with a default
argument.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/avroSqlFunctions.scala:
##########
@@ -62,12 +63,30 @@ import org.apache.spark.util.Utils
since = "4.0.0"
)
// scalastyle:on line.size.limit
-case class FromAvro(child: Expression, jsonFormatSchema: Expression, options:
Expression)
+case class FromAvro(
+ child: Expression,
+ jsonFormatSchema: Expression,
+ options: Expression,
+ // PERSISTED: copy()/views keep analysis-time CHAR/VARCHAR flags. FromAvro
is
+ // RuntimeReplaceable, so a new replacement would otherwise re-read
SQLConf.
+ // No default args: copy() must keep these, and FunctionRegistry must still
Review Comment:
Dropped the default-arg rationale in 882633ae4e9. `copy()` keeps the fields
either way; the explicit 3-arg `def this` is what FunctionRegistry needs for
the SQL form.
--
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]
