Row-Level Security (forced column filter)¶

TL;DR: Hand SlayerQueryEngine (or the local-mode SlayerClient) a SessionPolicy, and every query it runs is silently scoped to one tenant — base tables, joins, sub-queries, and sample data included. The agent issuing queries cannot read, override, or disable the policy.

This notebook tells a franchise story on the Jaffle Shop data: a session that belongs to one store's operator must only ever see that store's data. orders carries a store_id; the forced column filter scopes every table that has store_id to a single value. Shared tables (the product catalog, the customer list) have no store_id and are handled by on_unapplicable.

The data, unscoped¶

With no policy, the engine sees every store. Let's list orders per store and pick the busiest one to scope our session to.

In [2]:

Copied!





by_store = engine.execute_sync(
    query={
        "source_model": "orders",
        "measures": ["*:count"],
        "dimensions": ["stores.name", "store_id"],
        "order": [{"column": "_count", "direction": "desc"}],
    }
)

print(f"{'Store':<16}{'Orders':>10}")
print("-" * 26)
for row in by_store.data:
    print(f"{row['orders.stores.name']:<16}{row['orders._count']:>10,}")

# Scope this session to the busiest store (a franchise operator's view).
busiest = max(by_store.data, key=lambda r: r["orders._count"])
STORE_ID = busiest["orders.store_id"]
STORE_NAME = busiest["orders.stores.name"]
TOTAL_ORDERS = sum(r["orders._count"] for r in by_store.data)
print(f"\nScoping the session to: {STORE_NAME}  ({STORE_ID})")
by_store = engine.execute_sync(
    query={
        "source_model": "orders",
        "measures": ["*:count"],
        "dimensions": ["stores.name", "store_id"],
        "order": [{"column": "_count", "direction": "desc"}],
    }
)

print(f"{'Store':<16}{'Orders':>10}")
print("-" * 26)
for row in by_store.data:
    print(f"{row['orders.stores.name']:<16}{row['orders._count']:>10,}")

# Scope this session to the busiest store (a franchise operator's view).
busiest = max(by_store.data, key=lambda r: r["orders._count"])
STORE_ID = busiest["orders.store_id"]
STORE_NAME = busiest["orders.stores.name"]
TOTAL_ORDERS = sum(r["orders._count"] for r in by_store.data)
print(f"\nScoping the session to: {STORE_NAME}  ({STORE_ID})")

Store               Orders
--------------------------
Brooklyn           248,271
Philadelphia       191,573
Chicago            104,276
San Francisco       92,842
New Orleans         16,023

Scoping the session to: Brooklyn  (3d1b233f-5b8b-49d1-a175-adc48d7d3938)

Configure the policy¶

A SessionPolicy holds one or more ColumnFilterRules. The rule below means "every table that has a store_id column is filtered to this store." The value is a scalar, so the operator is = (a list would mean IN (...)).

on_unapplicable="pass" says: a table that lacks store_id (the shared product catalog, the customer list) is left unfiltered rather than blocked — those tables aren't store-specific. We'll see the stricter "block" mode further down.

The policy is set once, at engine construction. There is no query field that can change it.

In [3]:

Copied!





from slayer.core.policy import SessionPolicy, ColumnFilterRule
from slayer.engine.query_engine import SlayerQueryEngine

policy = SessionPolicy(
    data_filters=[
        ColumnFilterRule(column="store_id", value=STORE_ID, on_unapplicable="pass"),
    ]
)
store_engine = SlayerQueryEngine(storage=storage, policy=policy)
print("Scoped engine ready.")
from slayer.core.policy import SessionPolicy, ColumnFilterRule
from slayer.engine.query_engine import SlayerQueryEngine

policy = SessionPolicy(
    data_filters=[
        ColumnFilterRule(column="store_id", value=STORE_ID, on_unapplicable="pass"),
    ]
)
store_engine = SlayerQueryEngine(storage=storage, policy=policy)
print("Scoped engine ready.")

Scoped engine ready.

Same query, now tenant-scoped¶

The query is identical to one you'd run unscoped — no store_id filter, no model changes. The engine injects the scoping for us.

