Skip to content

fix(reasoning): support Lakebase/Postgres dialect and rebuild class URIs against the current Base URI #146

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jeremiaspf wants to merge 1 commit into
base: develop
Choose a base branch
from
Open

fix(reasoning): support Lakebase/Postgres dialect and rebuild class URIs against the current Base URI #146

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 16 additions & 4 deletions src/back/core/helpers/SQLHelpers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ def sql_escape(value: str) -> str:
return str(value).replace("\\", "\\\\").replace("'", "''")

@staticmethod
def sql_cast(expr: str, sql_type: str) -> str:
def sql_cast(expr: str, sql_type: str, dialect: str = "databricks") -> str:
"""Cast a value to *sql_type* without failing the query.

Databricks warehouses run with ANSI mode on, so a plain ``CAST``
Expand All @@ -33,18 +33,30 @@ def sql_cast(expr: str, sql_type: str) -> str:
emits — stringification, typed literals, typed NULLs, numeric
coercion — must use this helper (or inline ``TRY_CAST``) instead of a
bare ``CAST``.

PostgreSQL (Lakebase) has no ``TRY_CAST``. When *dialect* is
``"postgres"`` this instead emits a regex-guarded
``CASE WHEN … THEN …::type ELSE NULL END`` that yields the same
"cast or NULL, never raise" semantics as ``TRY_CAST``.
"""
if dialect == "postgres":
pg_type = "DOUBLE PRECISION" if sql_type.upper() == "DOUBLE" else sql_type
return (
f"(CASE WHEN ({expr}) ~ '^\\s*[-+]?[0-9]*\\.?[0-9]+([eE][-+]?[0-9]+)?\\s*$' "

@benoitcayladbx benoitcayladbx Aug 23, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This postgres branch is a numeric try-cast, but sql_cast is the generic helper. The regex is applied for every sql_type.

Please either:

  • keep the regex/:: path only inside sql_numeric, and have postgres sql_cast for non-numeric types use a type-appropriate guard (or document that dialect="postgres" is numeric-only and reject other types), or
  • make the guard depend on sql_type.

Also: ({expr}) is evaluated twice (~ then ::). Column refs are fine; don’t copy this pattern onto arbitrary expressions.

"5." fails this regex; Databricks TRY_CAST accepts it. Worth a unit test so the two dialects don’t drift silently.

Add tests in tests/units/core/test_try_cast_policy.py for dialect="postgres" (no TRY_CAST, no statement-aborting bare CAST).

f"THEN ({expr})::{pg_type} ELSE NULL END)"
)
return f"TRY_CAST({expr} AS {sql_type})"

@staticmethod
def sql_numeric(expr: str, sql_type: str = "DOUBLE") -> str:
def sql_numeric(expr: str, sql_type: str = "DOUBLE", dialect: str = "databricks") -> str:
"""Cast a triple-store value to a number without failing the query.

Every object in the triple store is stored as a string, so a numeric
comparison has to cast. See :meth:`sql_cast` for why this is a
``TRY_CAST`` rather than a bare ``CAST``.
``TRY_CAST`` (or the Postgres-safe equivalent) rather than a bare
``CAST``.
"""
return SQLHelpers.sql_cast(expr, sql_type)
return SQLHelpers.sql_cast(expr, sql_type, dialect=dialect)

@staticmethod
def validate_uc_identifier(name: str, *, role: str = "identifier") -> str:
Expand Down
45 changes: 36 additions & 9 deletions src/back/core/reasoning/AggregateRuleEngine.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
from back.core.w3c.rdf_utils import uri_local_name
from back.core.reasoning.constants import AGG_FUNCTIONS, AGG_OPERATORS, RDF_TYPE
from back.core.reasoning.models import InferredTriple, ReasoningResult, RuleViolation
from back.core.helpers import sql_numeric

logger = get_logger(__name__)

Expand All @@ -33,10 +34,15 @@ def _resolve_rule(rule: Dict, ontology: Dict) -> Dict:
data_ns = base_uri.rstrip("#").rstrip("/") + "/" if base_uri else ""

