chore: fallback to Spark if legacy sql configurations are set#4799
chore: fallback to Spark if legacy sql configurations are set#4799comphead wants to merge 23 commits into
Conversation
| `spark.sql.legacy.timeParserPolicy`). Comet implements current Spark semantics and does **not** | ||
| reproduce these legacy behaviors. | ||
|
|
||
| By default, when Comet detects that any `spark.sql.legacy.*` config is set to `true`, it disables |
There was a problem hiding this comment.
can't we just update the serde for specific expressions to return Unsupported rather than fall back for all queries, even queries that would not be affected by a legacy config?
There was a problem hiding this comment.
we should just fall back to the codegen dispatch approach in most cases
There was a problem hiding this comment.
Let me investigate this, sometimes the conf param is vaguely explained like
val LEGACY_JAVA_CHARSETS = buildConf("spark.sql.legacy.javaCharsets")
.internal()
.doc("When set to true, the functions like `encode()` can use charsets from JDK while " +
"encoding or decoding string values. If it is false, such functions support only one of " +
"the charsets: 'US-ASCII', 'ISO-8859-1', 'UTF-8', 'UTF-16BE', 'UTF-16LE', 'UTF-16', " +
"'UTF-32'.")
.version("4.0.0")
.booleanConf
.createWithDefault(false)
it refers to encode() explicitly but also to something else. Maybe I can fish out exact functions from Spark code depending on legacy params and in this case we can fallback to codegen dispatch
There was a problem hiding this comment.
Also, Spark has legacy configs that default to true, so this will make all jobs fall back by default.
val VIEW_SCHEMA_BINDING_ENABLED = buildConf("spark.sql.legacy.viewSchemaBindingMode")
.internal()
.doc("Set to false to disable the WITH SCHEMA clause for view DDL and suppress the line in " +
"DESCRIBE EXTENDED and SHOW CREATE TABLE.")
.version("4.0.0")
.booleanConf
.createWithDefault(true)
comphead
force-pushed
the
chore
branch
4 times, most recently
from
f5d2abc to
7125db0
Compare
| - test("cast to variant/to_variant_object with scan input") { | ||
| + test("cast to variant/to_variant_object with scan input", | ||
| + IgnoreComet("\"VariantType\" not supported")) { |
There was a problem hiding this comment.
This seems like a regression. This test was previously passing because we fell back to Spark and now we fail the query?
| // without duplicating the legacy branch natively. | ||
| private val legacyNegativeIndexConfig = "spark.sql.legacy.negativeIndexInArrayInsert" | ||
|
|
||
| private val legacyNegativeIndexReason = | ||
| s"`$legacyNegativeIndexConfig=true` legacy negative-index semantics are not implemented" + | ||
| " natively" | ||
|
|
||
| override def getIncompatibleReasons(): Seq[String] = Seq(legacyNegativeIndexReason) | ||
|
|
||
| override def getSupportLevel(expr: ArrayInsert): SupportLevel = { | ||
| if (SQLConf.get.getConfString(legacyNegativeIndexConfig, "false").toBoolean) { | ||
| Incompatible(Some(legacyNegativeIndexReason)) | ||
| } else { | ||
| Compatible() | ||
| } | ||
| } |
There was a problem hiding this comment.
This change looks good and could really be a standalone PR. This also looks like a bug fix for an issue where we were lacking tests?
| } | ||
|
|
||
| object CometCount extends CometAggregateExpressionSerde[Count] { | ||
|
|
There was a problem hiding this comment.
could you revert this whitespace change
| @@ -119,6 +128,8 @@ object CometEqualTo extends CometExpressionSerde[EqualTo] { | |||
| } | |||
|
|
|||
| object CometEqualNullSafe extends CometExpressionSerde[EqualNullSafe] { | |||
| override def getSupportLevel(expr: EqualNullSafe): SupportLevel = | |||
| ComparisonUtils.collationSupportLevel(expr.left, expr.right) | |||
| override def convert( | |||
| expr: EqualNullSafe, | |||
| inputs: Seq[Attribute], | |||
There was a problem hiding this comment.
These changes look good and probably should be separate PR since this is unrelated to legacy configs?
| "spark.sql.legacy.parquet.datetimeRebaseModeInRead" -> "CORRECTED", | ||
| "spark.sql.legacy.parquet.datetimeRebaseModeInWrite" -> "CORRECTED", | ||
| "spark.sql.legacy.parquet.int96RebaseModeInRead" -> "CORRECTED", | ||
| "spark.sql.legacy.parquet.int96RebaseModeInWrite" -> "CORRECTED", | ||
| "spark.sql.legacy.parquet.nanosAsLong" -> "false", |
There was a problem hiding this comment.
Can we not handle the parquet ones in scan planning?
2 participants
Which issue does this PR close?
Closes #4786.
Summary
Splits Spark's
spark.sql.legacy.*handling in Comet into two tiers, and hardens a fewSpark 4 features (VariantType, string collation,
lpad/rpadBinaryType) that werepreviously slipping past the serdes.
1. Session-wide fallback for legacy configs Comet cannot honor per-expression. Introduces
spark.comet.legacyConfFallback.enabled(defaulttrue). When any legacy config in acurated list (parquet datetime/int96 rebase modes,
charVarcharAsString,timeParserPolicy,disableMapKeyNormalization,viewSchemaCompensation,scalarSubqueryCountBugBehavior,setopsPrecedence, etc. — seeShimLegacyConfFallback) is set to a non-default value,isCometLoadedreturnsfalsefor the session with a WARN naming the offending keys. Thedefaults are derived from live
SQLConfConfigEntryreferences on Spark 4 and hardcoded tomatching values on Spark 3.
2. Per-expression codegen-dispatch fallback for legacy configs bound to a supported
expression. These no longer disable Comet or fall back the whole operator; they route the
expression through the JVM codegen dispatcher (Spark's own
doGenCodeinside the Cometkernel) via the
CodegenDispatchFallbackmixin:Cast—spark.sql.legacy.castComplexTypesToString.enabledarray_insert—spark.sql.legacy.negativeIndexInArrayInsertIN/InSet—spark.sql.legacy.nullInEmptyListBehaviorstr_to_map—spark.sql.legacy.truncateForEmptyRegexSplitto_csv—spark.sql.legacy.nullValueWrittenAsQuotedEmptyStringCsv3. Spark 4 hardening.
isVariantTypeshim;CometCastrejects any cast whose source or target isVariantType(including the folded-literal path), before serialization can fail.EqualTo,EqualNullSafe,GreaterThan(OrEqual),LessThan(OrEqual),In,InSet, andNotnow consultComparisonUtils.collationSupportLeveland mark non-UTF8_BINARY collatedoperands
Unsupported(native compare is byte-wise;'a' = 'A'underUNICODE_CIwould bewrong).
hasNonDefaultStringCollationwalks nested types so collated strings insidearray/map/struct operands are caught too.
lpad/rpadrejectBinaryTypestrargument (surfaced byspark.sql.legacy.lpadRpadAlwaysReturnString=true).4. Tests & diffs.
array_insert_legacy_dispatch.sqlcovering the dispatch path (noallowIncompatible).cast_complex_types_to_string*.sql: flipped the expectedexpect_fallback(...)markers toplain
querynow that these run via the dispatcher.isCometLoadedtest covering both the boolean and enum default cases, the opt-out viaspark.comet.legacyConfFallback.enabled=false, and confirming per-expression legacy configsdo NOT trigger the session-wide fallback.
dev/diffs/4.0.2.diffand4.1.2.diff: addIgnoreComettoVariantEndToEndSuitecasttest.