Skip to content

fix(spark)!: preserve decimal aggregate types and arithmetic #1448

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
alexandrefimov wants to merge 6 commits into
base: main
Choose a base branch
from
Open

fix(spark)!: preserve decimal aggregate types and arithmetic #1448

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
de05bc8
fix(spark)!: preserve decimal aggregate semantics in exported plans
alexandrefimov Nov 6, 2026
bb750a4
fix(spark): retain decimal AVG arithmetic across conversion
alexandrefimov Nov 6, 2026
7a7dd53
fix(spark): preserve decimal SUM overflow and clarify buffer behavior
alexandrefimov Nov 6, 2026
7b2f0e9
fix(spark): validate aggregate overflow preferences case insensitively
alexandrefimov Nov 6, 2026
94e05d6
fix(spark): retain optimized SUM arithmetic and decimal conversion
alexandrefimov Nov 6, 2026
ac901cd
test(spark): distinguish aggregate and window declarations
alexandrefimov Nov 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions readme.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@ Substrait Java is a project that makes it easier to build [Substrait](https://su
3) **Spark** is the module that provides an API for translating a Substrait plan to and from a Spark query plan. The most commonly used logical relations and functions are supported, including those generated from all of the TPC-H and TPC-DS queries.
The supported features are formally specified in the Substrait dialect file [spark_dialect.yaml](spark/spark_dialect.yaml).

Spark exports decimal and i64 `SUM`/`AVG` through `extension:substrait:spark`, including window calls. These functions carry Spark's result types, decimal buffer behavior, rounding and overflow modes. Decimal overflow outcomes can still depend on physical buffer materialization, as described in the function definitions. Consumers need the [Spark extension definitions](spark/src/main/resources/spark.yml) and implementations of those semantics. The functions support `INITIAL_TO_RESULT` only; exporting or importing staged calls is rejected. Optimized decimal `AVG` retains its unscaled i64 average, fp64 division and final decimal cast. Decimal optimizations can also reference the Spark `unscaled_value`, `make_decimal` and `try_make_decimal` scalar functions. Optimized SUM retains its i64 accumulator and the final unscaled decimal conversion.