uri_map: Dict[str, str] = {}

base_ns = base_uri.rstrip("#").rstrip("/") if base_uri else ""
for cls in ontology.get("classes", []):
name = cls.get("name", "") or cls.get("localName", "")
uri = cls.get("uri", "")
if not uri and name:
if base_ns and uri and not uri.startswith(base_ns):

@benoitcayladbx benoitcayladbx Aug 23, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

not uri.startswith(base_ns) is too broad.

This will rewrite any class whose URI is not under the current domain base — OWL imports, schema.org, owl:Thing, etc. — to {base_uri}{sep}{localName}. That is a behaviour change, not a rebase fix.

startswith is also the wrong namespace test: base http://example.org/foo will not treat http://example.org/foobar#X as stale.

Please extract one helper (used by all four engines) that rebuilds only when:

  • the local name matches a domain class, and
  • the stored URI’s namespace is a previous domain Base URI (or the class’s own known-stale uri),

and leaves foreign namespaces untouched.

Related: _resolve below still does if not rule.get(field_uri): return. A rule that persisted target_class_uri / result_class_uri at save time keeps the stale URI and still returns 0 rows. Rebuild those fields too when they fail the same “current base” check.

local = uri_local_name(uri)
uri = base_uri + sep + local
elif not uri and name:
uri = base_uri + sep + name
if name:
uri_map[name.lower()] = uri
Expand Down Expand Up @@ -78,6 +84,12 @@ def _resolve(field_name: str, field_uri: str, is_class: bool = False) -> None:
_resolve("result_class", "result_class_uri", is_class=True)
return rule

@staticmethod
def _dialect_for_store(store) -> str:
"""Best-effort SQL dialect detection from the store's class name."""
name = type(store).__name__.lower()
return "postgres" if "lakebase" in name or "postgres" in name else "databricks"

@benoitcayladbx benoitcayladbx Aug 23, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please don’t detect dialect from type(store).__name__. Wrappers, test doubles, and a class rename all silently fall back to Databricks SQL (TRY_CAST) on Lakebase.

GraphDBBackend.query_dialect already exists but Lakebase and Delta both return "sql", so it cannot distinguish them today. Add something like sql_flavor ("databricks" | "postgres") on the backend and read that here.

This method is also duplicated on DecisionTableEngine — one helper, not two copies.


