fix(databricks): preserve non-equi join conditions in custom_extensions

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

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.

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"(?<![\w.`])([A-Za-z_]\w*)\.([A-Za-z_]\w*)(?![\w.(`])")
 
 
 def _is_wildcard(col):
@@ -146,7 +155,10 @@ def walk(parent_name, parent_alias, joins):
             # -- recovering key info Apache Ossie would otherwise lack. Only a many_to_one join
             # has the child on the `to` side (one_to_many flips it), so this naturally
             # skips one_to_many joins.
+            # Skipped for a stashed raw `on`: its columns are only an approximation of the
+            # condition, so they can't be asserted as a key.
             if (rel["to"] == child and rel.get("to_columns")
+                    and STASH_ON_KEY not in read_stash(rel)
                     and (join.get("rely") or {}).get("at_most_one_match")):
                 child_ds["unique_keys"] = [list(rel["to_columns"])]
             walk(child, child, join.get("joins"))
@@ -218,12 +230,20 @@ def _convert_join(join, parent_name, parent_alias, child):
     # _decompose_on returns (parent-side columns, child-side columns).
     parent_cols, child_cols, raw_on = _decompose_on(join, parent_alias, parent_name, child)
     if raw_on is not None:
-        raise ConversionError(
-            f"Join '{child}' uses a non-equi or unsupported join condition ('on: {raw_on}') "
-            f"that an Apache Ossie relationship cannot represent. Apache Ossie joins are equi-joins of simple "
-            f"`alias.column` pairs (the fact side may be qualified with `source`, the source "
-            f"table name, or left bare). Cannot import."
-        )
+        pair = _column_pair(raw_on, {parent_alias, parent_name}, child)
+        if pair is None:
+            raise ConversionError(
+                f"Join '{child}' uses a non-equi or unsupported join condition ('on: {raw_on}') "
+                f"that an Apache Ossie relationship cannot represent. Apache Ossie joins are equi-joins of simple "
+                f"`alias.column` pairs (the fact side may be qualified with `source`, the source "
+                f"table name, or left bare), and no such pair could be read from any equality "
+                f"in the condition. Cannot import."
+            )
+        parent_cols, child_cols = [pair[0]], [pair[1]]
+        _warn(f"join '{child}'",
+              f"condition '{raw_on}' is not a plain equi-join; preserved verbatim in "
+              f"custom_extensions (restored on export) but represented in Apache Ossie only "
+              f"as the column pair {pair[0]} = {pair[1]}")
     if "using" in join and not parent_cols:
         # `using: [cols]` -> 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

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

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():

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")