From ceae48e4e02100b396532b3ebcbb999adb01062f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jerem=C3=ADas=20P=C3=A9rez=20Fern=C3=A1ndez?= Date: Sat, 22 Aug 2026 05:57:16 +0000 Subject: [PATCH] fix(reasoning): support Lakebase/Postgres dialect and rebuild class URIs against the current Base URI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SQLHelpers.to_number() gains a dialect parameter: on Lakebase (Postgres) it emits a CASE/regex guard instead of TRY_CAST, which does not exist in PostgreSQL. AggregateRuleEngine and DecisionTableEngine now detect the triple-store backend and pass the right dialect through, so numeric comparisons in Aggregate Rules and Decision Tables work against both Databricks SQL and Lakebase. AggregateRuleEngine, DecisionTableEngine, SPARQLRuleEngine and SWRLEngine all reconstruct a rule's target class URI from its local name against the domain's *current* Base URI instead of trusting the class's stored uri field, which goes stale after the domain is re-based (Base URI changed after the class already existed). Without this, a rule silently stopped finding its target class and returned zero results/inferred triples. AggregateRuleEngine also now computes inferred triples whenever a rule has a Result Entity configured, not only when materialize=True — previously, clicking "Run" (preview, no materialize) on such a rule always showed 0 inferred triples even when matching rows existed. DecisionTableEngine additionally ignores an input column left without an associated property (blank mapping) instead of generating a SQL condition that references a table never joined, which raised a "missing FROM-clause entry" error. --- src/back/core/helpers/SQLHelpers.py | 20 ++++- .../core/reasoning/AggregateRuleEngine.py | 45 ++++++++--- .../core/reasoning/DecisionTableEngine.py | 74 ++++++++++++------- src/back/core/reasoning/SPARQLRuleEngine.py | 11 ++- src/back/core/reasoning/SWRLEngine.py | 15 +++- 5 files changed, 119 insertions(+), 46 deletions(-) diff --git a/src/back/core/helpers/SQLHelpers.py b/src/back/core/helpers/SQLHelpers.py index 6076bcda..b680a6fa 100644 --- a/src/back/core/helpers/SQLHelpers.py +++ b/src/back/core/helpers/SQLHelpers.py @@ -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`` @@ -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*$' " + 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: diff --git a/src/back/core/reasoning/AggregateRuleEngine.py b/src/back/core/reasoning/AggregateRuleEngine.py index 9658f515..c7c652a1 100644 --- a/src/back/core/reasoning/AggregateRuleEngine.py +++ b/src/back/core/reasoning/AggregateRuleEngine.py @@ -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__) @@ -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): + 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 @@ -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" + def execute_rules( self, rules: List[Dict], @@ -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): @@ -103,6 +116,7 @@ def execute_rules( table_name, base_uri, materialize, + dialect, ) result.merge(rule_result) except Exception as e: @@ -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 @@ -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, @@ -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", "") @@ -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)" ) diff --git a/src/back/core/reasoning/DecisionTableEngine.py b/src/back/core/reasoning/DecisionTableEngine.py index 84cbdada..7d77dc1e 100644 --- a/src/back/core/reasoning/DecisionTableEngine.py +++ b/src/back/core/reasoning/DecisionTableEngine.py @@ -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" + @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): + # 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", "") @@ -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, @@ -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 @@ -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 @@ -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, @@ -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): @@ -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( @@ -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, @@ -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)}'", @@ -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 + 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)}'" @@ -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" ) diff --git a/src/back/core/reasoning/SPARQLRuleEngine.py b/src/back/core/reasoning/SPARQLRuleEngine.py index 8f167b21..bc88f20b 100644 --- a/src/back/core/reasoning/SPARQLRuleEngine.py +++ b/src/back/core/reasoning/SPARQLRuleEngine.py @@ -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) 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 diff --git a/src/back/core/reasoning/SWRLEngine.py b/src/back/core/reasoning/SWRLEngine.py index 51fcad6e..171b70c4 100644 --- a/src/back/core/reasoning/SWRLEngine.py +++ b/src/back/core/reasoning/SWRLEngine.py @@ -195,8 +195,10 @@ def _build_uri_map(self) -> Dict[str, str]: Property URIs are normalised to the **data namespace** (``base_uri`` with a trailing ``/``) so they match the predicates written by the R2RML generator when syncing data to the triple store. Class URIs - keep their original ``#`` separator because ``rdf:type`` objects in - the store use the ontology class URI as-is. + are normalised against the current ``base_uri`` too — if a class's + stored ``uri`` was minted under a previous Base URI (e.g. the domain + was re-based after the class was created), it is rebuilt against the + current ``base_uri``, mirroring what already happens for properties. """ uri_map: Dict[str, str] = {} base_uri = self._ontology.get("base_uri", "") @@ -204,10 +206,17 @@ def _build_uri_map(self) -> Dict[str, str]: data_ns = base_uri.rstrip("#").rstrip("/") + "/" if base_uri else "" + base_ns = base_uri.rstrip("#").rstrip("/") if base_uri else "" for cls in self._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): + # 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