fix(spark)!: derive the dialect from core's model and the runtime's functions

spark/spark-3.4_2.12/build.gradle.kts

@@ -156,6 +156,19 @@ tasks {
   test {
     useJUnitPlatform { includeEngines("scalatest") }
 
+    // DialectSuite reads the published dialect, so a change to it has to invalidate the tests.
+    // Only the content matters, so ignore the path and keep the task cacheable across checkouts.
+    inputs
+      .file("../spark_dialect.yaml")
+      .withPropertyName("publishedDialect")
+      .withPathSensitivity(PathSensitivity.NONE)
+
+    // The dialect task rewrites that same file, and Gradle infers no ordering from the input and
+    // output declarations -- it reports the overlap as a validation failure instead. Without this,
+    // `./gradlew test dialect` validates the pre-regeneration content. The task is registered on
+    // the 4.0 variant only, so every variant has to name it by path.
+    mustRunAfter(":spark:spark-4.0_2.13:dialect")
+
     // Set system properties for variant identification
     systemProperty("spark.version", sparkVersion)
     systemProperty("scala.version", scalaVersion)

spark/spark-3.5_2.12/build.gradle.kts

@@ -157,6 +157,19 @@ tasks {
   test {
     useJUnitPlatform { includeEngines("scalatest") }
 
+    // DialectSuite reads the published dialect, so a change to it has to invalidate the tests.
+    // Only the content matters, so ignore the path and keep the task cacheable across checkouts.
+    inputs
+      .file("../spark_dialect.yaml")
+      .withPropertyName("publishedDialect")
+      .withPathSensitivity(PathSensitivity.NONE)
+
+    // The dialect task rewrites that same file, and Gradle infers no ordering from the input and
+    // output declarations -- it reports the overlap as a validation failure instead. Without this,
+    // `./gradlew test dialect` validates the pre-regeneration content. The task is registered on
+    // the 4.0 variant only, so every variant has to name it by path.
+    mustRunAfter(":spark:spark-4.0_2.13:dialect")
+
     // Set system properties for variant identification
     systemProperty("spark.version", sparkVersion)
     systemProperty("scala.version", scalaVersion)

spark/spark-4.0_2.13/build.gradle.kts

@@ -150,6 +150,9 @@ tasks.register<JavaExec>("dialect") {
   classpath = java.sourceSets["main"].runtimeClasspath
   mainClass = "io.substrait.spark.utils.DialectGenerator"
   args = listOf("../spark_dialect.yaml")
+  // Declared so the task gets up-to-date checking, and so Gradle can see that the test tasks
+  // reading this file consume what this task produces.
+  outputs.file("../spark_dialect.yaml")
 }
 
 tasks {
@@ -163,6 +166,19 @@ tasks {
   test {
     useJUnitPlatform { includeEngines("scalatest") }
 
+    // DialectSuite reads the published dialect, so a change to it has to invalidate the tests.
+    // Only the content matters, so ignore the path and keep the task cacheable across checkouts.
+    inputs
+      .file("../spark_dialect.yaml")
+      .withPropertyName("publishedDialect")
+      .withPathSensitivity(PathSensitivity.NONE)
+
+    // The dialect task rewrites that same file, and Gradle infers no ordering from the input and
+    // output declarations -- it reports the overlap as a validation failure instead. Without this,
+    // `./gradlew test dialect` validates the pre-regeneration content. The task is registered on
+    // the 4.0 variant only, so every variant has to name it by path.
+    mustRunAfter(":spark:spark-4.0_2.13:dialect")
+
     // Set system properties for variant identification
     systemProperty("spark.version", sparkVersion)
     systemProperty("scala.version", scalaVersion)

spark/spark_dialect.yaml

@@ -1,5 +1,18 @@
 ---
 name: "Spark Dialect"
+dependencies:
+  aggregate_approx: "extension:io.substrait:functions_aggregate_approx"
+  aggregate_generic: "extension:io.substrait:functions_aggregate_generic"
+  arithmetic: "extension:io.substrait:functions_arithmetic"
+  arithmetic_decimal: "extension:io.substrait:functions_arithmetic_decimal"
+  boolean: "extension:io.substrait:functions_boolean"
+  comparison: "extension:io.substrait:functions_comparison"
+  datetime: "extension:io.substrait:functions_datetime"
+  logarithmic: "extension:io.substrait:functions_logarithmic"
+  rounding: "extension:io.substrait:functions_rounding"
+  rounding_decimal: "extension:io.substrait:functions_rounding_decimal"
+  spark: "extension:substrait:spark"
+  string: "extension:io.substrait:functions_string"
 supported_types:
 - type: "I8"
   system_metadata:
@@ -54,20 +67,20 @@ supported_types:
     name: "BooleanType"
     supported_as_column: true
 - type: "PRECISION_TIMESTAMP"
+  max_precision: 6
   system_metadata:
     name: "TimestampNTZType"
     supported_as_column: true
-  max_precision: 6
 - type: "PRECISION_TIMESTAMP_TZ"
+  max_precision: 6
   system_metadata:
     name: "TimestampType"
     supported_as_column: true
-  max_precision: 6
 - type: "INTERVAL_DAY"
+  max_precision: 6
   system_metadata:
     name: "DayTimeIntervalType"
     supported_as_column: true
-  max_precision: 6
 - type: "INTERVAL_YEAR"
   system_metadata:
     name: "YearMonthIntervalType"
@@ -84,17 +97,6 @@ supported_types:
   system_metadata:
     name: "StructType"
     supported_as_column: true
-supported_expressions:
-- "LITERAL"
-- "SELECTION"
-- "SCALAR_FUNCTION"
-- "IF_THEN"
-- "SINGULAR_OR_LIST"
-- "CAST"
-- expression: "SUBQUERY"
-  subquery_types:
-  - "SCALAR"
-  - "IN_PREDICATE"
 supported_relations:
 - "FILTER"
 - "FETCH"
@@ -125,19 +127,17 @@ supported_relations:
 - relation: "SET"
   operations:
   - "UNION_ALL"
-dependencies:
-  rounding: "extension:io.substrait:functions_rounding"
-  comparison: "extension:io.substrait:functions_comparison"
-  logarithmic: "extension:io.substrait:functions_logarithmic"
-  datetime: "extension:io.substrait:functions_datetime"
-  arithmetic: "extension:io.substrait:functions_arithmetic"
-  aggregate_generic: "extension:io.substrait:functions_aggregate_generic"
-  boolean: "extension:io.substrait:functions_boolean"
-  aggregate_approx: "extension:io.substrait:functions_aggregate_approx"
-  rounding_decimal: "extension:io.substrait:functions_rounding_decimal"
-  string: "extension:io.substrait:functions_string"
-  spark: "extension:substrait:spark"
-  arithmetic_decimal: "extension:io.substrait:functions_arithmetic_decimal"
+supported_expressions:
+- "LITERAL"
+- "SELECTION"
+- "SCALAR_FUNCTION"
+- "IF_THEN"
+- "SINGULAR_OR_LIST"
+- "CAST"
+- expression: "SUBQUERY"
+  subquery_types:
+  - "SCALAR"
+  - "IN_PREDICATE"
 supported_scalar_functions:
 - source: "arithmetic"
   name: "abs"

spark/src/main/scala/io/substrait/spark/SparkExtension.scala

@@ -42,10 +42,19 @@ object SparkExtension {
     ret.toSeq
   }
 
-  val toAggregateFunction: ToAggregateFunction = ToAggregateFunction(
-    EXTENSION_COLLECTION.aggregateFunctions().asScala.toSeq)
-
-  val toWindowFunction: ToWindowFunction = ToWindowFunction(
+  /**
+   * 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)
+
+  val toWindowFunction: ToWindowFunction = ToWindowFunction(StandardWindowFunctions)
 }