Validate source/sink types and inject assignment casts on write - #250
Merged
Merged
Conversation
Code Coverage
|
When a materialized view or INSERT writes into a sink with a declared schema, Hoptimator previously deferred all query-vs-sink type checking to the Flink job, so a mismatch (e.g. a raw Kafka primitive key exposed as STRING projected onto a typed BIGINT key column) only failed at job submission. This adds plan-time reconciliation in PipelineRel: - New TypeCoercion classifier with a graded `castMode` hint (strict/assign/explicit, fail-closed). strict casts only cross-family assignment-compatible conversions; assign adds a character->scalar carve-out (covers STRING->BIGINT keys); explicit allows any explicitly-castable scalar pair. - PipelineRel.resolveCasts compares each projected column to the sink column, injects an assignment CAST where allowed, respects an existing user CAST/SAFE_CAST (never double-wrapped), and raises a clear SQLNonTransientException up front for incompatible columns. - Same-family differences (VARCHAR length, DECIMAL precision, nullability) are assigned as-is and enforced by the engine, matching a native INSERT; only genuine cross-family mismatches are cast. Complex ROW/ARRAY/MAP shape mismatches are always rejected. - ScriptImplementor threads per-column cast targets into the sink-aligned forced projection; all prior insert() overloads are preserved. Tests: TypeCoercionTest + PipelineRelCastTest cover the full matrix (safe/unsafe per mode, user cast, nullability, precision, complex types). Integration Quidem fixtures for Venice and k8s assert both the injected CAST and the early error. Docs updated (hints, ddl-reference). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
The hoptimator-k8s intTest runtime only has the demodb and mysql drivers, so the earlier k8s cast fixtures (which referenced VENICE stores) failed with "Cannot create JDBC driver" for jdbc:venice. Rebase them onto the in-memory demodb ADS/PROFILE schema: - k8s-cast.id: strict mode rejects a VARCHAR (profile.members.member_urn) projected onto a BIGINT sink column (a created ads table), asserted via !error. - k8s-cast-assign.id: castMode=assign injects CAST(member_urn AS BIGINT) into the generated Flink INSERT, asserted via !specify. Also fix a corrupted venice-ddl-cast.id that had duplicated/garbled blocks (backticks in the input SQL caused a lexical parse error); reduce it to the single verified assign-cast case. All three fixtures verified end-to-end against a live integration environment (hoptimator-k8s:intTest and hoptimator-venice:intTest green). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Calcite's non-coercive canCastFrom implements implicit-assignment semantics: it permits lossless widening (INTEGER->BIGINT, INTEGER->DOUBLE) but rejects lossy narrowing (BIGINT->INTEGER, DOUBLE->INTEGER, DECIMAL->INTEGER). So under strict and assign a narrowing is an early error; only explicit opts into it. Add TypeCoercionTest cases locking this in, and correct the hints / ddl-reference docs (and the EXPLICIT javadoc) which had implied strict was purely same-family. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Unit: fold a comprehensive parametrized matrix into TypeCoercionTest (single test file per source file). Covers an exact scalar matrix (numeric widening/narrowing, character coercions, temporal, binary, boolean), a complex-type matrix (ARRAY/MAP/MULTISET/ROW), and property tests over the full cartesian product of a rich type list: mode monotonicity (strict subset of assign subset of explicit), reflexivity, complex-vs-scalar incompatibility, and nullability. Integration (hoptimator-k8s, demodb + created typed sinks): strict rejects a VARCHAR->BIGINT projection; a user CAST is respected with no double-wrap; a user CAST whose result is not assignable to the sink errors; matching types (incl. nullable source into a NOT NULL sink) pass through with no cast; assign injects the cast and still rejects VARCHAR->VARBINARY and nullable->NOT NULL; explicit performs the VARBINARY cast assign refuses and reports the explicit-cast-not-assignable case. All verified end-to-end against a live environment. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
jogrogan
force-pushed
the
jogrogan/typeCoercion
branch
from
September 4, 2026 16:02
dea3a93 to
f4a0993
Compare
Test 9's guard MV projected KAFKA VALUE (BINARY) onto the BIGINT memberId sink column purely to establish a pipeline dependency. That mapping was a latent bug — it would have failed at Flink — and is now caught up front by the new plan-time type validation. Map into type-compatible sink columns (KEY -> KEY, KEY -> pageKey, both VARCHAR) instead; the dependency the test exercises is on the sink store and is unaffected by which columns are projected. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Replace the shallow equalSansNullability check with a recursive structurallyEqualSansNullability comparator so nested structs, arrays, multisets, and maps are compared for structural equality at every level while ignoring nullability throughout. A nested field that differs only in nullability is now a pass-through (matching the scalar rule and how the engine enforces NOT NULL at runtime), while nested shape, field-name, or base-type mismatches remain errors. Add 17 structural unit cases to TypeCoercionTest and 12 end-to-end pipeline cases to PipelineRelCastTest covering array-of-struct, map struct values/keys, deeply nested structs, recursive nested nullability, and Avro-union-collapsed struct shapes. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
ryannedolan
approved these changes
Sep 4, 2026
| !specify cast_asgn | ||
|
|
||
| create or replace materialized view ads."cast_asgn$to-binary" ("vb") as select member_urn AS "vb" from profile.members; | ||
| Incompatible types for sink column 'vb': query produces VARCHAR but sink expects VARBINARY (no safe conversion is available under castMode=assign). |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
When a materialized view or
INSERTwrites into a sink with a declared schema, Hoptimator previously deferred all query-vs-sink type checking to the Flink job, so a mismatch — e.g. a raw Kafka primitive key exposed asSTRINGprojected onto a typedBIGINTkey column — only failed at job submission with a cryptic Flink error. This adds plan-time reconciliation inPipelineRel: Hoptimator now compares each projected column to its sink column and either injects an assignmentCASTor raises a clear error up front.Behavior
A new
castModehint (fail-closed, defaults tostrict) controls how aggressively casts are injected. It maps directly onto Calcite'sSqlTypeUtil.canCastFrom:strict(default)INTEGER -> BIGINTSTRING -> BIGINTand lossy narrowingBIGINT -> INTEGERassignstrict+ a character-to-scalar carve-out (covers the raw keySTRING -> BIGINT)explicitRules that hold at every mode:
NOT NULLsink column that also needs a cast is rejected (a cast can't add a null guard); matching types pass through regardless of nullability.ROW/ARRAY/MAP/MULTISETshape mismatches are always rejected.CAST/SAFE_CASTis respected and never double-wrapped; if its result type isn't assignable to the sink, we fail early with a targeted message.Implementation
TypeCoercion(new):CastMode+decide(source, target, mode)classifier over Calcite's assignment/cast rules.PipelineRel.Implementor.sql():resolveCasts()validates the query against the sink and returns per-column cast targets, throwingSQLNonTransientExceptionearly on incompatibility.ScriptImplementor: threads per-column cast targets into the sink-aligned forced projection; all priorinsert()overloads are preserved (backward compatible).Note on scope: this governs the
CREATE MATERIALIZED VIEWpath (where Hoptimator resolves the sink separately, so Calcite does not coerce — the original failure). PlainINSERT INTO <existing table>is already coerced by Calcite's validator, socastModeis effectively a no-op there.Docs
docs/user-guide/hints.md— thecastModeplanner hint.docs/user-guide/ddl-reference.md— "Type checking and casts on write".Testing
TypeCoercionTest(single test file per source file): exact scalar matrix (numeric widening/narrowing, character coercions, temporal, binary, boolean), complex-type matrix, and cartesian property tests (mode monotonicity strict subset-of assign subset-of explicit, reflexivity, complex-vs-scalar, nullability). ~1400 cases..idfixtures verified end-to-end against a live environment:hoptimator-venice:venice-ddl-cast.id(assign injectsCASTinto a Venice keyed sink).hoptimator-k8s: strict reject, user-CAST respected (no double-wrap), user-CAST-not-assignable error, matching-type/nullable passthrough, assign auto-cast +VARBINARY/nullable rejections, explicit-only casts.No golden-fixture changes for existing pipelines; matching-type passthroughs are unaffected.