fix(spark)!: derive the dialect from core's model and the runtime's functions - #1133
nielspardon merged 5 commits into
Conversation
f7dd4b9 to
cb4ab38
Compare
alexandrefimov
left a comment
alexandrefimov
left a comment
There was a problem hiding this comment.
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.
cb4ab38 to
3207810
Compare
|
Rebased onto Thanks for checking the aggregate/window premise independently — that was the part I could not pin with a test. |
c58455a to
fca26d8
Compare
…unctions 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. 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. The published file is unchanged apart from key order: core's field order, dependencies sorted, and max_precision ahead of system_metadata. An ordering change cannot be caught by comparing the parsed models, since Dialect.dependencies is a Map, so the published text is now compared as text. Closes substrait-io#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.
fca26d8 to
666fe3d
Compare
alexandrefimov
left a comment
alexandrefimov
left a comment
There was a problem hiding this comment.
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
left a comment
alexandrefimov
left a comment
There was a problem hiding this comment.
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.
Name the standard collections `StandardAggregateFunctions` and `StandardWindowFunctions`, so they no longer read as a pair with the merged `SparkScalarFunctions` sitting directly above them. Both were added in this branch, so the rename costs nothing beyond it. Ignore the path when fingerprinting the published dialect as a test input. `inputs.file` defaults to absolute-path sensitivity, which made the test task non-relocatable across checkouts for no gain: only the file's content matters to `DialectSuite`. Read the file back in the CLI test instead of asserting that it exists. `main` creates the file before it opens a writer on it, so the old assertion held whatever was written -- including nothing. Confirmed by dropping the `out.write` in `main`: the test now fails where it used to pass. The temp file is also cleaned up on the way out rather than only on the next run's entry.
|
All three applied in 350ce87.
Renamed to
Filing the two you set aside separately: the |
|
Filed the two follow-ups: #1233 (the #1233 turned out sharper than the ordering question it started as. Because |
alexandrefimov
left a comment
alexandrefimov
left a comment
There was a problem hiding this comment.
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
left a comment
andrew-coleman
left a comment
There was a problem hiding this comment.
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 reorderedspark_dialect.yamlexactly,SupportedTypeSerializerwritesmax_precisionbeforesystem_metadatamatching the moved keys, and the function-section field orders are untouched. - The DDL asymmetry is correct — the likeliest silent-mismatch spot.
write_typesroutes back toddlWriteTypesonly forRelationKind.DDL, so.addDdlWriteTypes(NAMED_OBJECT)round-trips to an equal model. - The derived
dependenciesblock is the same 12 alias→URN pairs as the old hardcoded map, all still referenced by an emitted function, andURN_CHECKERguaranteeslastIndexOf(':')lands on the name segment. - The aggregate/window narrowing is a genuine no-op today —
spark.ymldeclares onlyscalar_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, andSortedMapfixes 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
isAssignableFromguard 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.
| 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) | ||
| } | ||
| } |
There was a problem hiding this comment.
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:
| 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.
There was a problem hiding this comment.
.sorted applied, and the collision test now asserts the whole message rather than contains("extra") so the order is actually pinned. One honest caveat: removing .sorted again leaves both 2.12 and 2.13 green here, so the assertion pins the message but does not demonstrate the fix on these two versions — it guards the behaviour rather than proving a current difference.
On the aliasing scheme: deliberate not-now, agreed on the reasoning. Namespace-qualifying on collision means the collision set has to be computed before either dependencies or the source(dependencyAlias(urn)) call can decide what to write, so dependencyAlias stops being a pure function of one URN and becomes a function of the whole emitted set — which is a different shape from what is here, not an addition to it. It is also unreachable from Spark today: every URN the generator emits comes from the standard extensions plus spark.yml, so no two can collide without an upstream rename. The loud failure is the right behaviour until something actually needs to span two namespaces. Happy to file it as a follow-up if you would rather have it tracked than buried in a thread.
Declare `spark_dialect.yaml` as the `dialect` task's output and order every variant's `test` after it. Gradle infers no ordering from a plain `inputs.file`/`outputs.file` overlap -- it reports the overlap as a validation failure on the producer, after the consumer has already run and passed. So `./gradlew test dialect` used to validate the pre-regeneration content, and under `--parallel` a single interleaving banked a green test result under the fingerprint of failing content, where it survived as an UP-TO-DATE pass. The task is registered on the 4.0 variant only, so all three name it by path. Write the file with `Files.writeString` in the CLI path. The old `FileWriter` was flushed but never closed, so the file was complete only by luck of the flush, and `createNewFile()` was dead work ahead of a writer that creates the file anyway. Sort the URNs the `dependencies` fold walks. They come out of a `groupBy(...).toMap`, so which of two URNs claiming one alias is reported as the incumbent depended on how a Scala version hashes them. The collision test now asserts the whole message rather than a substring, so the order is pinned. Assert that no declared dependency alias goes unreferenced, the reverse of the existing dangling-`source` check. The byte-for-byte comparison cannot see an unreferenced alias once the published file is regenerated with it. Document why the aggregate and window collections are the standard ones only. Closes substrait-io#1233
|
Understand this PR’s impact Explore downstream dependencies and potential security impact with Blast Radius. Important Review skippedReview was skipped as selected files did not have any reviewable changes. ⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Advanced Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Advanced Run ID: 📒 Files selected for processing (7)
Included review availability: Your plan provides up to 10 included reviews per hour; 9 remain after this review. 📝 WalkthroughWalkthroughThe Spark dialect generator now uses generated Substrait models and runtime-compatible function collections. It derives deterministic dependencies, updates the published YAML, adds validation coverage, and coordinates dialect generation with Spark test tasks. ChangesSpark dialect generation
Priority: ➖ Normal Estimated code review effort: 4 (Complex) | ~45 minutes Change: Bug fix Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
3 participants
The generator restated things it could derive, and each restatement was a way for the published
spark_dialect.yamlto drift without a test failing.Aggregate and window functions came from the collection merged with
spark.ymlwhile the runtime converters bind against the standard extensions only, so an aggregate added tospark.ymlwould be advertised and then fail with "Unable to find binding for call". All three sections now come from the collectionsSparkExtensionhands the runtime, passed to the generator explicitly.The
dependenciesblock was an unsortedMap, so its order was an artifact of one Scala version's String hashing; it is now built as aSortedMapfrom the URNs the emitted functions actually reference. A hand-maintained URN-to-alias map returned""for an unmapped URN, emitting asourcethat dangled againstdependencieswhile still validating against the schema, which declaressourceas 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.Dialectrather than a parallel set of Scala case classes that typed enums asString, 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,
dependenciessorted, andmax_precisionahead ofsystem_metadata. Comparing the parsed models cannot catch an ordering change, sinceDialect.dependenciesis aMap, so the published text is now compared as text — which in turn needs the file declared as aTestinput so that editing it invalidates the tests.Worth knowing while reviewing: the aggregate/window fix has no test that can fail today. With
spark.ymldeclaring 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 againstDefaultExtensionCatalog.DEFAULT_COLLECTIONdirectly rather than against what the generator was handed, so it arms the momentspark.ymlgains one — the same moment the bug would go live.dependenciesis still derived from function URNs only, so aUSER_DEFINEDsupported type would need its own alias folded in; the dangling-source test covers type sources too, so that would fail rather than ship.Declaring the file as a test input also exposed that nothing ordered the
dialecttask against thetesttasks that read what it writes. Gradle infers no ordering from a plaininputs.file/outputs.fileoverlap — it reports the overlap as a validation failure on the producer, after the consumer has already run and passed — so./gradlew test dialectvalidated the pre-regeneration content, and under--parallela single interleaving banked a green test result under the fingerprint of failing content, where it survived as anUP-TO-DATEpass. Every variant'stestnow runs after thedialecttask, which is registered on the 4.0 variant alone and so has to be named by path.Two unreachable branches in the function probe went with the rewrite, one of them a
printlnaimed at the sameSystem.outthatmainwrites the dialect to.Closes #1087
Closes #1233
BREAKING CHANGE:
io.substrait.spark.utils.Dialect,SupportedType,TypeMetadata,FunctionMetadataandSupportedFunctionare removed; the dialect is modelled byio.substrait.dialect.Dialectand friends.DialectGenerator.generate()returnsio.substrait.dialect.Dialect, and theDialectGeneratorclass now takes the scalar, aggregate and window function collections it generates from.Summary by CodeRabbit
New Features
Bug Fixes