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

[nielspardon]

The generator restated things it could derive, and each restatement was a way for the published spark_dialect.yaml to drift without a test failing.

Aggregate and window functions came from the collection merged with spark.yml while the runtime converters bind against the standard extensions only, so an aggregate added to spark.yml would be advertised and then fail with "Unable to find binding for call". All three sections now come from the collections SparkExtension hands the runtime, passed to the generator explicitly.

The dependencies block was an unsorted Map, so its order was an artifact of one Scala version's String hashing; it is now built as a SortedMap from the URNs the emitted functions actually reference. A hand-maintained URN-to-alias map returned "" for an unmapped URN, emitting a source that dangled against dependencies while still validating against the schema, which declares source as a plain string; the alias is now derived from the URN's last segment, and two URNs deriving the same alias fail rather than one silently displacing the other.

The dialect is emitted through io.substrait.dialect.Dialect rather than a parallel set of Scala case classes that typed enums as String, so the generator can no longer express a dialect core would reject, and a dialect-schema field core gains no longer has to be added a second time before Spark can express it.

The published file is unchanged apart from key order: core's field order, dependencies sorted, and max_precision ahead of system_metadata. Comparing the parsed models cannot catch an ordering change, since Dialect.dependencies is a Map, so the published text is now compared as text — which in turn needs the file declared as a Test input so that editing it invalidates the tests.

Worth knowing while reviewing: the aggregate/window fix has no test that can fail today. With spark.yml declaring no aggregate or window function, the merged collection and the standard collection are indistinguishable, so reverting that wiring alone breaks nothing observable. The guard checks the advertised aggregates and windows against DefaultExtensionCatalog.DEFAULT_COLLECTION directly rather than against what the generator was handed, so it arms the moment spark.yml gains one — the same moment the bug would go live. dependencies is still derived from function URNs only, so a USER_DEFINED supported type would need its own alias folded in; the dangling-source test covers type sources too, so that would fail rather than ship.

Two unreachable branches in the function probe went with the rewrite, one of them a println aimed at the same System.out that main writes the dialect to.

Closes #1087

BREAKING CHANGE: io.substrait.spark.utils.Dialect, SupportedType, TypeMetadata, FunctionMetadata and SupportedFunction are removed; the dialect is modelled by io.substrait.dialect.Dialect and friends. DialectGenerator.generate() returns io.substrait.dialect.Dialect, and the DialectGenerator class now takes the scalar, aggregate and window function collections it generates from.

[alexandrefimov]

Read this against main at 961f83e, which now includes #1128 — that is where the one thing I would act on comes from.

max_precision for the three temporal types is the last restatement left in supportedTypes, and it is the kind this PR is about. Some(9) says what Util.MICROSECOND_PRECISION and the conversion guard already say, and they now say 6: #1128 pinned the type conversions at exactly microseconds and fed the generator from that constant. Rebasing conflicts in DialectGenerator.scala and spark_dialect.yaml, and resolving either in this branch's favour puts the 9 back — I tried it, and ./gradlew dialect regenerates the file with max_precision: 9 in three places, which fails core's SparkDialectParseTest.parsesPrecisionTypes with expected: <6> but was: <9>. CI catches it, so this is a heads-up for the rebase rather than a defect in what is here.

Separately, I checked the premise of the aggregate/window fix, since you note it has no test that can fail today. It holds: on main the generator reads COLLECTION.aggregateFunctions() where COLLECTION = EXTENSION_COLLECTION.merge(SparkImpls), while toAggregateFunction and toWindowFunction are built from EXTENSION_COLLECTION alone. So the advertised set really is the wider one, and the scalar side was already consistent because the generator used SparkScalarFunctions, which is the merged one on both sides.

[nielspardon]

Rebased onto 961f83e and resolved both conflicts in #1128's favour: supportedTypes reads Util.MICROSECOND_PRECISION and carries your comment, so ./gradlew dialect emits max_precision: 6 in all three places. The regenerated file's content now matches main's exactly — the only difference left is the key ordering this PR introduces (core's field order, dependencies sorted, max_precision ahead of system_metadata). Your suggestion applies as-is apart from scalafmt reflowing two of the three calls, which are over 100 columns on one line.

Thanks for checking the aggregate/window premise independently — that was the part I could not pin with a test.

[alexandrefimov]

