johanl-db commented on code in PR #58632:
URL: https://github.com/apache/spark/pull/58632#discussion_r4185235390
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/AssignmentUtils.scala:
##########
@@ -57,13 +59,17 @@ object AssignmentUtils extends SQLConfHelper with
CastSupport {
* (preserving existing fields).
* @param coerceNestedTypes whether to coerce nested types to match the
target type
* for complex types
+ * @param ansiStoreAssignmentCastCheck when the ANSI store-assignment cast
check runs:
+ * ANALYSIS (default) rejects an unsafe
cast at analysis
+ * time; RUNTIME rejects invalid values
at execution time.
* @return aligned update assignments that match table attributes
*/
def alignUpdateAssignments(
attrs: Seq[Attribute],
assignments: Seq[Assignment],
fromStar: Boolean,
- coerceNestedTypes: Boolean): Seq[Assignment] = {
+ coerceNestedTypes: Boolean,
+ ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck = ANALYSIS):
Seq[Assignment] = {
Review Comment:
Fixed by:
- Adding `AnsiStoreAssignmentCastCheck.DEFAULT` (= `ANALYSIS`)
- Making the parameter required in a number of method calls instead of
always defaulting to ANALYSIS.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/DataTypeUtils.scala:
##########
@@ -213,13 +217,27 @@ object DataTypeUtils {
// If write-side data type is a user-defined type, check with its
underlying data type.
case (w, r) if w.isInstanceOf[UserDefinedType[_]] &&
!r.isInstanceOf[UserDefinedType[_]] =>
canWrite(tableName, w.asInstanceOf[UserDefinedType[_]].sqlType, r,
byName, resolver,
- context, storeAssignmentPolicy, addError)
+ context, storeAssignmentPolicy, addError,
ansiStoreAssignmentCastCheck)
// If read-side data type is a user-defined type, check with its
underlying data type.
case (w, r) if r.isInstanceOf[UserDefinedType[_]] &&
!w.isInstanceOf[UserDefinedType[_]] =>
canWrite(tableName, w, r.asInstanceOf[UserDefinedType[_]].sqlType,
byName, resolver,
- context, storeAssignmentPolicy, addError)
-
+ context, storeAssignmentPolicy, addError,
ansiStoreAssignmentCastCheck)
+
+ // RUNTIME defers to the inserted ANSI cast, which fails on malformed or
overflowing
+ // values at runtime. Clearly invalid conversion are still rejected
either in an earlier
+ // branch of this match clause, or in checkAnalysis.
+ case (w, r) if storeAssignmentPolicy == ANSI &&
ansiStoreAssignmentCastCheck == RUNTIME =>
+ (w, r) match {
+ // Long/decimal -> timestamp can silently overflow, and variant ->
complex types doesn't
+ // enforce nested field non-nullability. Keep rejecting them for now.
+ case (LongType | _: DecimalType, TimestampType) |
+ (VariantType, _: StructType | _: ArrayType | _: MapType) |
+ (_: TimeType | _: AnyTimestampNanoType, VariantType) =>
Review Comment:
Now also rejecting nested TIME -> Variant to avoid a MatchError downstream +
extended tests
##########
sql/catalyst/src/main/java/org/apache/spark/sql/internal/connector/SchemaAlignmentConfig.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.internal.connector;
+
+import org.apache.spark.sql.connector.catalog.Table;
+
+/**
+ * Schema alignment configuration for DSv2 batch/row-level writes to a {@link
Table}, exposed via
+ * {@link SupportsConfigurableSchemaAlignment}.
+ */
+public class SchemaAlignmentConfig {
+
+ /** The default data source v2 configuration. */
+ public static final SchemaAlignmentConfig DEFAULT = new
SchemaAlignmentConfig();
+
+ /**
+ * When the {@code ANSI} store-assignment cast check runs for writes to a
table, under
+ * {@code spark.sql.storeAssignmentPolicy=ANSI}. Has no effect under the
{@code STRICT} policy.
+ */
+ public enum AnsiStoreAssignmentCastCheck {
Review Comment:
Examples I have in mind:
- Allow storeAssignmentPolicy = LEGACY - an earlier version of this PR had
this
-
[coerceNestedTypes](https://github.com/apache/spark/blob/f6953f77e61359bffca4e7b11818d05cd724194b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveRowLevelCommandAssignments.scala#L46):
currently a session-level conf, but it would/could be better as a per-table
connector provided config
- How to handle missing/extra struct fields, default/generated values
Schema alignment is complex, in general we want all connectors to behave the
same, but in some cases connectors may have to deviate from that. I do expect
that need is larger than what storeAssignmentPolicy covers and
`SchemaAlignmentConfig` makes room for that
--
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]
