Skip to content

SQL Compiler

Bases: Compiler

Compiler that translates TypedLogic rules plus a goal/query to SQL.

Source code in src/typedlogic/compilers/sql_compiler.py
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
@dataclass
class SQLCompiler(Compiler):
    """Compiler that translates TypedLogic rules plus a goal/query to SQL."""

    default_suffix: ClassVar[str] = "sql"
    config: Optional[SQLCompilerConfig] = None

    def compile(self, theory: Theory, syntax: Optional[Union[str, ModelSyntax]] = None, **kwargs: Any) -> str:
        """
        Compile a theory and optional goal into SQL.

        :param theory: logical theory to translate
        :param syntax: ignored; present for the compiler interface
        :param kwargs: ``goal``, ``bindings``, or any ``SQLCompilerConfig`` field
        :return: SQL text
        """
        goal = kwargs.pop("goal", None)
        config = self._resolve_config(kwargs)
        if goal is not None:
            return self.compile_query(theory, goal, config=config)
        if theory.goals:
            return "\n\n".join(
                self.compile_query(theory, goal_sentence, config=config) + ";" for goal_sentence in theory.goals
            )
        return self.compile_relations(theory, config=config)

    def compile_query(
        self,
        theory: Theory,
        goal: Sentence,
        config: Optional[SQLCompilerConfig] = None,
    ) -> str:
        """Compile a goal/query against a theory to a SQL ``SELECT`` statement."""
        resolved = config or self.config or SQLCompilerConfig()
        horn_rules = horn_rules_for(theory)
        context = build_context(theory, resolved, horn_rules, extra_sentences=[goal])
        ctes = compile_ctes(horn_rules, context)
        query = compile_goal_select(goal, context)
        return attach_with_clause(query, ctes, recursive=uses_recursive_cte(horn_rules, context))

    def compile_constraints(
        self,
        theory: Theory,
        config: Optional[SQLCompilerConfig] = None,
    ) -> list[str]:
        """Compile FOL denial constraints to SQL queries that return violating rows."""
        resolved = config or self.config or SQLCompilerConfig()
        horn_rules = horn_rules_for(theory)
        context = build_context(theory, resolved, horn_rules)
        ctes = compile_ctes(horn_rules, context)
        constraints = [rule for rule in horn_rules if is_constraint_rule(rule)]
        return [
            attach_with_clause(
                compile_constraint_select(rule, context, index), ctes, uses_recursive_cte(horn_rules, context)
            )
            for index, rule in enumerate(constraints, start=1)
        ]

    def compile_relations(
        self,
        theory: Theory,
        config: Optional[SQLCompilerConfig] = None,
    ) -> str:
        """Compile each derived relation into a standalone inspection query."""
        resolved = config or self.config or SQLCompilerConfig()
        horn_rules = horn_rules_for(theory)
        context = build_context(theory, resolved, horn_rules)
        ctes = compile_ctes(horn_rules, context)
        recursive = uses_recursive_cte(horn_rules, context)
        queries = []
        for predicate in context.cte_order:
            select = f"SELECT * FROM {context.ident(predicate)}"  # noqa: S608
            queries.append(attach_with_clause(select, ctes, recursive=recursive) + ";")
        return "\n\n".join(queries)

    def _resolve_config(self, kwargs: Mapping[str, Any]) -> SQLCompilerConfig:
        config = replace(self.config) if self.config else SQLCompilerConfig()
        options = dict(kwargs)
        if "bindings" in options:
            config.bindings = normalize_bindings(options.pop("bindings"))
        for key, value in options.items():
            if not hasattr(config, key):
                raise TypeError(f"Unknown SQL compiler option: {key}")
            setattr(config, key, value)
        return config

compile(theory, syntax=None, **kwargs)

Compile a theory and optional goal into SQL.

Parameters:

Name Type Description Default
theory Theory

logical theory to translate

required
syntax Optional[Union[str, ModelSyntax]]

ignored; present for the compiler interface

None
kwargs Any

goal, bindings, or any SQLCompilerConfig field

{}

Returns:

Type Description
str

SQL text