In [4]:

Copied!





all_orders = engine.execute_sync(
    query={"source_model": "orders", "measures": ["*:count"]}
)
scoped_orders = store_engine.execute_sync(
    query={"source_model": "orders", "measures": ["*:count"]}
)

print(f"All stores:        {all_orders.data[0]['orders._count']:>10,} orders")
print(f"{STORE_NAME} only: {scoped_orders.data[0]['orders._count']:>10,} orders")
all_orders = engine.execute_sync(
    query={"source_model": "orders", "measures": ["*:count"]}
)
scoped_orders = store_engine.execute_sync(
    query={"source_model": "orders", "measures": ["*:count"]}
)

print(f"All stores:        {all_orders.data[0]['orders._count']:>10,} orders")
print(f"{STORE_NAME} only: {scoped_orders.data[0]['orders._count']:>10,} orders")

All stores:           652,985 orders
Brooklyn only:    248,271 orders

Preview the rewritten SQL¶

dry_run=True returns exactly the SQL that would execute — including the per-table wrap. Each physical orders reference becomes a filtered sub-query, with the original alias preserved. The literal is bound, so it's injection-safe.

In [5]:

Copied!





preview = store_engine.execute_sync(
    query={"source_model": "orders", "measures": ["order_total:sum"]},
    dry_run=True,
)
print(preview.sql)
preview = store_engine.execute_sync(
    query={"source_model": "orders", "measures": ["order_total:sum"]},
    dry_run=True,
)
print(preview.sql)

SELECT SUM(orders.order_total) AS "orders.order_total_sum" FROM (SELECT * FROM orders WHERE store_id = '3d1b233f-5b8b-49d1-a175-adc48d7d3938') AS orders

Joins stay scoped on every side¶

Cross-tenant leakage usually hides in joins. Here we rank customers by revenue — a query that joins orders to customers. Under the policy, the orders side is wrapped (it has store_id); customers is a shared table (no store_id, so it passes through). The result only ever reflects this store's orders.

In [6]:

Copied!





top_customers = store_engine.execute_sync(
    query={
        "source_model": "orders",
        "measures": ["order_total:sum"],
        "dimensions": ["customers.name"],
        "order": [{"column": "order_total_sum", "direction": "desc"}],
        "limit": 5,
    }
)

print(f"Top customers at {STORE_NAME} (this store's revenue only):")
for row in top_customers.data:
    print(f"  {row['orders.customers.name']:<22} ${row['orders.order_total_sum']:>9,.2f}")
top_customers = store_engine.execute_sync(
    query={
        "source_model": "orders",
        "measures": ["order_total:sum"],
        "dimensions": ["customers.name"],
        "order": [{"column": "order_total_sum", "direction": "desc"}],
        "limit": 5,
    }
)

print(f"Top customers at {STORE_NAME} (this store's revenue only):")
for row in top_customers.data:
    print(f"  {row['orders.customers.name']:<22} ${row['orders.order_total_sum']:>9,.2f}")