Re-read this after the rebase. Checked the claim about the published file by regenerating rather than by reading: ./gradlew :spark:spark-4.0_2.13:dialect rewrites spark_dialect.yaml byte for byte, and a normalised comparison against main's copy is identical, so the only difference really is key order.

Two things inline, plus one that would not anchor because the line falls between the diff hunks. In DialectSuite, generate validated YAML asserts only that the file exists, and main calls f.createNewFile() before it constructs the writer — so exists() holds even if nothing was written, and the test passes whatever the generator emits. Comparing the file's content against published would make it cover the path it is named for. (The temp file is deleted on entry but not on exit, so build/tmp/test/dialect.yaml outlives the run.)

None of the three is blocking.

[alexandrefimov]

Re-checked this on top of current main (934a60e), since the branch sits on 961f83e: the merge is clean — only the three build.gradle.kts files auto-merge — and ./gradlew build on the merged tree passes, with DialectSuite's ten tests green on all three variants. That includes the byte-for-byte comparison, so the published file matches what the generator emits at this main.

The precision question from my first pass is settled by the branch as it stands: supportedTypes reads Util.MICROSECOND_PRECISION, and the published file carries max_precision: 6 in all three places.

The three things I left earlier are non-blocking and independent of each other: PathSensitivity.NONE on the new inputs.file, the name of SparkAggregateFunctions/SparkWindowFunctions sitting next to the merged SparkScalarFunctions, and generate validated YAML, which asserts only that the file exists — main creates it before writing, so it passes whatever is written; the new byte-for-byte test covers the content, which leaves that one covering the CLI path alone.

[nielspardon]

All three applied in 350ce87.

PathSensitivity.NONE on the new inputs.file: -Dorg.gradle.caching.debug=true now prints IGNORED_PATH{...spark_dialect.yaml=IGNORED / 98ceb602...} for publishedDialect instead of ABSOLUTE_PATH.

Renamed to StandardAggregateFunctions / StandardWindowFunctions rather than commenting the lines — both are new in this branch, so it costs nothing beyond it, and the name carries further than a comment does.

generate validated YAML now reads the file back and compares it to published, and deletes it on the way out. Checked the premise the way you'd want rather than by reading: dropping out.write(yaml) from main makes the rewritten test fail where the old one passed. It's the only test covering the CLI write path, so it's named for that now.

Filing the two you set aside separately: the dialect task's missing outputs, and COLLECTION having no main-source consumer left.

[nielspardon]

Filed the two follow-ups: #1233 (the dialect task's missing outputs) and #1234 (COLLECTION without a main-source consumer).

#1233 turned out sharper than the ordering question it started as. Because dialect declares no output, Gradle fingerprints spark_dialect.yaml at whatever state the test task finds it in — so --parallel :spark:spark-4.0_2.13:dialect :spark:spark-3.4_2.12:test on a hand-edited file passes and banks that pass under the edited file's fingerprint. Re-applying the same edit then gets UP-TO-DATE and a green build on content that fails under --rerun-tasks. Repro steps are in the issue, along with your dependsOn-defeats-the-comparison point, which is why the ordering half needs mustRunAfter rather than the obvious fix.

[alexandrefimov]

Rechecked 4f5c32b2. All three follow-ups look good: the path is excluded from the dialect input fingerprint, the standard collection names are used consistently, and the CLI test checks the written YAML.

DialectSuite passes on Spark 3.4, 3.5 and 4.0 (10 tests each). Removing out.write(yaml) makes the CLI test fail; the other nine still pass.

[andrew-coleman]

The derived-model refactor holds up well. I checked the things most likely to break and they're all fine:

• Byte-for-byte round-trip: Dialect's declaration order matches the reordered spark_dialect.yaml exactly, SupportedTypeSerializer writes max_precision before system_metadata matching the moved keys, and the function-section field orders are untouched.
• The DDL asymmetry is correct — the likeliest silent-mismatch spot. write_types routes back to ddlWriteTypes only for RelationKind.DDL, so .addDdlWriteTypes(NAMED_OBJECT) round-trips to an equal model.
• The derived dependencies block is the same 12 alias→URN pairs as the old hardcoded map, all still referenced by an emitted function, and URN_CHECKER guarantees lastIndexOf(':') lands on the name segment.
• The aggregate/window narrowing is a genuine no-op today — spark.yml declares only scalar_functions:.
• Ordering is deterministic: sortedFunctions' (source, name, impls) key is a total order over all 87 emitted functions with no duplicate (section, source, name) triples, and SortedMap fixes the one real cross-Scala-version nondeterminism.
• No dangling references to the removed case classes, and the two removed branches were genuinely unreachable given the isAssignableFrom guard above them.