def execute_rules(
self,
rules: List[Dict],
Expand All @@ -90,6 +102,7 @@ def execute_rules(
t0 = time.time()
result = ReasoningResult()
base_uri = ontology.get("base_uri", "")
dialect = self._dialect_for_store(store)

for rule in rules:
if not rule.get("enabled", True):
Expand All @@ -103,6 +116,7 @@ def execute_rules(
table_name,
base_uri,
materialize,
dialect,
)
result.merge(rule_result)
except Exception as e:
Expand Down Expand Up @@ -133,11 +147,12 @@ def _execute_one(
table_name: str,
base_uri: str,
materialize: bool,
dialect: str,
) -> ReasoningResult:
result = ReasoningResult()
name = rule.get("name", "unnamed")

query = self.build_sql(rule, store.sql_table_reference(table_name), base_uri)
query = self.build_sql(rule, store.sql_table_reference(table_name), base_uri, dialect)

if not query:
return result
Expand All @@ -156,7 +171,14 @@ def _execute_one(
rule_type="aggregate",
)
)
if materialize and rule.get("result_class_uri"):
# Computed for every dry-run preview, same as SWRL/Decision
# Tables/SPARQL — whether these triples get written to the
# store is decided later by the top-level "Materialise"
# action, not by this internal flag. Gating this on
# `materialize` made Aggregate Rules the only engine whose
# preview ("Run") always showed 0 inferred even when a
# Result Entity was configured and real rows matched.
if rule.get("result_class_uri"):
result.inferred_triples.append(
InferredTriple(
subject=subj,
Expand All @@ -171,7 +193,9 @@ def _execute_one(

return result

def build_sql(self, rule: Dict, table: str, base_uri: str) -> Optional[str]:
def build_sql(
self, rule: Dict, table: str, base_uri: str, dialect: str = "databricks"
) -> Optional[str]:
"""Build SQL with GROUP BY / HAVING for an aggregate rule."""
target_uri = rule.get("target_class_uri", "")
group_prop_uri = rule.get("group_by_property_uri", "")
Expand All @@ -188,28 +212,31 @@ def build_sql(self, rule: Dict, table: str, base_uri: str) -> Optional[str]:
def esc(v: str) -> str:
return v.replace("'", "''")

def cast(expr: str) -> str:
return sql_numeric(expr, dialect=dialect)

if group_prop_uri and agg_prop_uri:
return (
f"SELECT t0.subject AS s, {func}(TRY_CAST(t_agg.object AS DOUBLE)) AS agg_val\n"
f"SELECT t0.subject AS s, {func}({cast('t_agg.object')}) AS agg_val\n"
f"FROM {table} t0\n"
f"JOIN {table} t_grp ON t_grp.subject = t0.subject AND t_grp.predicate = '{esc(group_prop_uri)}'\n"
f"JOIN {table} t_agg ON t_agg.subject = t_grp.object AND t_agg.predicate = '{esc(agg_prop_uri)}'\n"
f"WHERE t0.predicate = '{RDF_TYPE}' AND t0.object = '{esc(target_uri)}'\n"
f"GROUP BY t0.subject\n"
f"HAVING {func}(TRY_CAST(t_agg.object AS DOUBLE)) {sql_op} {threshold}"
f"HAVING {func}({cast('t_agg.object')}) {sql_op} {threshold}"
)
elif agg_prop_uri:
return (
f"SELECT t0.subject AS s, {func}(TRY_CAST(t_agg.object AS DOUBLE)) AS agg_val\n"
f"SELECT t0.subject AS s, {func}({cast('t_agg.object')}) AS agg_val\n"
f"FROM {table} t0\n"
f"JOIN {table} t_agg ON t_agg.subject = t0.subject AND t_agg.predicate = '{esc(agg_prop_uri)}'\n"
f"WHERE t0.predicate = '{RDF_TYPE}' AND t0.object = '{esc(target_uri)}'\n"
f"GROUP BY t0.subject\n"
f"HAVING {func}(TRY_CAST(t_agg.object AS DOUBLE)) {sql_op} {threshold}"
f"HAVING {func}({cast('t_agg.object')}) {sql_op} {threshold}"
)
elif group_prop_uri:
agg_expr = (
f"{func}(TRY_CAST(t_grp.object AS DOUBLE))"
f"{func}({cast('t_grp.object')})"
if func != "COUNT"
else "COUNT(t_grp.object)"
)
Expand Down
74 changes: 46 additions & 28 deletions src/back/core/reasoning/DecisionTableEngine.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,19 +47,34 @@ def _is_numeric(val: str) -> bool:
except (ValueError, TypeError):
return False

@staticmethod
def _dialect_for_store(store) -> str:
"""Best-effort SQL dialect detection from the store's class name."""
name = type(store).__name__.lower()
return "postgres" if "lakebase" in name or "postgres" in name else "databricks"

@benoitcayladbx benoitcayladbx Aug 23, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same as AggregateRuleEngine._dialect_for_store: class-name sniff + duplication. Please share one backend-driven helper (see comment there).


@staticmethod
def _build_uri_map(ontology: Dict) -> Dict[str, str]:
uri_map: Dict[str, str] = {}
base_uri = ontology.get("base_uri", "")
sep = "" if base_uri.endswith("#") or base_uri.endswith("/") else "#"
data_ns = base_uri.rstrip("#").rstrip("/") + "/" if base_uri else ""

base_ns = base_uri.rstrip("#").rstrip("/") if base_uri else ""
for cls in ontology.get("classes", []):
name = cls.get("name", "") or cls.get("localName", "")
uri = cls.get("uri", "")
if not uri and name:
if base_ns and uri and not uri.startswith(base_ns):

@benoitcayladbx benoitcayladbx Aug 23, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same class-URI rewrite issue as in AggregateRuleEngine (imports get rewritten). Please call the shared helper instead of copying this loop.

Also: comments in this repo are English — please rewrite this block (and the copies in SPARQL/SWRL) in English.

# URI obsoleta de un Base URI anterior (p.ej. el dominio se
# re-baso tras crear la clase) - se reconstruye contra el
# base_uri actual, igual que ya se hace con las propiedades.
local = uri_local_name(uri)
uri = base_uri + sep + local
elif not uri and name:
uri = base_uri + sep + name
if name:
uri_map[name.lower()] = uri

for prop in ontology.get("properties", []):
name = prop.get("name", "") or prop.get("localName", "")
uri = prop.get("uri", "")
Expand Down Expand Up @@ -142,6 +157,7 @@ def _execute_one(self, dt, store, table_name, base_uri, materialize):
if not rows:
logger.debug("Decision table '%s': no rows defined, skipping", dt_name)
return result
dialect = self._dialect_for_store(store)
logger.debug(
"Decision table '%s': target_class_uri=%s, inputs=%s",
dt_name,
Expand All @@ -157,30 +173,15 @@ def _execute_one(self, dt, store, table_name, base_uri, materialize):
has_output = bool(output_prop_uri)
if row_logic == "and" or not has_output:
self._execute_combined(
dt,
store,
table_name,
base_uri,
result,
dt_name,
output_prop_uri,
output_prop_name,
rows,
output_default_val,
dt, store, table_name, base_uri, result, dt_name,
output_prop_uri, output_prop_name, rows, output_default_val,
dialect,
)
else:
self._execute_per_row(
dt,
store,
table_name,
base_uri,
result,
dt_name,
output_prop_uri,
output_prop_name,
rows,
hit_policy,
output_default_val,
dt, store, table_name, base_uri, result, dt_name,
output_prop_uri, output_prop_name, rows, hit_policy, output_default_val,
dialect,
)
return result

Expand All @@ -196,9 +197,10 @@ def _execute_combined(
output_prop_name,
rows,
output_default_val="",
dialect="databricks",
):
tbl_ref = store.sql_table_reference(table_name)
query = self.build_violation_sql(dt, tbl_ref, base_uri)
query = self.build_violation_sql(dt, tbl_ref, base_uri, dialect)
if not query:
logger.warning("Decision table '%s': query builder returned None", dt_name)
return
Expand All @@ -212,7 +214,7 @@ def _execute_combined(
for subj in self._run_query(store, query, dt_name):
msg = f"Matches decision table '{dt_name}'"
if action_val and output_prop_name:
msg += f" → {output_prop_name} = {action_val}"
msg += f" -> {output_prop_name} = {action_val}"
result.violations.append(
RuleViolation(
rule_name=dt_name,
Expand Down Expand Up @@ -246,6 +248,7 @@ def _execute_per_row(
rows,
hit_policy,
output_default_val="",
dialect="databricks",
):
seen: set = set()
for ri, row in enumerate(rows):
Expand All @@ -254,7 +257,7 @@ def _execute_per_row(
single_dt["rows"] = [row]
single_dt["row_logic"] = "or"
tbl_ref = store.sql_table_reference(table_name)
query = self.build_violation_sql(single_dt, tbl_ref, base_uri)
query = self.build_violation_sql(single_dt, tbl_ref, base_uri, dialect)
if not query:
continue
logger.debug(
Expand All @@ -266,7 +269,7 @@ def _execute_per_row(
seen.add(subj)
msg = f"Row {ri + 1} of '{dt_name}'"
if action_val and output_prop_name:
msg += f" → {output_prop_name} = {action_val}"
msg += f" -> {output_prop_name} = {action_val}"
result.violations.append(
RuleViolation(
rule_name=dt_name,
Expand Down Expand Up @@ -300,13 +303,14 @@ def _run_query(store, query, dt_name):
logger.error("Decision table query failed for '%s': %s", dt_name, e)
return subjects

def build_violation_sql(self, dt, table, base_uri):
def build_violation_sql(self, dt, table, base_uri, dialect="databricks"):
target_cls_uri = dt.get("target_class_uri", "")
inputs = dt.get("input_columns", [])
rows = dt.get("rows", [])
if not target_cls_uri or not inputs or not rows:
return None
joins = []
joined_aliases = set()
base_where = [
f"t0.predicate = '{RDF_TYPE}'",
f"t0.object = '{self._esc_sql(target_cls_uri)}'",
Expand All @@ -315,7 +319,19 @@ def build_violation_sql(self, dt, table, base_uri):
alias = f"inp{i}"
prop_uri = inp.get("property_uri", "")
if not prop_uri:
# Input column has no property mapped (e.g. left blank in the
# decision table editor) — skip the join. Any row condition
# referencing this column's alias is skipped below too, so
# we never emit a WHERE clause referencing a table that was
# never joined ("missing FROM-clause entry" in Postgres).
logger.warning(
"Decision table input column %d ('%s') has no property_uri — "
"its conditions will be ignored",
i,
inp.get("property", inp.get("label", "")),
)
continue

@benoitcayladbx benoitcayladbx Aug 23, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Skipping an unmapped input is the right fix for missing FROM-clause entry.

Please also fail closed when no input was joined: with joined_aliases empty the query is only rdf:type = target class and every instance matches. Prefer returning None / skipping the table (and keep the warning).

Add a unit test that an unmapped column never appears in FROM/WHERE, and that a table with only unmapped inputs does not match the whole class.

joined_aliases.add(alias)
joins.append(
f"INNER JOIN {table} {alias} ON {alias}.subject = t0.subject "
f"AND {alias}.predicate = '{self._esc_sql(prop_uri)}'"
Expand All @@ -330,13 +346,15 @@ def build_violation_sql(self, dt, table, base_uri):
if op == "any" or not val:
continue
alias = f"inp{j}"
if alias not in joined_aliases:
continue
sql_op = DT_OP_SQL.get(op)
if sql_op is None:
continue
if self._is_numeric(val):
v_expr = val
lhs = (
sql_numeric(f"{alias}.object")
sql_numeric(f"{alias}.object", dialect=dialect)
if op in DT_NUMERIC_OPS
else f"{alias}.object"
)
Expand Down
11 changes: 9 additions & 2 deletions src/back/core/reasoning/SPARQLRuleEngine.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,10 +37,17 @@ def _build_uri_map(ontology: Dict) -> Dict[str, str]:
sep = "" if base_uri.endswith("#") or base_uri.endswith("/") else "#"
data_ns = base_uri.rstrip("#").rstrip("/") + "/" if base_uri else ""

for cls in ontology.get("classes", []):
base_ns = base_uri.rstrip("#").rstrip("/") if base_uri else ""
for cls in ontology.get("classes", []): # (o self._ontology en SWRLEngine)

@benoitcayladbx benoitcayladbx Aug 23, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please drop the leftover # (o self._ontology en SWRLEngine) and the Spanish comment below. Same URI-rewrite helper as the other engines — don’t fork a fourth copy of this loop.

name = cls.get("name", "") or cls.get("localName", "")
uri = cls.get("uri", "")
if not uri and name:
if base_ns and uri and not uri.startswith(base_ns):
# URI obsoleta de un Base URI anterior (p.ej. el dominio se
# re-basó tras crear la clase) — se reconstruye contra el
# base_uri actual, igual que ya se hace con las propiedades.
local = uri_local_name(uri)
uri = base_uri + sep + local
elif not uri and name:
uri = base_uri + sep + name
if name:
uri_map[name.lower()] = uri
Expand Down
Loading