Top customers at Brooklyn (this store's revenue only):
  Amy Simmons            $11,380.72
  Richard Jones          $11,294.40
  John Wilson            $11,206.00
  Matthew Carrillo       $10,962.64
  Michael Davis          $10,936.57

`pass` vs `block`: the tables without the column¶

store_id lives only on orders. What happens when a query touches a table that doesn't have it — like the shared products catalog?

With on_unapplicable="pass" (our policy), that table flows through unfiltered — products aren't store-specific, so this is correct.

In [7]:

Copied!





products = store_engine.execute_sync(
    query={"source_model": "products", "measures": ["*:count"]}
)
print(f"Products (shared catalog, passed through): {products.data[0]['products._count']}")
products = store_engine.execute_sync(
    query={"source_model": "products", "measures": ["*:count"]}
)
print(f"Products (shared catalog, passed through): {products.data[0]['products._count']}")

Products (shared catalog, passed through): 10

With the default on_unapplicable="block", a table that confirms it lacks the tenant column fails the whole query. Use this when every table is expected to carry the tenant column (true RLS-on-every-table), so a missing column surfaces as an error instead of a silent leak.

In [8]:

Copied!





from slayer.core.errors import ForcedFilterError

strict_engine = SlayerQueryEngine(
    storage=storage,
    policy=SessionPolicy(
        data_filters=[ColumnFilterRule(column="store_id", value=STORE_ID)]  # block (default)
    ),
)

try:
    strict_engine.execute_sync(
        query={"source_model": "products", "measures": ["*:count"]}
    )
except ForcedFilterError as exc:
    print(f"Blocked: {exc}")
    print(f"  table={exc.table!r}  column={exc.column!r}")
from slayer.core.errors import ForcedFilterError

strict_engine = SlayerQueryEngine(
    storage=storage,
    policy=SessionPolicy(
        data_filters=[ColumnFilterRule(column="store_id", value=STORE_ID)]  # block (default)
    ),
)

try:
    strict_engine.execute_sync(
        query={"source_model": "products", "measures": ["*:count"]}
    )
except ForcedFilterError as exc:
    print(f"Blocked: {exc}")
    print(f"  table={exc.table!r}  column={exc.column!r}")

Blocked: Forced filter rule on column 'store_id' requires column 'store_id' on table 'products', which does not have it.
  table='products'  column='store_id'

A table whose column presence cannot be confirmed at all (an introspection error) always fails closed, regardless of on_unapplicable — SLayer never emits an unscoped query against a table it couldn't verify.

Immutable and agent-invisible¶

The policy is frozen engine state. An agent can't widen its own scope by mutating the policy or by adding a contradictory filter — the wrap is applied to the final SQL after the agent's query has been generated.

In [9]:

Copied!





# The policy object itself is immutable.
try:
    store_engine.policy.data_filters = ()
except Exception as exc:
    print(f"Mutating the policy is rejected: {type(exc).__name__}")

# Even if the agent asks for another store explicitly, it still only sees
# this store's rows — the forced filter is ANDed onto the final SQL.
other = next(
    (r["orders.store_id"] for r in by_store.data if r["orders.store_id"] != STORE_ID),
    None,
)
if other:
    sneaky = store_engine.execute_sync(
        query={
            "source_model": "orders",
            "measures": ["*:count"],
            "filters": [f"store_id = '{other}'"],
        }
    )
    print(f"Agent asked for a different store -> rows returned: {sneaky.data[0]['orders._count']}")
# The policy object itself is immutable.
try:
    store_engine.policy.data_filters = ()
except Exception as exc:
    print(f"Mutating the policy is rejected: {type(exc).__name__}")

# Even if the agent asks for another store explicitly, it still only sees
# this store's rows — the forced filter is ANDed onto the final SQL.
other = next(
    (r["orders.store_id"] for r in by_store.data if r["orders.store_id"] != STORE_ID),
    None,
)
if other:
    sneaky = store_engine.execute_sync(
        query={
            "source_model": "orders",
            "measures": ["*:count"],
            "filters": [f"store_id = '{other}'"],
        }
    )
    print(f"Agent asked for a different store -> rows returned: {sneaky.data[0]['orders._count']}")

Mutating the policy is rejected: ValidationError
Agent asked for a different store -> rows returned: 0

Summary¶

Aspect	Behaviour
Where	`SlayerQueryEngine(storage, policy=...)` / `SlayerClient(storage=..., policy=...)`
What	Wraps every physical table that has the rule's column in a filtered sub-query
Operator	scalar value -> `=`, list value -> `IN (...)`
`on_unapplicable="pass"`	tables lacking the column are left unfiltered (shared data)
`on_unapplicable="block"`	tables lacking the column fail the query (every table must carry it)
Unconfirmable table	always fails closed (security control)
Scope	base, joins, CTEs, sql-mode, query-backed stages, profiling/sample data
Mutability	immutable; set only at engine/client init; the agent can't read or change it

Forced filters are the tightest useful slice of tenant isolation. Join-path rules, per-model scoping, and server-side (REST/MCP) policy are future additions — see Row-Level Security.