dongjoon-hyun commented on code in PR #58913: URL: https://github.com/apache/spark/pull/58913#discussion_r4428161276
########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala: ########## @@ -119,7 +119,9 @@ case class CheckOverflow( override def nullIntolerant: Boolean = true - override def nullable: Boolean = true + // When `nullOnOverflow` is false, an overflow throws instead of producing null, so the result is + // null only when the input is. When true, an overflow yields null regardless of the input. + override def nullable: Boolean = child.nullable || nullOnOverflow Review Comment: **This is a user-facing change** (correcting my earlier comment). The PR description says the argument of the `StaticInvoke` is always a nullable `BoundReference`, but that is not true for a Java bean property whose getter has `@javax.annotation.Nonnull`: - `JavaTypeInference` sets `EncoderField.nullable = encoder.nullable && !hasNonNull` ([JavaTypeInference.scala#L173-L178](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/api/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala#L173-L178)). - The `JavaBeanEncoder` serializer reads the property via `Invoke(KnownNotNull(input), getter, ..., propagateNull = f.nullable, returnNullable = f.nullable)`, which is non-nullable for such a field ([SerializerBuildHelper.scala#L451-L461](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/SerializerBuildHelper.scala#L451-L461)). - So `StaticInvoke(..., returnNullable = false)` is non-nullable, and with this PR `CheckOverflow` becomes non-nullable when `nullOnOverflow` (`!ansiEnabled`) is `false`. `ExpressionEncoder.schema` is derived from the serializer ([ExpressionEncoder.scala#L215-L217](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoder.scala#L215-L217)). For example, given ```java public class B { private BigDecimal amount; @Nonnull public BigDecimal getAmount() { return amount; } public void setAmount(BigDecimal amount) { this.amount = amount; } } ``` `spark.createDataset(list, Encoders.bean(B.class)).schema()` reports `amount` as follows. | | `nullable` | |---|---| | master | `true` | | This PR, ANSI on (default) | `false` | | This PR, ANSI off | `true` | The same applies to `BigInteger` getters, to `ds.map(f, Encoders.bean(B.class))` outputs, and to nested bean fields. Downstream, this changes, for example: - Parquet files: `OPTIONAL` -> `REQUIRED`. - `to_avro(struct(col("amount")))`: the writer schema changes from a `["fixed", "null"]` union to a plain `fixed`, which is not binary-compatible with consumers using the previous schema. - `otherDf.to(ds.schema)` with a nullable source column now fails with `NULLABLE_COLUMN_OR_FIELD`. - JDBC writes now generate `NOT NULL` columns in `CREATE TABLE`. Could you update the "user-facing change" section accordingly? We also need to decide whether depending on `spark.sql.ansi.enabled` here is acceptable. The following comments explain why that dependency is tricky. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala: ########## @@ -119,7 +119,9 @@ case class CheckOverflow( override def nullIntolerant: Boolean = true - override def nullable: Boolean = true + // When `nullOnOverflow` is false, an overflow throws instead of producing null, so the result is + // null only when the input is. When true, an overflow yields null regardless of the input. + override def nullable: Boolean = child.nullable || nullOnOverflow Review Comment: **`ScalaAggregator`: analysis-time vs. execution-time nullability** The serializer is built with `nullOnOverflow = !SQLConf.get.ansiEnabled` ([SerializerBuildHelper.scala#L37](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/SerializerBuildHelper.scala#L37)), so with this PR the nullability depends on the conf *at the time the encoder is built*. In `ScalaAggregator`, `dataType` comes from `outputEncoder.objSerializer.dataType`, but `outputEncoder` is a `@transient lazy val` ([udaf.scala#L512-L515](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/udaf.scala#L512-L515)). The driver captures the analysis-time conf, while executors rebuild it with the execution-time conf. For example, take an `Aggregator` whose output bean has `@Nonnull BigDecimal getBal()`, registered via `functions.udaf`: 1. A DataFrame using it is analyzed with ANSI on, so the result type is `struct<bal: decimal(38,18) NOT NULL>`. 2. The user sets `spark.sql.ansi.enabled=false` (as the overflow error message suggests) and runs the same DataFrame. 3. Executors now write `null` on overflow into a field that the plan declares `NOT NULL`. This can lead to an NPE (e.g., `dropDuplicates()`, because struct hashing trusts `nullable = false` and skips the null check) or to wrong results (e.g., after `.where("r IS NOT NULL")`, `r.bal IS NULL` is folded to `false`). Before this PR it was harmless, because `CheckOverflow` was always nullable. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala: ########## @@ -119,7 +119,9 @@ case class CheckOverflow( override def nullIntolerant: Boolean = true - override def nullable: Boolean = true + // When `nullOnOverflow` is false, an overflow throws instead of producing null, so the result is + // null only when the input is. When true, an overflow yields null regardless of the input. + override def nullable: Boolean = child.nullable || nullOnOverflow Review Comment: **`transformWithState`: the state schema now depends on the ANSI conf** With the default UnsafeRow state encoding, the value schemas are not widened to nullable ([statefulOperators.scala#L1760-L1761](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/statefulOperators.scala#L1760-L1761), [TransformWithStateExec.scala#L155-L159](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/transformwithstate/TransformWithStateExec.scala#L155-L159)). So a `ValueState`, `ListState`, or `MapState` declared with `Encoders.bean(Acct.class)`, where `Acct` has a `@Nonnull BigDecimal` getter, stores that field as `NOT NULL` when the query starts with ANSI on. `spark.sql.ansi.enabled` is not pinned in the offset log. If the query is restarted with ANSI off, or rolled back to a build without this PR, the new schema is nullable, and `StateSchemaCompatibilityChecker` rejects it with `STATE_STORE_VALUE_SCHEMA_NOT_COMPATIBLE` ([StateSchemaCompatibilityChecker.scala#L179-L180](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateSchemaCompatibilityChecker.scala#L179-L180)). `MapState` user keys hit the key schema check. The query then cannot be restarted. This could not happen before, because the field was always nullable. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala: ########## @@ -119,7 +119,9 @@ case class CheckOverflow( override def nullIntolerant: Boolean = true - override def nullable: Boolean = true + // When `nullOnOverflow` is false, an overflow throws instead of producing null, so the result is + // null only when the input is. When true, an overflow yields null regardless of the input. + override def nullable: Boolean = child.nullable || nullOnOverflow Review Comment: **ANSI overflow errors can be optimized away** Once these serializer outputs are non-nullable, `NullPropagation` folds `IsNull`/`IsNotNull` and `count(col)` on them ([expressions.scala#L1010-L1015](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala#L1010-L1015)). `ObjectSerializerPruning` can then drop the unused serializer field, including its `CheckOverflow` ([objects.scala#L222](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/objects.scala#L222)). For example, take a bean whose `@Nonnull getAmount()` returns `new BigDecimal("1e30")`, which overflows `decimal(38,18)`. With ANSI on, `spark.range(1).map(f, Encoders.bean(Bean.class)).filter(col("amount").isNotNull()).count()` throws `NUMERIC_VALUE_OUT_OF_RANGE` on master. With this PR, it returns `1` without an error. Similarly, `select(col("amount").isNull())` returns `false`. ANSI decimal arithmetic has a precedent for this, but it is still a behavior change that the description should mention. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala: ########## @@ -119,7 +119,9 @@ case class CheckOverflow( override def nullIntolerant: Boolean = true - override def nullable: Boolean = true + // When `nullOnOverflow` is false, an overflow throws instead of producing null, so the result is + // null only when the input is. When true, an overflow yields null regardless of the input. + override def nullable: Boolean = child.nullable || nullOnOverflow Review Comment: **The null-input error of a UDAF changes as well** `ResolveEncodersInScalaAgg` binds the input deserializer to the encoder's *own* schema (`resolveAndBind()` with the default attributes), not to the actual children ([udaf.scala#L580-L590](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/udaf.scala#L580-L590)). With ANSI on, a `@Nonnull` decimal bean field becomes a non-nullable `BoundReference`, so the generated code skips the null check. For example, apply `functions.udaf(agg, Encoders.bean(Txn.class))` to a nullable column that has some `NULL` rows: - On master, it fails with `NOT_NULL_ASSERT_VIOLATION` from the bean deserializer's `AssertNotNull` ([DeserializerBuildHelper.scala#L477-L483](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/DeserializerBuildHelper.scala#L477-L483)). - With this PR, the codegen path calls `toJavaBigDecimal` on `null` and fails with an NPE wrapped in `EXPRESSION_DECODING_FAILED`. - The interpreted path still throws `NOT_NULL_ASSERT_VIOLATION`, so codegen and interpreted execution also disagree. The same pattern applies to other places that `resolveAndBind()` an encoder against its own schema, e.g., the `flatMapGroupsWithState` state deserializer. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala: ########## @@ -119,7 +119,9 @@ case class CheckOverflow( override def nullIntolerant: Boolean = true - override def nullable: Boolean = true + // When `nullOnOverflow` is false, an overflow throws instead of producing null, so the result is + // null only when the input is. When true, an overflow yields null regardless of the input. + override def nullable: Boolean = child.nullable || nullOnOverflow Review Comment: **`TransformingEncoder` over a decimal (third-party encoders only)** `TransformingEncoder.nullable` defaults to `false` ([AgnosticEncoder.scala#L305-L309](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/api/src/main/scala/org/apache/spark/sql/catalyst/encoders/AgnosticEncoder.scala#L305-L309)). So the top-level input and the `encode` `Invoke` are non-nullable ([SerializerBuildHelper.scala#L469-L478](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/SerializerBuildHelper.scala#L469-L478)), and with ANSI on `CheckOverflow` becomes non-nullable too. As a result, `ExpressionEncoder.schema` reports `NOT NULL`, while `TransformingEncoder.schema` (= `transformed.schema`) stays nullable. For example, `spark.createDataset(seq)(enc).schema` and `spark.emptyDataset(enc).schema` disagree. It also affects the "`nullable` is false iff the type is primitive" heuristic in `ScalaUDF.inputPrimitives` ([ScalaUDF.scala#L94-L95](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ScalaUDF.scala#L94-L95)). Such an input encoder would be treated as primitive, so `HandleNullInputsForUDF` would skip calling the function for null inputs. No in-tree code builds such an encoder, so this is minor, but it is another reachable non-nullable child. ########## sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/DecimalExpressionSuite.scala: ########## @@ -78,6 +78,16 @@ class DecimalExpressionSuite extends SparkFunSuite with ExpressionEvalHelper { Literal.create(null, DecimalType(2, 1)), DecimalType(3, 2), false), null) } + test("SPARK-59643: CheckOverflow nullability") { Review Comment: **Could you add an encoder-level test?** This expression-level test (the one I suggested earlier) only uses `Literal` children, so nothing pins down the user-facing behavior discussed above. Neither `JavaDatasetSuite.testNonNullField` (only `@Nonnull` bean and map fields) nor `ExpressionEncoderSuite`'s "nullable of encoder schema" (only `Int`, `Option`, `Integer`, and `String`) covers a non-null decimal input. Whatever behavior we settle on, could you add a test with a bean that has `@Nonnull BigDecimal` and `BigInteger` getters, checking the encoder schema nullability with ANSI on and off? nit (on my own snippet): `Literal(Decimal("10.1"))` is the same value as `d1` in the "CheckOverflow" test above, and the last `checkEvaluation` is the `nullOnOverflow = false` counterpart of line 59. This could reuse that fixture. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala: ########## @@ -132,11 +134,13 @@ case class CheckOverflow( override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = { val errorContextCode = getContextOrNullCode(ctx, !nullOnOverflow) nullSafeCodeGen(ctx, ev, eval => { + // Only assign isNull when nullable; nullSafeCodeGen makes it a literal otherwise. + val setIsNull = if (nullable) s"${ev.isNull} = ${ev.value} == null;" else "" Review Comment: **The non-nullable branch is already the hot path for `RowEncoder`** The description says the fix "only matters if a construction with a non-nullable child is added later". My earlier comment mentioned only the non-null-field path of `RowEncoder`. In fact, the decimal input there is `ValidateExternalType(GetExternalRowField(...))`, which is non-nullable for *every* field, nullable or not ([SerializerBuildHelper.scala#L427-L449](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/SerializerBuildHelper.scala#L427-L449), [objects.scala#L2018-L2023](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/objects/objects.scala#L2018-L2023)). So with ANSI on (the default), every `Row` decimal field (e.g., `createDataFrame(rows, schema)`, `Dataset[Row].map`) already takes this `else ""` branch. For non-nullable fields, `NullPropagation` now also removes the `AssertNotNull` around `CheckOverflow` ([expressions.scala#L1020](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala#L1020)), which changes optimized plans and the `EXPLAIN` output. Results and errors stay the same, because `GetExternalRowField` throws first, but the description should reflect this. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala: ########## @@ -119,7 +119,9 @@ case class CheckOverflow( override def nullIntolerant: Boolean = true - override def nullable: Boolean = true + // When `nullOnOverflow` is false, an overflow throws instead of producing null, so the result is + // null only when the input is. When true, an overflow yields null regardless of the input. Review Comment: **nit / follow-up: `CheckOverflowInSum`** The same reasoning applies even more strongly to `CheckOverflowInSum` in this file. With `nullOnOverflow = false` it never returns `null`, because a `null` input is treated as an overflow and throws ([DecimalExpressionUtils.java#L47-L51](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/DecimalExpressionUtils.java#L47-L51)). Yet it still declares `nullable = true` ([decimalExpressions.scala#L166](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala#L166)), while `DecimalDivideWithOverflowCheck` already uses `nullable = nullOnOverflow` ([decimalExpressions.scala#L275](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala#L275)). There is no functional impact, since `Sum` wraps it in `If(isEmpty, null, ...)`, so this could be a follow-up if we want consistency. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala: ########## @@ -119,7 +119,9 @@ case class CheckOverflow( override def nullIntolerant: Boolean = true - override def nullable: Boolean = true + // When `nullOnOverflow` is false, an overflow throws instead of producing null, so the result is + // null only when the input is. When true, an overflow yields null regardless of the input. + override def nullable: Boolean = child.nullable || nullOnOverflow Review Comment: **Root cause and a follow-up idea** The underlying issue is that `ExpressionEncoder.schema` is derived from the serializer, while the declared contract is `EncoderField.nullable` (i.e., `AgnosticEncoder.schema`). This PR aligns only decimal leaves, and only with ANSI on. For example, a `@Nonnull Integer` getter in the same bean is still reported as nullable. So `createDataset(...).schema`, `Encoders.bean(...).schema()`, `createDataFrame(list, B.class)`, and Spark Connect can disagree, and the answer also depends on the ANSI mode. Wrapping `!f.nullable` fields with `AssertNotNull` in the `JavaBeanEncoder` branch, as the `RowEncoder` branch does ([SerializerBuildHelper.scala#L436-L446](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/SerializerBuildHelper.scala#L436-L446)), would make this consistent regardless of ANSI. However, it turns a non-ANSI overflow into `NOT_NULL_ASSERT_VIOLATION`, so it deserves a separate JIRA. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala: ########## @@ -119,7 +119,9 @@ case class CheckOverflow( override def nullIntolerant: Boolean = true - override def nullable: Boolean = true + // When `nullOnOverflow` is false, an overflow throws instead of producing null, so the result is + // null only when the input is. When true, an overflow yields null regardless of the input. + override def nullable: Boolean = child.nullable || nullOnOverflow Review Comment: nit: `nullOnOverflow || child.nullable` would avoid touching `child` when `nullOnOverflow` is `true`. Currently, calling `nullable` on a `CheckOverflow` over an unresolved child throws `UnresolvedException`, even though the answer is known to be `true`. No in-tree code hits this, and `MakeDecimal` uses the same order, so this is optional. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala: ########## @@ -132,11 +134,13 @@ case class CheckOverflow( override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = { val errorContextCode = getContextOrNullCode(ctx, !nullOnOverflow) nullSafeCodeGen(ctx, ev, eval => { + // Only assign isNull when nullable; nullSafeCodeGen makes it a literal otherwise. + val setIsNull = if (nullable) s"${ev.isNull} = ${ev.value} == null;" else "" Review Comment: nit: when the child is nullable and `nullOnOverflow = false` (the most common case under ANSI, e.g., a case class `BigDecimal` field), `nullable` is `true`, so this still emits `isNull = value == null;`. That is always `false`, because `toPrecision(..., nullOnOverflow = false, ...)` never returns `null`. `if (nullOnOverflow)` would be more precise, since it implies `nullable`, at the cost of differing from `MakeDecimal`. This is not a regression, so feel free to ignore it. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala: ########## @@ -132,11 +134,13 @@ case class CheckOverflow( override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = { val errorContextCode = getContextOrNullCode(ctx, !nullOnOverflow) nullSafeCodeGen(ctx, ev, eval => { + // Only assign isNull when nullable; nullSafeCodeGen makes it a literal otherwise. Review Comment: nit: "makes it a literal" is not quite why the guard is needed. The closure still receives the `isNull_N` variable name. The non-nullable branch of `nullSafeCodeGen` just never declares that variable (only the returned `ExprCode` gets `FalseLiteral`), so the assignment would not compile. Maybe something like this: ```suggestion // `nullSafeCodeGen` declares `ev.isNull` only when this expression is nullable. ``` `VariantFromEntries` has a similar comment ([variantExpressions.scala#L337-L341](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/variant/variantExpressions.scala#L337-L341)). ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala: ########## @@ -132,11 +134,13 @@ case class CheckOverflow( override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = { val errorContextCode = getContextOrNullCode(ctx, !nullOnOverflow) nullSafeCodeGen(ctx, ev, eval => { + // Only assign isNull when nullable; nullSafeCodeGen makes it a literal otherwise. + val setIsNull = if (nullable) s"${ev.isNull} = ${ev.value} == null;" else "" Review Comment: nit / follow-up: this "assign `isNull` only when nullable" idiom now exists in `MakeDecimal`, in `VariantFromEntries` (`markNull`), and here. `VariantManipulationExpressionUtils.assignVariantResult` ([variantExpressions.scala#L465-L474](https://github.com/apache/spark/blob/2487c98bbd8de1dc5e23c8df66203af0dc6da763/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/variant/variantExpressions.scala#L465-L474)) already implements the same thing. It does not belong in this PR, but it might be worth moving into a shared helper (e.g., near `nullSafeCodeGen`) as a follow-up. Then all copies stay in sync if the declaration rule of `nullSafeCodeGen` changes. -- 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]