Source code in src/typedlogic/compilers/sql_compiler.py
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
def compile(self, theory: Theory, syntax: Optional[Union[str, ModelSyntax]] = None, **kwargs: Any) -> str:
    """
    Compile a theory and optional goal into SQL.

    :param theory: logical theory to translate
    :param syntax: ignored; present for the compiler interface
    :param kwargs: ``goal``, ``bindings``, or any ``SQLCompilerConfig`` field
    :return: SQL text
    """
    goal = kwargs.pop("goal", None)
    config = self._resolve_config(kwargs)
    if goal is not None:
        return self.compile_query(theory, goal, config=config)
    if theory.goals:
        return "\n\n".join(
            self.compile_query(theory, goal_sentence, config=config) + ";" for goal_sentence in theory.goals
        )
    return self.compile_relations(theory, config=config)

compile_query(theory, goal, config=None)

Compile a goal/query against a theory to a SQL SELECT statement.

Source code in src/typedlogic/compilers/sql_compiler.py
272
273
274
275
276
277
278
279
280
281
282
283
284
def compile_query(
    self,
    theory: Theory,
    goal: Sentence,
    config: Optional[SQLCompilerConfig] = None,
) -> str:
    """Compile a goal/query against a theory to a SQL ``SELECT`` statement."""
    resolved = config or self.config or SQLCompilerConfig()
    horn_rules = horn_rules_for(theory)
    context = build_context(theory, resolved, horn_rules, extra_sentences=[goal])
    ctes = compile_ctes(horn_rules, context)
    query = compile_goal_select(goal, context)
    return attach_with_clause(query, ctes, recursive=uses_recursive_cte(horn_rules, context))

compile_constraints(theory, config=None)

Compile FOL denial constraints to SQL queries that return violating rows.

Source code in src/typedlogic/compilers/sql_compiler.py
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
def compile_constraints(
    self,
    theory: Theory,
    config: Optional[SQLCompilerConfig] = None,
) -> list[str]:
    """Compile FOL denial constraints to SQL queries that return violating rows."""
    resolved = config or self.config or SQLCompilerConfig()
    horn_rules = horn_rules_for(theory)
    context = build_context(theory, resolved, horn_rules)
    ctes = compile_ctes(horn_rules, context)
    constraints = [rule for rule in horn_rules if is_constraint_rule(rule)]
    return [
        attach_with_clause(
            compile_constraint_select(rule, context, index), ctes, uses_recursive_cte(horn_rules, context)
        )
        for index, rule in enumerate(constraints, start=1)
    ]

compile_relations(theory, config=None)

Compile each derived relation into a standalone inspection query.

Source code in src/typedlogic/compilers/sql_compiler.py
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
def compile_relations(
    self,
    theory: Theory,
    config: Optional[SQLCompilerConfig] = None,
) -> str:
    """Compile each derived relation into a standalone inspection query."""
    resolved = config or self.config or SQLCompilerConfig()
    horn_rules = horn_rules_for(theory)
    context = build_context(theory, resolved, horn_rules)
    ctes = compile_ctes(horn_rules, context)
    recursive = uses_recursive_cte(horn_rules, context)
    queries = []
    for predicate in context.cte_order:
        select = f"SELECT * FROM {context.ident(predicate)}"  # noqa: S608
        queries.append(attach_with_clause(select, ctes, recursive=recursive) + ";")
    return "\n\n".join(queries)

Purpose

The SQL compiler translates a TypedLogic theory plus a query or goal into SQL. It is intended for cases where logical predicates are backed by relational tables and rule expansion should happen inside the database engine.

The design follows the same idea as Christoph Draxler's PL2SQL compiler, which translated a projection term plus a Prolog database goal to SQL. Chris Mungall's later TypedLogic-oriented experiments treated rules as database views so that predicates defined by rules could be rewritten on demand. This compiler uses common table expressions for that view layer:

  • Extensional predicates are backed by configured tables or inline ground facts.
  • Non-recursive Horn rules compile to ordinary CTE branches.
  • Linear self-recursive rules compile to WITH RECURSIVE.
  • Denial constraints compile to SQL queries that return violating rows.

Related modern systems include Logica, which compiles a Datalog-family language to SQL engines; Souffle, which is a modern Datalog engine with database import/export support; and SQL engines that expose recursion through recursive CTEs, such as Materialize. Recent Datalog surveys also emphasize rules as virtual relations or views for modular query pipelines, e.g. Modern Datalog: Concepts, Methods, Applications.

