From 3ba48b057679bf94565e2fa6662ab15a33fcdad5 Mon Sep 17 00:00:00 2001 From: Adegbite Ayoade Date: Thu, 24 Sep 2026 00:17:42 +0100 Subject: [PATCH] fix(databricks): preserve non-equi join conditions in custom_extensions A Metric View join whose `on` doesn't decompose into equi-join columns (function-wrapped keys, extra filter predicates, non-equi operators) no longer fails the import. The relationship takes the first parent and child columns referenced on either side of one of the condition's equalities, the original `on` is stored in the relationship's DATABRICKS stash, and a warning says the condition was preserved but not represented natively. The exporter restores the stashed `on` verbatim, and refuses to when the join would be emitted under a different alias or position than it was imported with. A condition with no usable equality is still rejected. Closes #321 --- converters/databricks/README.md | 9 ++- .../src/ossie_databricks/_common.py | 5 ++ .../ossie_databricks/metric_view_to_ossie.py | 63 ++++++++++++++--- .../ossie_databricks/ossie_to_metric_view.py | 24 ++++++- .../tests/test_metric_view_to_ossie.py | 70 ++++++++++++++++--- converters/databricks/tests/test_roundtrip.py | 36 ++++++++++ 6 files changed, 183 insertions(+), 24 deletions(-) diff --git a/converters/databricks/README.md b/converters/databricks/README.md index 865b96386..d82b2aaf6 100644 --- a/converters/databricks/README.md +++ b/converters/databricks/README.md @@ -84,7 +84,7 @@ Each row maps in both directions; the **Notes** flag where a behavior is specifi | `description` | `comment` | Model-level description only. | | root dataset | `source` | The fact/grain. | | other `datasets` | nested `joins[]` | Export: the relationship graph is reassembled into the join tree; a dataset reached by two paths (a diamond) fans out into one aliased join per path. | -| `relationship` `from_columns`/`to_columns` | join `on` (differing names) / `using` (shared names) | Decomposed into columns on import; rebuilt into `on`/`using` on export. | +| `relationship` `from_columns`/`to_columns` | join `on` (differing names) / `using` (shared names) | Decomposed into columns on import; rebuilt into `on`/`using` on export. An `on` that doesn't decompose (function-wrapped keys, a filter predicate, a non-equi operator) is imported with the first column referenced on each side of one of its equalities, and the full condition is stashed and restored verbatim on export (with a warning). | | `relationship.from`/`to` direction | join `cardinality` | Export: source on the many (`from`) side -> `many_to_one`; on the one (`to`) side -> `one_to_many`. | | `dataset.primary_key` / `unique_keys` | join `rely.at_most_one_match` | Both directions: export sets `at_most_one_match` when a key covers the join columns; import recovers a `unique_keys` from it. | | `dataset.fields[]` | `dimensions[]` | Export: fields flatten into one list and a joined column is qualified by its full join path (`customer.c_name`; `customer.region.r_name` when nested). | @@ -106,8 +106,11 @@ invalid) when an input breaks one of these: multiple candidate facts without `--source`, is rejected (a diamond is allowed and fanned out); - a join has no condition (a cross join has no Apache Ossie relationship form); -- a join condition is non-equi or otherwise can't be decomposed into equi-join columns - (Apache Ossie relationships are equi-joins, so the join has no Apache Ossie representation); +- a join condition can't be decomposed into equi-join columns and no equality in it + compares a parent column to a child column (Apache Ossie relationships need at least + one column pair); +- a stashed join condition would be exported under a different join alias or position + than it was imported with (for example after `--source` re-roots the tree); - the input YAML is malformed. ## Development diff --git a/converters/databricks/src/ossie_databricks/_common.py b/converters/databricks/src/ossie_databricks/_common.py index 97f872ec2..f2cd508e3 100644 --- a/converters/databricks/src/ossie_databricks/_common.py +++ b/converters/databricks/src/ossie_databricks/_common.py @@ -68,6 +68,11 @@ # fact on re-export. Absent for plain many-to-one stars, so they stay clean. STASH_SOURCE_KEY = "source_dataset" +# Relationship-level stash key holding a join `on` that doesn't decompose into equi-join +# columns (function-wrapped keys, extra filter predicates, non-equi operators). The +# relationship's from/to columns are only an approximation of it; export restores it. +STASH_ON_KEY = "on" + # A bare SQL identifier (single column reference), e.g. `c_name`. Used to decide # whether an expression can be safely alias-prefixed on export / de-prefixed on # import. diff --git a/converters/databricks/src/ossie_databricks/metric_view_to_ossie.py b/converters/databricks/src/ossie_databricks/metric_view_to_ossie.py index 4799416b8..e6ab10fd7 100644 --- a/converters/databricks/src/ossie_databricks/metric_view_to_ossie.py +++ b/converters/databricks/src/ossie_databricks/metric_view_to_ossie.py @@ -21,8 +21,10 @@ tree). Metric View features Apache Ossie has no native field for -- filter, window, format, rely, cardinality, parameters, materialization -- are preserved in `custom_extensions[DATABRICKS]` so that converting back reproduces the original view. -A join condition an Apache Ossie relationship cannot represent (a non-equi or cross join) is -rejected, not stashed. See README.md. +A join condition that doesn't decompose into equi-join columns (function-wrapped keys, extra +filter predicates, a non-equi operator) is kept verbatim in the relationship's stash when a +column pair can still be read from one of its equalities; otherwise it is rejected, as is a +cross join. See README.md. Usage (CLI): ossie-databricks import -i view.yaml [-o model.yaml] [--name NAME] @@ -38,11 +40,13 @@ DIALECT_DATABRICKS, MV_VERSION, OSSIE_VERSION, + STASH_ON_KEY, STASH_SOURCE_KEY, dump_yaml, is_simple_identifier, last_identifier, load_yaml, + read_stash, require, require_str, validate_source, @@ -63,8 +67,13 @@ def _warn(scope, msg): # Operators that mean a join condition is NOT a simple equi-join (so it cannot be -# expressed as from_columns/to_columns, and the join is rejected on import). +# expressed as from_columns/to_columns; the raw condition is stashed instead). _NON_EQUI_RE = re.compile(r"[<>!]=|<>|[<>]") +# A single `=` (not part of `<=`, `>=`, `!=`, `==`). +_EQ_RE = re.compile(r"(?!=])=(?!=)") +# A qualified column reference `alias.column` (not part of a longer dotted path or a +# `schema.function(` call). +_QUALIFIED_REF_RE = re.compile(r"(? equal lists on both sides. Two distinct list objects, so the # emitted YAML doesn't serialize one as an anchor/alias of the other. @@ -241,16 +261,39 @@ def _convert_join(join, parent_name, parent_alias, child): "from_columns": parent_cols, "to_columns": child_cols} stash = {k: join[k] for k in _JOIN_STASH_KEYS if k in join} + if raw_on is not None: + stash[STASH_ON_KEY] = raw_on write_stash(rel, stash) return rel +def _column_pair(on, parent_aliases, child_alias): + """Best-effort (parent_column, child_column) for a condition `_decompose_on` couldn't + split, or None. Takes the first `AND` clause holding a single `=` whose one side + references only the parent (via a qualifier in `parent_aliases`) and whose other side + references only the child, and returns the first column referenced on each side -- + e.g. `UPPER(source.x) = UPPER(COALESCE(dim.a, dim.b))` -> ('x', 'a').""" + for clause in re.split(r"\s+AND\s+", on, flags=re.IGNORECASE): + sides = _EQ_RE.split(clause) + if len(sides) != 2: + continue + refs = [_QUALIFIED_REF_RE.findall(side) for side in sides] + if not all(refs): + continue + aliases = [{a for a, _ in r} for r in refs] + for p, c in ((0, 1), (1, 0)): + if aliases[p] <= parent_aliases and aliases[c] == {child_alias}: + return refs[p][0][1], refs[c][0][1] + return None + + def _decompose_on(join, parent_alias, parent_name, child_alias): """Return (from_columns, to_columns, raw_on). raw_on is None when `on` decomposes cleanly into equi-join column pairs; it holds the original string otherwise (a non-equi/complex condition the caller - rejects). `using` short-circuits to empty columns here and is handled by the caller. + stashes, or rejects if no column pair can be read from it). `using` short-circuits + to empty columns here and is handled by the caller. The child side of a clause is always referenced by its join name. The parent side may be referenced by its alias (`source` at the top level, else the parent join diff --git a/converters/databricks/src/ossie_databricks/ossie_to_metric_view.py b/converters/databricks/src/ossie_databricks/ossie_to_metric_view.py index 99fa51034..1cbfb5fe3 100644 --- a/converters/databricks/src/ossie_databricks/ossie_to_metric_view.py +++ b/converters/databricks/src/ossie_databricks/ossie_to_metric_view.py @@ -34,6 +34,7 @@ ConversionError, MV_VERSION, OSSIE_VERSION, + STASH_ON_KEY, STASH_SOURCE_KEY, SYNONYM_LIMIT, dump_yaml, @@ -409,7 +410,11 @@ def _build_join(node, parent_alias, datasets): # it -- from_columns when it is the `from`, to_columns when it is the `to`. parent_cols, child_cols = ( (from_cols, to_cols) if node["parent_is_from"] else (to_cols, from_cols)) - if parent_cols == child_cols: + if STASH_ON_KEY in stash: + # A condition the importer couldn't decompose (the columns only approximate it) + # is restored verbatim. + join["on"] = _stashed_on(rel, node, stash) + elif parent_cols == child_cols: # Equal column lists are an equi-join on shared names -> `using`, which # round-trips faithfully (the importer maps `using` to equal lists). join["using"] = list(parent_cols) @@ -438,6 +443,23 @@ def _build_join(node, parent_alias, datasets): return join +def _stashed_on(rel, node, stash): + """Return the stashed raw `on`, provided the join is emitted as it was imported: the + joined dataset keeps its original join name as its alias and is still the child of + the relationship. The condition names the original aliases, so it can't be reused + under a renamed (fanned-out) alias or a re-rooted join tree.""" + same_direction = node["parent_is_from"] == ( + str(stash.get("cardinality") or "").lower() != CARD_ONE_TO_MANY) + if node["alias"] != node["dataset"] or not same_direction: + raise ConversionError( + f"Relationship '{rel.get('name')}': its stashed join condition " + f"('on: {stash[STASH_ON_KEY]}') refers to the join aliases it was imported " + f"with, but this join tree emits it as '{node['alias']}' in a different " + f"position; cannot re-qualify it." + ) + return stash[STASH_ON_KEY] + + def _covers_unique_key(dataset, join_cols): """True if `join_cols` include a declared `primary_key` or one of `unique_keys` of `dataset` -- i.e. joining on them matches at most one target row, so a many_to_one diff --git a/converters/databricks/tests/test_metric_view_to_ossie.py b/converters/databricks/tests/test_metric_view_to_ossie.py index cf8d9c0d8..824dbf85f 100644 --- a/converters/databricks/tests/test_metric_view_to_ossie.py +++ b/converters/databricks/tests/test_metric_view_to_ossie.py @@ -125,16 +125,66 @@ def test_non_equi_on_rejected(): def test_complex_equi_on_rejected(): - """An equi `on` whose operand is a SQL fragment (OR, computed) can't be decomposed - into from/to columns, so it's rejected rather than producing schema-invalid Apache Ossie with - empty column lists.""" - for cond in ("source.a = dim.b OR source.c = dim.d", "source.a = dim.b + 1"): - mv = ( - "version: '1.1'\nsource: c.s.fact\n" - f"joins:\n- name: dim\n source: c.s.dim\n on: {cond}\n" - ) - with pytest.raises(ConversionError, match="non-equi"): - importer.convert_metric_view_to_ossie(mv) + """An `on` with no equality that reads as one parent column = one child column (here + an OR of two) has no column pair to put in from/to columns, so it's rejected rather + than producing schema-invalid Apache Ossie with empty column lists.""" + mv = ( + "version: '1.1'\nsource: c.s.fact\n" + "joins:\n- name: dim\n source: c.s.dim\n on: source.a = dim.b OR source.c = dim.d\n" + ) + with pytest.raises(ConversionError, match="non-equi"): + importer.convert_metric_view_to_ossie(mv) + + +def _import_with_warnings(mv): + import warnings + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + ossie = parse(importer.convert_metric_view_to_ossie(mv)) + return ossie, [str(w.message) for w in caught] + + +def test_function_wrapped_on_is_stashed(): + """A join on function-wrapped keys (apache/ossie#321) is imported: the relationship + takes the first column referenced on each side of the equality, and the original + condition is stashed verbatim, with a warning.""" + cond = "UPPER(source.EXTERNAL_ID) = UPPER(COALESCE(accounts.ID_1, accounts.ID_2))" + mv = ("version: '1.1'\nsource: c.s.events\n" + f"joins:\n- name: accounts\n source: c.s.accounts\n on: {cond}\n") + ossie, warned = _import_with_warnings(mv) + rel = canon(ossie)["relationships"][0] + assert rel["from"] == "events" and rel["to"] == "accounts" + assert rel["from_columns"] == ["EXTERNAL_ID"] + assert rel["to_columns"] == ["ID_1"] + assert rel["custom_extensions"][0]["data"]["on"] == cond + assert any("join 'accounts'" in w and "preserved verbatim" in w for w in warned) + + +def test_on_with_filter_predicate_is_stashed(): + """An equi-join plus a filter predicate keeps the equi columns and stashes the whole + condition; the child side may come first in the equality.""" + cond = "accounts.id = source.account_id AND accounts.status NOT IN ('closed', 'test')" + mv = ("version: '1.1'\nsource: c.s.events\n" + f"joins:\n- name: accounts\n source: c.s.accounts\n on: {cond}\n" + " rely: {at_most_one_match: true}\n") + ossie, _ = _import_with_warnings(mv) + rel = canon(ossie)["relationships"][0] + assert rel["from_columns"] == ["account_id"] + assert rel["to_columns"] == ["id"] + assert rel["custom_extensions"][0]["data"]["on"] == cond + # The columns only approximate the condition, so no unique key is inferred from rely. + accounts = next(d for d in ossie["datasets"] if d["name"] == "accounts") + assert "unique_keys" not in accounts + + +def test_plain_equi_on_has_no_stashed_condition(): + """A decomposable equi-join is unchanged: no `on` in the stash, no warning.""" + mv = ("version: '1.1'\nsource: c.s.orders\n" + "joins:\n- name: customer\n source: c.s.customer\n" + " on: source.o_custkey = customer.c_custkey\n") + ossie, warned = _import_with_warnings(mv) + assert "custom_extensions" not in ossie["relationships"][0] + assert not warned def test_one_to_many_join_flips_from_to_and_stashes_source(): diff --git a/converters/databricks/tests/test_roundtrip.py b/converters/databricks/tests/test_roundtrip.py index 32c2044d2..37ecd3036 100644 --- a/converters/databricks/tests/test_roundtrip.py +++ b/converters/databricks/tests/test_roundtrip.py @@ -111,3 +111,39 @@ def test_property_ossie_to_mv_to_ossie_seeded(): from _roundtrip_helpers import RandomRnd, assert_ossie_roundtrip, build_ossie for seed in range(250): assert_ossie_roundtrip(build_ossie(RandomRnd(1_000_000 + seed))) + + +def test_non_equi_on_round_trips_verbatim(): + """A join condition the importer can't decompose (function-wrapped keys, a filter + predicate, a range) is restored exactly on export, including in a nested join.""" + mv_in = ( + "version: '1.1'\nsource: c.s.events\n" + "joins:\n- name: accounts\n source: c.s.accounts\n" + " on: UPPER(source.EXTERNAL_ID) = UPPER(COALESCE(accounts.ID_1, accounts.ID_2))\n" + " joins:\n - name: plans\n source: c.s.plans\n" + " on: accounts.plan_id = plans.id AND plans.valid_from <= accounts.created_at\n" + "- name: regions\n source: c.s.regions\n" + " on: source.region = regions.code AND regions.active NOT IN ('n')\n" + " rely: {at_most_one_match: true}\n" + "measures:\n- name: n\n expr: count(*)\n" + ) + ossie = importer.convert_metric_view_to_ossie(mv_in) + mv_out = exporter.convert_ossie_to_metric_view(ossie) + assert parse(mv_out) == parse(mv_in) + + +def test_stashed_on_under_renamed_alias_rejected(): + """A stashed condition names the aliases it was imported with; if the export would + emit the join under a different alias (here, re-rooted at the joined dataset), it + can't be re-qualified and is rejected rather than emitted wrong.""" + import pytest + + from ossie_databricks import ConversionError + mv_in = ( + "version: '1.1'\nsource: c.s.events\n" + "joins:\n- name: accounts\n source: c.s.accounts\n" + " on: UPPER(source.ext_id) = UPPER(accounts.id)\n" + ) + ossie = importer.convert_metric_view_to_ossie(mv_in) + with pytest.raises(ConversionError, match="stashed join condition"): + exporter.convert_ossie_to_metric_view(ossie, source="accounts")