Five comments below, all low severity. The build-wiring one is the only one I'd call a real gap; the rest are a leak, a determinism nit, a tautological assertion, and one judgement call worth making explicitly.

I applied all five in a scratch worktree and verified them rather than eyeballing: :spark:spark-4.0_2.13:test, :spark:spark-3.5_2.12:test and :spark:spark-3.4_2.12:test all pass at 270 tests each with zero failures, and spotlessCheck is clean. Where a suggestion claims to catch something, I mutation-tested that it does — details in the individual comments. One correction to my own reasoning is noted inline: outputs.file alone does not order the tasks, which I only found by testing it.

[andrew-coleman]

dialect rewrites this same file, and nothing orders the two tasks: ./gradlew test dialect runs test first, so it can validate the pre-regeneration content, pass, and leave the file modified. Confirmed with --dry-run.

Declaring the file as an outputs.file on dialect is not enough on its own — Gradle does not infer the dependency from plain-String inputs.file/outputs.file pairs, and I verified test still ran first with the output declared. The ordering has to be explicit:

Suggested change

	inputs
	.file("../spark_dialect.yaml")
	.withPropertyName("publishedDialect")
	.withPathSensitivity(PathSensitivity.NONE)
	inputs
	.file("../spark_dialect.yaml")
	.withPropertyName("publishedDialect")
	.withPathSensitivity(PathSensitivity.NONE)
	// The dialect task rewrites that same file and Gradle does not infer the ordering from the
	// input/output declarations, so without this `./gradlew test dialect` validates the
	// pre-regeneration content.
	mustRunAfter("dialect")

Worth adding outputs.file("../spark_dialect.yaml") to the registration on line 149 as well, separately — that gets dialect up-to-date checking instead of re-running on every invocation:

tasks.register<JavaExec>("dialect") {
  classpath = java.sourceSets["main"].runtimeClasspath
  mainClass = "io.substrait.spark.utils.DialectGenerator"
  args = listOf("../spark_dialect.yaml")
  outputs.file("../spark_dialect.yaml")
}

I checked drift is still handled with that in place: corrupting the YAML makes the task re-run and repair it, so the only visible change is dialect reporting UP-TO-DATE when nothing moved.

Note this variant is the only one that needs mustRunAfter — dialect is registered here alone, so the same line in the 3.4/3.5 build files would fail on an unknown task. A short comment there explaining why the input is content-only would save the next reader the trip.

[andrew-coleman]

The FileWriter is never closed — main only flushes. The new CLI test at DialectSuite:68 reads the file back immediately and passes because of that flush, so a future refactor dropping it would silently produce a truncated dialect. f.createNewFile() is also dead: new FileWriter(t) already creates the file.

Pre-existing, but the new test now depends on it. The lines to change are below the diff so this can't be a suggestion:

  def main(args: Array[String]): Unit = {
    val yaml = generateYaml()

    args match {
      case Array(t) => Files.writeString(Path.of(t), yaml)
      case _ =>
        val out = new OutputStreamWriter(System.out)
        out.write(yaml)
        out.flush()
    }
  }

with import java.nio.file.{Files, Path} added and File/FileWriter dropped from the line 14 import. I verified this: ./gradlew dialect regenerates spark_dialect.yaml byte-identically, and all 10 DialectSuite cases still pass including the CLI one.

[andrew-coleman]

The narrowing makes the generator match the runtime, which is right. The side effect is that a spark.yml aggregate would now be silently absent from the dialect while ToSubstraitRel:108 still cannot bind it — the loud runtime failure survives, but the dialect no longer hints at why. Pointing these at COLLECTION instead would make both truthful.

If the narrowing is the deliberate choice, worth marking so the next person doesn't read the omission as an oversight:

Suggested change

	lazy val StandardAggregateFunctions: Seq[SimpleExtension.AggregateFunctionVariant] =
	EXTENSION_COLLECTION.aggregateFunctions().asScala.toSeq
	val toWindowFunction: ToWindowFunction = ToWindowFunction(
	lazy val StandardWindowFunctions: Seq[SimpleExtension.WindowFunctionVariant] =
	EXTENSION_COLLECTION.windowFunctions().asScala.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

[andrew-coleman]

groups at :284 is a hash-ordered Map, and this folds over that order, so which URN is reported as other in the collision message differs between Scala 2.12 and 2.13. sortedFunctions fixes the emitted order but not this one. Cosmetic — the test only asserts contains("extra") — but it is a one-word fix:

Suggested change

	private def dependencies(functions: Seq[SourcedFunction]): SortedMap[String, String] =
	functions.map(_.urn).distinct.foldLeft(SortedMap.empty[String, String]) {
	(deps, urn) =>
	val alias = dependencyAlias(urn)
	deps.get(alias) match {
	case Some(other) if other != urn =>
	throw new IllegalStateException(
	s"Dependency alias '$alias' is claimed by both '$other' and '$urn'")
	case _ => deps + (alias -> urn)
	}
	}
	private def dependencies(functions: Seq[SourcedFunction]): SortedMap[String, String] =
	functions.map(_.urn).distinct.sorted.foldLeft(SortedMap.empty[String, String]) {
	(deps, urn) =>
	val alias = dependencyAlias(urn)
	deps.get(alias) match {
	case Some(other) if other != urn =>
	throw new IllegalStateException(
	s"Dependency alias '$alias' is claimed by both '$other' and '$urn'")
	case _ => deps + (alias -> urn)
	}
	}

Separately, on the aliasing scheme itself: because the alias is only the URN's third segment, a dialect that legitimately spans extension:io.substrait:functions_string and extension:acme:functions_string cannot be generated at all, and there is no hook to supply an alias. Failing loudly beats silently dropping one, agreed. But graceful degradation (qualify with the namespace on collision) would need the collision set threaded into both dependencies here and the source(dependencyAlias(urn)) call at :291, since those two have to stay in agreement — so it is a design call rather than a patch, and worth a deliberate "not now" if that is the answer.

[andrew-coleman]

DialectFunction.source is dependencyAlias(urn) (DialectGenerator:291) and the dependencies keys are dependencyAlias(urn) over the same URNs (:100), so for functions this assertion is a tautology. Its only non-trivial clause is the types one, and the generator never sets SupportedType.source(), so that side is always empty. As written it documents intent but cannot regress.

Adding the reverse direction gives it something real to catch:

Suggested change

	test("every source resolves to a declared dependency") {
	val dialect = DialectGenerator.generate()
	val aliases = dialect.dependencies().asScala.keySet
	val functions = allFunctions(dialect)
	assert(functions.nonEmpty)
	// The dialect schema declares `source` as a plain string, so a dangling alias validates
	// against it; nothing but this assertion ties the two sections together. Types are covered as
	// well as functions: `dependencies` is derived from the functions' URNs, so a USER_DEFINED type
	// pointing at an extension no function comes from would dangle.
	val sources = functions.map(_.source()) ++
	dialect.supportedTypes().asScala.toSeq.flatMap(t => Option(t.source().orElse(null)))
	assertResult(Seq.empty)(sources.distinct.filterNot(aliases.contains))
	test("every source resolves to a declared dependency") {
	val dialect = DialectGenerator.generate()
	val aliases = dialect.dependencies().asScala.keySet
	val functions = allFunctions(dialect)
	assert(functions.nonEmpty)
	// `source` is a plain string in the schema, so a dangling alias validates against it. The
	// forward direction cannot fail while `dependencies` and `source` are both derived from
	// `dependencyAlias`; the reverse guards `dependencies` gaining an alias no emitted function
	// references.
	val sources = functions.map(_.source()) ++
	dialect.supportedTypes().asScala.toSeq.flatMap(t => Option(t.source().orElse(null)))
	assertResult(Seq.empty)(sources.distinct.filterNot(aliases.contains))
	assertResult(Seq.empty)(aliases.toSeq.filterNot(sources.contains))
	}

I mutation-tested this to check it isn't redundant with the byte-for-byte test. Injecting an unreferenced alias into dependencies and regenerating the YAML (i.e. what the real workflow does) fails exactly one test — this one. Dropping just the added line while leaving the injection in place makes everything pass again, so the byte-for-byte comparison genuinely cannot see it.