Basic Usage

from typedlogic import And, Forall, Implies, PredicateDefinition, Term, Theory, Variable
from typedlogic.compilers.sql_compiler import PredicateBinding, SQLCompiler

x = Variable("x")
y = Variable("y")
z = Variable("z")

theory = Theory(
    predicate_definitions=[
        PredicateDefinition("Parent", {"parent": "str", "child": "str"}),
        PredicateDefinition("Ancestor", {"ancestor": "str", "descendant": "str"}),
    ]
)
theory.add(Forall([x, y], Implies(Term("Parent", x, y), Term("Ancestor", x, y))))
theory.add(
    Forall(
        [x, y, z],
        Implies(And(Term("Ancestor", x, y), Term("Parent", y, z)), Term("Ancestor", x, z)),
    )
)

sql = SQLCompiler().compile(
    theory,
    goal=Term("Ancestor", "Alice", Variable("who")),
    bindings={
        "Parent": PredicateBinding(
            table="parent_edges",
            columns={"parent": "src", "child": "dst"},
        )
    },
)

The generated SQL uses the table binding for Parent and a recursive CTE for Ancestor:

WITH RECURSIVE
Ancestor(ancestor, descendant) AS (
  SELECT _r0.src AS ancestor, _r0.dst AS descendant FROM parent_edges AS _r0
  UNION
  SELECT _r0.ancestor AS ancestor, _r1.dst AS descendant
  FROM Ancestor AS _r0, parent_edges AS _r1
  WHERE _r0.descendant = _r1.src
)
SELECT DISTINCT _q0.descendant AS who
FROM Ancestor AS _q0
WHERE _q0.ancestor = 'Alice'

Table Bindings

Use PredicateBinding to bind logical predicate arguments to physical tables. If columns is omitted, logical column names are used as physical names. If a mapping is supplied, keys are logical argument names and values are physical SQL columns.

bindings = {
    "Parent": PredicateBinding("parent_edges", {"parent": "src", "child": "dst"}),
    "Person": "people",
}

Ground unit clauses do not need a binding. They compile to inline CTE branches:

theory.add(Term("Person", "Alice"))
theory.add(Term("Person", "Bob"))
WITH
Person(name) AS (
  SELECT 'Alice' AS name UNION SELECT 'Bob' AS name
)
SELECT DISTINCT _q0.name AS x FROM Person AS _q0

Goals And Queries

Pass an explicit goal to compile for goal-directed SQL:

SQLCompiler().compile(theory, goal=Term("Person", Variable("x")))

If no explicit goal is passed, the compiler uses sentences in goal sentence groups. Variables in the goal become projected SQL columns. Ground goals compile to a boolean holds column using EXISTS.

Constraints

First-order denial constraints compile to validation queries. A returned row is a witness that violates the constraint.

theory.add(Forall([x], Not(And(Term("Cat", x), Term("Dog", x)))))

queries = SQLCompiler().compile_constraints(
    theory,
    config=SQLCompilerConfig(
        bindings={"Cat": PredicateBinding("cats"), "Dog": PredicateBinding("dogs")}
    ),
)
SELECT DISTINCT _c1_0.name AS x
FROM cats AS _c1_0, dogs AS _c1_1
WHERE _c1_0.name = _c1_1.name

For portable SQL this is emitted as a violation query rather than a native CHECK, because cross-table assertions and recursive validations are not portable across SQL engines. Databases with assertion, trigger, or materialized view support can wrap these queries in engine-specific enforcement.

Supported Profile

The compiler targets a database-friendly fragment:

  • Ground terms.
  • Conjunctive queries and rule bodies.
  • Horn rules generated from TypedLogic sentences.
  • Positive relational atoms.
  • Negation as correlated NOT EXISTS.
  • Builtin binary comparisons: eq, ne, lt, le, gt, ge.
  • Binary arithmetic expressions in rule heads or comparisons: add, sub, mul, truediv, mod.
  • Linear self-recursive rules when use_recursive_cte=True.

The compiler intentionally raises SQLTranslationError for constructs that do not have a portable relational translation, such as mutually recursive CTE groups, unbound projected variables, non-conjunctive query bodies after Horn normalization, and nested relation-valued function terms.