## Building
Building requires a **JDK 17** installation, and Gradle itself must run on JDK 17 (see [Gradle & JDK 17](CONTRIBUTING.md#gradle--jdk-17) in the contributing guide for details).

Expand Down
172 changes: 156 additions & 16 deletions spark/spark_dialect.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -587,6 +587,27 @@ supported_scalar_functions:
notation: "FUNCTION"
supported_impls:
- "date_i32"
- source: "spark"
name: "make_decimal"
system_metadata:
name: "make_decimal"
notation: "FUNCTION"
supported_impls:
- "i64_type"
- source: "spark"
name: "try_make_decimal"
system_metadata:
name: "try_make_decimal"
notation: "FUNCTION"
supported_impls:
- "i64_type"
- source: "spark"
name: "unscaled_value"
system_metadata:
name: "unscaledvalue"
notation: "FUNCTION"
supported_impls:
- "dec"
- source: "string"
name: "contains"
system_metadata:
Expand Down Expand Up @@ -708,7 +729,6 @@ supported_aggregate_functions:
- "fp64"
- "i16"
- "i32"
- "i64"
- "i8"
- source: "arithmetic"
name: "max"
Expand Down Expand Up @@ -751,15 +771,7 @@ supported_aggregate_functions:
- "fp64"
- "i16"
- "i32"
- "i64"
- "i8"
- source: "arithmetic_decimal"
name: "avg"
system_metadata:
name: "avg"
notation: "FUNCTION"
supported_impls:
- "dec"
- source: "arithmetic_decimal"
name: "max"
system_metadata:
Expand All @@ -774,13 +786,6 @@ supported_aggregate_functions:
notation: "FUNCTION"
supported_impls:
- "dec"
- source: "arithmetic_decimal"
name: "sum"
system_metadata:
name: "sum"
notation: "FUNCTION"
supported_impls:
- "dec"
- source: "datetime"
name: "max"
system_metadata:
Expand All @@ -803,7 +808,48 @@ supported_aggregate_functions:
- "iyear"
- "pts"
- "ptstz"
- source: "spark"
name: "avg"
system_metadata:
name: "avg"
notation: "FUNCTION"
supported_impls:
- "dec"
- "i64"
- source: "spark"
name: "sum"
system_metadata:
name: "sum"
notation: "FUNCTION"
supported_impls:
- "dec"
- "i64"
supported_window_functions:
- source: "aggregate_approx"
name: "approx_count_distinct"
system_metadata:
name: "approx_count_distinct"
notation: "FUNCTION"
supported_impls:
- "any"
- source: "aggregate_generic"
name: "any_value"
system_metadata:
name: "first"
notation: "FUNCTION"
supported_impls:
- "any"
- source: "arithmetic"
name: "avg"
system_metadata:
name: "avg"
notation: "FUNCTION"
supported_impls:
- "fp32"
- "fp64"
- "i16"
- "i32"
- "i8"
- source: "arithmetic"
name: "cume_dist"
system_metadata:
Expand Down Expand Up @@ -832,6 +878,30 @@ supported_window_functions:
notation: "FUNCTION"
supported_impls:
- "any_i32_any"
- source: "arithmetic"
name: "max"
system_metadata:
name: "max"
notation: "FUNCTION"
supported_impls:
- "fp32"
- "fp64"
- "i16"
- "i32"
- "i64"
- "i8"
- source: "arithmetic"
name: "min"
system_metadata:
name: "min"
notation: "FUNCTION"
supported_impls:
- "fp32"
- "fp64"
- "i16"
- "i32"
- "i64"
- "i8"
- source: "arithmetic"
name: "nth_value"
system_metadata:
Expand Down Expand Up @@ -867,3 +937,73 @@ supported_window_functions:
notation: "FUNCTION"
supported_impls:
- ""
- source: "arithmetic"
name: "std_dev"
system_metadata:
name: "stddev_samp"
notation: "FUNCTION"
supported_impls:
- "fp64"
- source: "arithmetic"
name: "sum"
system_metadata:
name: "sum"
notation: "FUNCTION"
supported_impls:
- "fp32"
- "fp64"
- "i16"
- "i32"
- "i8"
- source: "arithmetic_decimal"
name: "max"
system_metadata:
name: "max"
notation: "FUNCTION"
supported_impls:
- "dec"
- source: "arithmetic_decimal"
name: "min"
system_metadata:
name: "min"
notation: "FUNCTION"
supported_impls:
- "dec"
- source: "datetime"
name: "max"
system_metadata:
name: "max"
notation: "FUNCTION"
supported_impls:
- "date"
- "iday"
- "iyear"
- "pts"
- "ptstz"
- source: "datetime"
name: "min"
system_metadata:
name: "min"
notation: "FUNCTION"
supported_impls:
- "date"
- "iday"
- "iyear"
- "pts"
- "ptstz"
- source: "spark"
name: "avg"
system_metadata:
name: "avg"
notation: "FUNCTION"
supported_impls:
- "dec"
- "i64"
- source: "spark"
name: "sum"
system_metadata:
name: "sum"
notation: "FUNCTION"
supported_impls:
- "dec"
- "i64"
102 changes: 102 additions & 0 deletions spark/src/main/resources/spark.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,42 @@
---
urn: extension:substrait:spark
scalar_functions:
- name: unscaled_value
description: >-
Return the decimal's unscaled integer as a signed i64. Return null for
null input and raise an error if the unscaled integer does not fit i64.
impls:
- args:
- name: x
value: DECIMAL<P,S>
nullability: MIRROR
return: i64
- name: make_decimal
description: >-
Interpret an i64 as an unscaled decimal with the precision and scale of
the type argument. Return null for null input and raise an error when
the unscaled value exceeds the target precision.
impls:
- args:
- name: value
value: i64
- name: target
type: DECIMAL<P,S>
nullability: MIRROR
return: DECIMAL<P,S>
- name: try_make_decimal
description: >-
Interpret an i64 as an unscaled decimal with the precision and scale of
the type argument. Return null for null input or when the unscaled value
exceeds the target precision.
impls:
- args:
- name: value
value: i64
- name: target
type: DECIMAL<P,S>
nullability: DECLARED_OUTPUT
return: DECIMAL?<P,S>
- name: add
description: >-
Adds days to a date
Expand All @@ -26,3 +62,69 @@ scalar_functions:
- name: days
value: i32
return: date

aggregate_functions:
- name: sum
description: >-
Spark sum. Decimal inputs have a decimal<min(P+10,38),S> buffer type
and result; i64 inputs use an i64 accumulator and result. Ignore null
inputs and return null for an empty or all-null group. Overflow returns
null for NULL, raises an error for ERROR, or wraps for WRAP (i64 only).
Decimal updates and merges do not check the buffer's declared precision.
Finalization checks the result precision. An out-of-range live decimal
buffer can therefore recover after cancellation, even after an earlier
window result was null. Materialization that checks the buffer precision
can instead store null; a stored null propagates through updates and merges
and is handled by the overflow option at finalization. Decimal overflow
outcomes can depend on physical buffer materialization. For i64 with NULL,
overflow during addition makes the accumulator remain null. Only
INITIAL_TO_RESULT is supported.
impls:
- args:
- name: x
value: DECIMAL<P,S>
options:
overflow:
values: ["NULL", ERROR]
nullability: DECLARED_OUTPUT
decomposable: NONE
return: DECIMAL?<min(P+10,38),S>
- args:
- name: x
value: i64
options:
overflow:
values: [WRAP, "NULL", ERROR]
nullability: DECLARED_OUTPUT
decomposable: NONE
return: i64?
- name: avg
description: >-
Spark average. Decimal inputs use a decimal<min(P+10,38),S> sum
buffer and an i64 count, then divide with HALF_UP rounding to
decimal<min(P+4,38),min(S+4,38)>. i64 inputs use an fp64 sum and i64
count and return fp64. Ignore null inputs and return null for an empty
or all-null group. The decimal sum buffer has the same unchecked updates,
merges and materialization behavior as sum:dec. A null sum buffer or result
overflow returns null for NULL and raises an error for ERROR at finalization.
Decimal overflow outcomes can depend on physical buffer materialization.
Only INITIAL_TO_RESULT is supported.
impls:
- args:
- name: x
value: DECIMAL<P,S>
options:
overflow:
values: ["NULL", ERROR]
nullability: DECLARED_OUTPUT
decomposable: NONE
return: DECIMAL?<min(P+4,38),min(S+4,38)>
- args:
- name: x
value: i64
options:
overflow:
values: ["NULL", ERROR]
nullability: DECLARED_OUTPUT
decomposable: NONE
return: fp64?
22 changes: 13 additions & 9 deletions spark/src/main/scala/io/substrait/spark/SparkExtension.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
*/
package io.substrait.spark

import io.substrait.spark.expression.{ToAggregateFunction, ToWindowFunction}
import io.substrait.spark.expression.{SparkAggregateFunctions, ToAggregateFunction, ToWindowFunction}

import io.substrait.extension.{DefaultExtensionCatalog, SimpleExtension}

Expand All @@ -42,19 +42,23 @@ object SparkExtension {
ret.toSeq
}

/**
* Standard extensions only: `spark.yml` declares no aggregates or windows, and
* [[toAggregateFunction]] / [[toWindowFunction]] bind against these same collections, so a
* `spark.yml` aggregate would be neither advertised in the dialect nor bindable at runtime.
*/
lazy val StandardAggregateFunctions: Seq[SimpleExtension.AggregateFunctionVariant] =
EXTENSION_COLLECTION.aggregateFunctions().asScala.toSeq

/** @see [[StandardAggregateFunctions]] */
lazy val StandardWindowFunctions: Seq[SimpleExtension.WindowFunctionVariant] =
EXTENSION_COLLECTION.windowFunctions().asScala.toSeq

val toAggregateFunction: ToAggregateFunction = ToAggregateFunction(StandardAggregateFunctions)
/** Functions bound by both the runtime converters and the generated dialect. */
lazy val AggregateFunctions: Seq[SimpleExtension.AggregateFunctionVariant] =
StandardAggregateFunctions.filterNot(SparkAggregateFunctions.replaces) ++
SparkImpls.aggregateFunctions().asScala

lazy val WindowFunctions: Seq[SimpleExtension.WindowFunctionVariant] =
StandardWindowFunctions.filterNot(SparkAggregateFunctions.replaces) ++
SparkImpls.windowFunctions().asScala

val toAggregateFunction: ToAggregateFunction = ToAggregateFunction(AggregateFunctions)

val toWindowFunction: ToWindowFunction = ToWindowFunction(WindowFunctions)

val toWindowFunction: ToWindowFunction = ToWindowFunction(StandardWindowFunctions)
}
1 change: 1 addition & 0 deletions spark/src/main/scala/io/substrait/spark/expression/FunctionMappings.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,7 @@ class FunctionMappings {
}

val SCALAR_SIGS: Seq[Sig] = Seq(
s[UnscaledValue]("unscaled_value"),
s[Add]("add"),
s[Subtract]("subtract"),
s[Multiply]("multiply"),
Expand Down
Loading
Loading