HyukjinKwon commented on code in PR #59468:
URL: https://github.com/apache/spark/pull/59468#discussion_r4281892901
##########
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:
**Nit (P3):** I don't think this rationale holds. `TreeNode.makeCopy` passes
every product element, and case-class `copy()` defaults each parameter to its
current value, so default-argument getters are never evaluated on copy.
`JsonToStructs`, cited just above, persists the same flag with a default
argument. Could we drop or reword this sentence so it doesn't suggest default
arguments are unsafe here?
See **Shared repair plan 1** in the review body.
##########
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:
**Non-blocking (P2):** `replaceCharVarcharWithString` checks the live
`SqlApiConf.get.charVarcharFirstClassTypes` again and returns the type
unchanged when the current session has standardSemantics or
preserveCharVarcharTypeInfo on. The persisted `charVarcharFirstClassTypes =
false` snapshot is therefore only honored if `dataType` happens to be evaluated
under a flag-off conf.
`dataType` is a per-instance lazy val, and `withNewChildInternal` and
`FromAvro.replacement` create fresh instances. A from_avro analyzed with both
flags off (a DataFrame, or a view body analyzed under its creation conf) can be
copied during optimization in a session with standardSemantics on, for example
when CollapseProject substitutes an aliased input. The copy then reports
`CHAR(n)` while the resolved attribute stays STRING, and its deserializer pads
values or throws `EXCEED_LIMIT_LENGTH`. The new view test only covers the
captured-true / live-false direction, which takes the other branch.
Using the conf-independent
`CharVarcharUtils.replaceCharVarcharWithStringAlways(converted)` here would
make the snapshot authoritative in both directions.
**Verification:**
- **Regression:** Create a from_avro over a CHAR-stamped schema with both
flags off (e.g. a temp view, with an input rewritten during optimization), then
query it with standardSemantics enabled. Assert the column is STRING and an
over-long or short value is returned unchanged.
- **Compatibility:** The existing captured-true / live-false persistence
case and the preserve-only and flags-off result-type cases still pass.
- **Compile:** The avro connector compiles against the private[sql]
conf-independent replacement helper.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharUtils.scala:
##########
@@ -208,6 +208,30 @@ object CharVarcharUtils extends Logging with
SparkCharVarcharUtils {
}
}
+ /**
+ * CHAR/VARCHAR map keys are never padded or trimmed. A CHAR(n) key must
already be exactly
+ * n characters; a VARCHAR(n) key must already be at most n. STRING and
other key types fit.
+ * Avro map-key checks use this; JSON/XML keep equivalent local checks for
now.
+ */
+ def charVarcharMapKeyExceedsLimit(key: UTF8String, keyType: DataType):
Boolean = {
Review Comment:
**Nit (P3):** For `CHAR(n)` this also returns true when the key is *shorter*
than `n` (`numChars != length`). The name reads like an upper-bound check, but
the method actually tests the CHAR exact-length / VARCHAR max-length rule. The
doc anticipates JSON/XML reusing this helper, so a name like
`charVarcharMapKeyViolatesLength`, or at least a doc line saying what `true`
means, would keep a future caller from treating short CHAR keys as valid.
**Verification:**
- **Compile:** The renamed helper and its single caller compile across
catalyst and sql/core.
- **Regression:** Existing short, long, and exact CHAR/VARCHAR map-key tests
keep the same accept/reject outcomes.
##########
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:
**Nit (P3):** File scans no longer assign CHAR/VARCHAR in the deserializer:
`AvroFileFormat` and `AvroPartitionReaderFactory` keep
`applyCharVarcharAssignment = false`. For this plain `CHAR(4)`/`VARCHAR(4)`
column, `EXCEED_LIMIT_LENGTH` therefore still comes from the read-side Project
check that `ApplyCharTypePadding` adds, exactly as on master, and it is not
wrapped in `FAILED_READ_FILE`. The comment describes an earlier revision. With
the cause-chain walk, the test would no longer catch a regression that starts
wrapping this error. Could we restore the original
`intercept[SparkRuntimeException]` and drop the comment?
**Verification:**
- **Regression:** The CHAR and VARCHAR overflow reads in both the V1 and V2
Avro suites fail the test if EXCEED_LIMIT_LENGTH arrives wrapped in another
exception, and pass with the current scan behavior.
##########
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:
**Nit (P3):** Same issue as on `AvroDataToCatalyst`. `copy()` keeps these
fields with or without defaults. Scala default arguments don't add JVM
constructors either, so `FunctionRegistry` would see the same Expression-only
constructors in both cases; the explicit 3-arg `def this` is what keeps the
3-arg SQL form working. Could we reword or drop this rationale?
See **Shared repair plan 1** in the review body.
--
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]
