A composable threat detection platform on ibis
This work is part of a larger detection validation effort by the R&D and
detection engineering teams at the DNB Cyber Defense Center.
This post is about the shape of detection engineering when telemetry lives in
open table formats (Delta, Iceberg) and gets queried by many engines: Spark,
Snowflake, ClickHouse, DuckDB, plus SIEMs like Splunk and Defender XDR.
The implementation behind it sits in a prototype called detection-ibis.
Numbers, rule names, and source tables are illustrative.
A detection rule is a boolean expression over a table. We have spent two decades writing five copies of the same rule - one per query language. The data layer has been consolidating on open formats (Delta and Iceberg) for years. The query layer is finally catching up: one expression, compiled to whichever engine holds the data. Spark or Flink for streaming over the lakehouse. DuckDB locally. Snowflake or ClickHouse for batch. Splunk and Defender XDR through thin bridges. The point of this post is what detection engineering looks like when you stop maintaining five copies.
The walkthrough below works bottom-up. One atom: a single query against one engine, expressed as ibis. Composition: many atoms running over the same source, fused into one scan. Signals: detection output as a queryable table, not just an alert stream. Whitelisting: another expression next to the rule, not an external config system. Streaming vs batch: the same expression, two compile paths. SIEMs: a bridge pattern that lands Splunk SPL and Defender KQL results as ibis tables, so they compose with everything else. The complete prototype is embedded throughout as runnable notebook cells.
Detection engineering today
Modern detection engineering runs against a genuinely fragmented data landscape. Endpoint telemetry flows through Microsoft Defender for Endpoint and surfaces in Splunk via SPL or in Defender XDR via KQL. Identity signals live in Azure AD and reach the SIEM through a different pipeline than the endpoint events. Vulnerability data lands in Snowflake and is queried with SQL. Network sensors write to ClickHouse for low-latency analytical queries. Cloud audit events route to BigQuery or Trino. Real-time detection work runs in Apache Flink or RisingWave against a streaming source. Each data source implies an engine, and each engine implies a query language.
The storage layer has started to converge. Microsoft has been moving endpoint telemetry into Delta Lake as part of the Fabric and OneLake push. Databricks and the Apache Iceberg ecosystem have done the same for warehouse-scale data. Both Delta and Iceberg are now readable from DuckDB locally via delta_scan and iceberg_scan. The bytes are converging on two open formats. The engines that read them are not.
The query layer has not caught up. A detection engineer who needs the same rule to cover endpoint events in Splunk and network events in ClickHouse writes it twice, in two dialects, in two places in version control. A rule that needs to run over an Iceberg snapshot in DuckDB locally and a Delta stream in Spark Structured Streaming in production is another rewrite. Multiply that by a catalogue of hundreds of rules and add the fact that each dialect drifts independently when the underlying telemetry changes: a field rename in the MDE schema means a PR in Splunk SPL, another PR in KQL for Sentinel, and potentially a third in whatever warehouse query is cross-referencing the same events. The rule is the same in intent. The implementations diverge.
Five dialects, one ideaThe diagram below makes the drift surface concrete. A single detection concept - osascript spawning a shell that pipes payload into Python, MITRE T1204 - needs a separate implementation for every engine that holds relevant telemetry. Each implementation is a living artifact that can diverge from the others whenever a parser changes, a field is renamed, or a colleague who knows only one dialect tunes the rule they understand and leaves the others untouched.
Fig 1: One detection concept fanning out into five separate implementations, one per query language, each landing on a different engine and data source. The rule is logically identical across all five. In practice the implementations diverge over time: a tuning change in one dialect rarely propagates to the others, and a field rename in the upstream source is a five-PR problem. Hover a language node for a short note on what engine it targets and where the drift is most likely to surface.
The per-dialect cost is not just the initial translation. It is the ongoing maintenance burden of five things that are supposed to stay equivalent. When the MDE team renames properties.InitiatingProcessFileName to something else in their Iceberg export schema, the Splunk rule, the KQL rule, and the Snowflake rule each need a separate PR, merged by whoever notices. When a detection engineer tunes the SPL version to exclude a noisy process, that exclusion does not automatically appear in the ClickHouse version. The implementations diverge silently, over time, and the divergence shows up as coverage gaps rather than errors: the rule still runs, it just does not match what it used to.
This is not a new problem. Sigma has been addressing it for years via a one-rule, many-translations model: write a rule in a common YAML schema, compile to the target SIEM. Sigma works well for SIEM-to-SIEM portability. It is less suited to the heterogeneous data lake case, where the rule needs to run as a streaming aggregation over Iceberg in Flink, a batch query in Snowflake, and a local test against a DuckDB snapshot - all from the same definition. The compilation target is not just a query dialect. It is a whole runtime with different execution semantics.
The next section describes a different angle: instead of translating a rule into many dialects, use a substrate that compiles one expression to many runtimes natively. The rule stays one thing. The engines are many. The storage format is open.
The one-line idea
A detection rule is one boolean expression over a table. ibis compiles one expression to many engines. Therefore: write the rule once.
That claim is worth unpacking because it holds some real weight. ibis is a Python dataframe API. You build an expression tree against a table you got from a connect() call. The expression tree lowers to whichever backend that connection targets: DuckDB, PySpark, Snowflake, ClickHouse, BigQuery, Flink, RisingWave, Trino, and a growing long tail. The rule author writes predicates, not query strings. The choice of engine is a runtime decision, not a code decision.
The same predicate - say, (t.InitiatingProcessFileName == "osascript") & t.FileName.lower().isin(["sh", "bash"]) & t.ProcessCommandLine.lower().contains("python &") - compiles to a DuckDB SQL query when the connection is DuckDB, to a PySpark DataFrame plan when the connection is Spark, and to a ClickHouse query when the connection is ClickHouse. The rule author writes it once. The engine is swapped at the connection layer, not in the rule body.
Two carve-outs are honest. Streaming primitives (tumbling windows with watermarks) exist on Spark Structured Streaming, Flink, and RisingWave but not on batch backends like DuckDB or Snowflake. The framework handles the dispatch inside a tumble_window_agg(...) helper, covered in the batch and streaming section below. SIEMs that don’t speak SQL - Splunk’s SPL, Defender XDR’s KQL - need a thin bridge that runs the native query and lands the result as an ibis table, the same bridge pattern used in the incident response notebooks post. Both cases compose with the same rule logic on top.
Fig 2: The consolidated shape. One detection rule flows through a single ibis expression tree and fans out to many backend engines. Below the engines, two storage formats - Delta Lake and Apache Iceberg - supply the telemetry. The detection engineer writes the predicate once. Choosing the connection is the only engine-specific decision.
The mechanism is worth being precise about, because the claim “write the rule once” has been made many times for things that turned out to mean “write a subset of SQL that a translator converts.” ibis is not a translator. You write Python. ibis builds an abstract expression tree from method calls on a lazy table object. When you call .execute(), or pass the expression to ibis.to_sql(), the tree lowers to whichever query language the backend requires. DuckDB gets SQL. Spark gets a DataFrame plan or SQL. ClickHouse gets its dialect. The rule author never touches those dialects.
The concrete implication for detection engineering is that the same rule file works in two completely different development contexts. Locally, you run it against a DuckDB in-process connection pointed at a snapshot of production telemetry read via iceberg_scan() or delta_scan(). In production, you swap the connection to Spark (pointing at the live Delta table) or Snowflake (pointing at your warehouse tables). The predicate is unchanged. The test you ran locally proved the same predicate the production job will run. There is no translation layer that could introduce a difference between the tested version and the deployed version.
The practical design decision this creates is that rule files and connection choices are separate concerns. A rule file is a Python function. A deployment configuration declares which connection that rule runs against: a DuckDB handle for local dev and CI, a Spark session for the Databricks lakehouse job, a Snowflake connection for the warehouse tier. The same rule can be registered against multiple connections in the same run, which is how cross-backend regression works: you assert that the DuckDB output and the PySpark output are equal against the same fixture events.
One thing the diagram deliberately omits: rules that run against Splunk or Defender XDR data don’t go through ibis’s compile path. Those SIEMs don’t have ibis backends. Instead, a thin bridge runs the native query (SPL or KQL), lands the result as a polars DataFrame, and registers it in DuckDB as an ibis Table. From that point on, the rule logic on top is the same ibis expression you’d write against a Delta table. The SIEM becomes another data source, and the rule becomes backend-agnostic again. This is covered in full in the SIEMs section below.
The atom: one rule, one engine
A detection rule is a function from a table to a boolean column. Everything else - metadata, registration, compilation - lives outside the predicate. The smallest useful thing in the framework is one such function, bound to one ibis connection, run against one event table.
Defining the ruleThe rule below detects osascript spawning a shell that pipes into a Python process - a macOS execution chain an attacker uses to run a downloaded payload without touching disk. Three predicates: parent process is osascript, spawned file is a shell, and the command line contains python &. All three have to hold.
Here is what the decorator form looks like with the hover annotations that unpack each decision:
from detectionkit import KillChain, MitreAttackTechnique, Status from detectionkit import stateless @stateless( number=373, name="AppleScriptToPython", source_table="cdc.DeviceProcessEvents", mitre=[MitreAttackTechnique.T1204], kill_chain=[KillChain.EXECUTION], status=Status.RELEASE, ) def apple_script_to_python(t): """osascript spawning a shell that threads a Python process.""" return ( (t.InitiatingProcessFileName == "osascript") & t.FileName.lower().isin(["sh", "bash"]) & t.ProcessCommandLine.lower().contains("python &") )
The complete detection in decorator form. Hover the numbered markers to see what each piece does and how it lowers across backends. The predicate is the four lines inside the return block. Everything above is metadata that the decorator consumes at import time. The alert_name is derived as CDC_373_DBX_AppleScriptToPython, so the number, tool token, and function name cannot get out of sync.
The predicate is just an ibis expression. To run it, you open a connection, register an event table against it, and call filter. The cell below does exactly that against a small synthetic event table - five rows with one attacker row that should match and four benign rows that should not. Edit the predicate and re-run. The output updates immediately.
r0 := InMemoryTable
data:
PandasDataFrameProxy:
Timestamp DeviceId DeviceName InitiatingProcessFileName \
0 2026-06-18 14:22:00 dev-001 WS-ENG-01 osascript
1 2026-06-18 14:23:00 dev-001 WS-ENG-01 osascript
2 2026-06-18 14:24:00 dev-002 WS-ENG-02 explorer.exe
3 2026-06-18 14:25:00 dev-003 WS-ENG-03 powershell.exe
4 2026-06-18 14:26:00 dev-lab-99 lab-test-01 osascript
.. ... ... ... ...
15 2026-06-18 14:00:00 dev-004 WS-OPS-01
16 2026-06-18 14:30:00 dev-004 WS-OPS-01
17 2026-06-18 15:00:00 dev-004 WS-OPS-01
18 2026-06-18 15:30:00 dev-004 WS-OPS-01
19 2026-06-18 16:00:00 dev-004 WS-OPS-01
FileName ProcessCommandLine RemoteIP \
0 sh sh -c 'echo abc | base64 -d | python &'
1 sh sh -c 'open /Applications/Safari.app'
2 cmd.exe cmd.exe /c dir
3 procdump.exe procdump.exe -ma lsass.exe out.dmp
4 sh sh -c 'echo xyz | python &'
.. ... ... ...
15 10.0.5.20
16 10.0.5.21
17 10.0.5.22
18 10.0.5.23
19 10.0.5.24
RemotePort ActionType ReportId
0 0 ProcessCreated 1001
1 0 ProcessCreated 1002
2 0 ProcessCreated 1003
3 0 ProcessCreated 1004
4 0 ProcessCreated 1005
.. ... ... ...
15 3389 ConnectionRequest 2010
16 3389 ConnectionRequest 2011
17 3389 ConnectionRequest 2012
18 3389 ConnectionRequest 2013
19 3389 ConnectionRequest 2014
[20 rows x 10 columns]
r1 := Filter[r0]
r0.InitiatingProcessFileName == 'osascript'
InValues(value=Lowercase(r0.FileName), options=['sh', 'bash'])
StringContains(haystack=Lowercase(r0.ProcessCommandLine), needle='python &')
Project[r1]
DeviceName: r1.DeviceName
FileName: r1.FileName
ProcessCommandLine: r1.ProcessCommandLineSELECT
"t0"."DeviceName",
"t0"."FileName",
"t0"."ProcessCommandLine"
FROM "ibis_pandas_memtable_fwbzej7h3rc4tcwr4dzokn6q6q" AS "t0"
WHERE
"t0"."InitiatingProcessFileName" = 'osascript'
AND LOWER("t0"."FileName") IN ('sh', 'bash')
AND CONTAINS(LOWER("t0"."ProcessCommandLine"), 'python &')
The rule running against synthetic events in DuckDB. The connection is in-process. There is no server, no cluster, and no warm-up time. A detection engineer can iterate on a predicate in under a second per cycle, which is the inner loop that makes rule-writing tractable. The output table shows only the matched row, with all event fields available for inspection.
The predicate is an ibis expression tree, not a string. That means ibis can compile it to any backend dialect it knows about. The cell below shows the same apple_script_to_python expression compiled to the DuckDB SQL dialect and to the PySpark dialect side by side. The predicate is unchanged - only the compile target differs. Swapping a production job from DuckDB to PySpark is a one-line change to the connection call, not a rewrite of any rule.
The same predicate compiled to two dialects. The DuckDB output is standard SQL with LOWER() and LIKE. The PySpark output is a DataFrame expression with lower() and contains(). Both express the same predicate, and both were generated from the same Python function without any dialect-specific code in the rule body. In production, the connection object is the only thing that changes between local DuckDB testing and a Spark Structured Streaming job.
For engines that do not speak SQL - Splunk’s SPL and Defender XDR’s KQL among them - a thin bridge decorator runs the native query, lands the result as a DuckDB-registered ibis table, and from that point on the rest of the framework treats it as any other source. That bridge pattern is covered in the SIEMs and other non-SQL engines section.
Composing atoms: many rules, one scan
A single rule is the atom. In practice a detection platform carries hundreds of them, covering process injection, persistence, lateral movement, credential theft, and dozens of technique sub-categories. Running one query per rule means one full table scan per rule - and at hundreds of rules over a source table with millions of events per day, that cost compounds quickly. The standard approach, sometimes called “fusion,” evaluates all rules in a single scan by building an array<struct> column of per-rule results and filtering it down to what actually fired.
The fused pipeline has five stages. Each stage transforms the table before handing it to the next:
- Mutate. For each event row, evaluate every rule predicate and wrap the result in a struct with a fixed shape:
alert_name,triggered(boolean), and any rule-specific metadata fields. Collect these structs into an array column calledalerts. - Array filter. Keep only the array elements where
triggeredis true, dropping the non-firing rules from each row. - Row filter. Drop rows whose
alertsarray is now empty - events that matched nothing. - Unnest. Expand the filtered array so each
(event, triggered rule)pair becomes its own row. At this point the table has one row per signal. - Project. Select the final output columns:
timestamp,origin_event_id,alert_name, and the alert metadata. This is the signals table.
The ibis expression for each stage maps cleanly to a single mutate, filter, or unnest call. Because ibis compiles the whole expression tree to the backend, the five stages lower to a single SQL query or a single Spark job, with no intermediate materializations.
The cell below shows the full fused pipeline over three example rules on synthetic process events.
r0 := InMemoryTable
data:
PandasDataFrameProxy:
Timestamp DeviceId DeviceName InitiatingProcessFileName \
0 2026-06-18 14:22:00 dev-001 WS-ENG-01 osascript
1 2026-06-18 14:23:00 dev-001 WS-ENG-01 osascript
2 2026-06-18 14:24:00 dev-002 WS-ENG-02 explorer.exe
3 2026-06-18 14:25:00 dev-003 WS-ENG-03 powershell.exe
4 2026-06-18 14:26:00 dev-lab-99 lab-test-01 osascript
.. ... ... ... ...
15 2026-06-18 14:00:00 dev-004 WS-OPS-01
16 2026-06-18 14:30:00 dev-004 WS-OPS-01
17 2026-06-18 15:00:00 dev-004 WS-OPS-01
18 2026-06-18 15:30:00 dev-004 WS-OPS-01
19 2026-06-18 16:00:00 dev-004 WS-OPS-01
FileName ProcessCommandLine RemoteIP \
0 sh sh -c 'echo abc | base64 -d | python &'
1 sh sh -c 'open /Applications/Safari.app'
2 cmd.exe cmd.exe /c dir
3 procdump.exe procdump.exe -ma lsass.exe out.dmp
4 sh sh -c 'echo xyz | python &'
.. ... ... ...
15 10.0.5.20
16 10.0.5.21
17 10.0.5.22
18 10.0.5.23
19 10.0.5.24
RemotePort ActionType ReportId
0 0 ProcessCreated 1001
1 0 ProcessCreated 1002
2 0 ProcessCreated 1003
3 0 ProcessCreated 1004
4 0 ProcessCreated 1005
.. ... ... ...
15 3389 ConnectionRequest 2010
16 3389 ConnectionRequest 2011
17 3389 ConnectionRequest 2012
18 3389 ConnectionRequest 2013
19 3389 ConnectionRequest 2014
[20 rows x 10 columns]
r1 := Project[r0]
Timestamp: r0.Timestamp
DeviceId: r0.DeviceId
DeviceName: r0.DeviceName
InitiatingProcessFileName: r0.InitiatingProcessFileName
FileName: r0.FileName
ProcessCommandLine: r0.ProcessCommandLine
RemoteIP: r0.RemoteIP
RemotePort: r0.RemotePort
ActionType: r0.ActionType
ReportId: r0.ReportId
_alerts: Array([StructColumn(names=['alert_name', 'alert_triggered'], values=['CDC_373_AppleScriptToPython', r0.InitiatingProcessFileName == 'osascript' & InValues(value=Lowercase(r0.FileName), options=['sh', 'bash']) & StringContains(haystack=Lowercase(r0.ProcessCommandLine), needle='python &')]), StructColumn(names=['alert_name', 'alert_triggered'], values=['CDC_011_ProcDumpLsass', r0.FileName == 'procdump.exe' & StringContains(haystack=Lowercase(r0.ProcessCommandLine), needle='lsass.exe')])])
r2 := Project[r1]
Timestamp: r1.Timestamp
DeviceId: r1.DeviceId
DeviceName: r1.DeviceName
InitiatingProcessFileName: r1.InitiatingProcessFileName
FileName: r1.FileName
ProcessCommandLine: r1.ProcessCommandLine
RemoteIP: r1.RemoteIP
RemotePort: r1.RemotePort
ActionType: r1.ActionType
ReportId: r1.ReportId
_alerts: ArrayFilter(r1._alerts, body=StructField(Argument(name='a', shape=<ibis.expr.datashape.Columnar object at 0x109e48b90>, dtype={'alert_name': string, 'alert_triggered': boolean}), field='alert_triggered'), param=Argument(name='a', shape=<ibis.expr.datashape.Columnar object at 0x109e48b90>, dtype={'alert_name': string, 'alert_triggered': boolean}))
r3 := Filter[r2]
ArrayLength(r2._alerts) > 0
r4 := Project[r3]
DeviceName: r3.DeviceName
FileName: r3.FileName
ProcessCommandLine: r3.ProcessCommandLine
alert: Unnest(r3._alerts)
Project[r4]
DeviceName: r4.DeviceName
FileName: r4.FileName
alert_name: StructField(r4.alert, field='alert_name')SELECT
"t3"."DeviceName",
"t3"."FileName",
"t3"."alert"."alert_name" AS "alert_name"
FROM (
SELECT
"t2"."DeviceName",
"t2"."FileName",
"t2"."ProcessCommandLine",
UNNEST("t2"."_alerts") AS "alert"
FROM (
SELECT
*
FROM (
SELECT
"t0"."Timestamp",
"t0"."DeviceId",
"t0"."DeviceName",
"t0"."InitiatingProcessFileName",
"t0"."FileName",
"t0"."ProcessCommandLine",
"t0"."RemoteIP",
"t0"."RemotePort",
"t0"."ActionType",
"t0"."ReportId",
LIST_FILTER(
CAST([
{'alert_name': 'CDC_373_AppleScriptToPython', 'alert_triggered': (
"t0"."InitiatingProcessFileName" = 'osascript'
)
AND LOWER("t0"."FileName") IN ('sh', 'bash')
AND CONTAINS(LOWER("t0"."ProcessCommandLine"), 'python &')},
{'alert_name': 'CDC_011_ProcDumpLsass', 'alert_triggered': (
"t0"."FileName" = 'procdump.exe'
)
AND CONTAINS(LOWER("t0"."ProcessCommandLine"), 'lsass.exe')}
] AS STRUCT("alert_name" TEXT, "alert_triggered" BOOLEAN)[]),
__ibis_param_a__ -> (
__ibis_param_a__
)."alert_triggered"
) AS "_alerts"
FROM "ibis_pandas_memtable_fwbzej7h3rc4tcwr4dzokn6q6q" AS "t0"
) AS "t1"
WHERE
ARRAY_LENGTH("t1"."_alerts") > 0
) AS "t2"
) AS "t3"
One scan, three rules. The intermediate tables at each stage show how the array shrinks as non-firing rules are dropped and how the unnest expands signals back to one row per match. Adding a fourth rule means registering one more struct in the mutate call - the rest of the pipeline is unchanged. The same expression compiles identically on DuckDB, Spark, Snowflake, or ClickHouse.
The composition property matters most when you are adding to the catalogue. A new rule is a new predicate function and a one-line registration in the rules list. There is no config file to update, no pipeline to redeploy, no schema migration. The fused job picks it up on the next run because the expression tree is rebuilt from the list at compile time.
The cell below lets you add a new predicate to the registry and immediately see it appear in the fused output.
r0 := InMemoryTable
data:
PandasDataFrameProxy:
Timestamp DeviceId DeviceName InitiatingProcessFileName \
0 2026-06-18 14:22:00 dev-001 WS-ENG-01 osascript
1 2026-06-18 14:23:00 dev-001 WS-ENG-01 osascript
2 2026-06-18 14:24:00 dev-002 WS-ENG-02 explorer.exe
3 2026-06-18 14:25:00 dev-003 WS-ENG-03 powershell.exe
4 2026-06-18 14:26:00 dev-lab-99 lab-test-01 osascript
.. ... ... ... ...
15 2026-06-18 14:00:00 dev-004 WS-OPS-01
16 2026-06-18 14:30:00 dev-004 WS-OPS-01
17 2026-06-18 15:00:00 dev-004 WS-OPS-01
18 2026-06-18 15:30:00 dev-004 WS-OPS-01
19 2026-06-18 16:00:00 dev-004 WS-OPS-01
FileName ProcessCommandLine RemoteIP \
0 sh sh -c 'echo abc | base64 -d | python &'
1 sh sh -c 'open /Applications/Safari.app'
2 cmd.exe cmd.exe /c dir
3 procdump.exe procdump.exe -ma lsass.exe out.dmp
4 sh sh -c 'echo xyz | python &'
.. ... ... ...
15 10.0.5.20
16 10.0.5.21
17 10.0.5.22
18 10.0.5.23
19 10.0.5.24
RemotePort ActionType ReportId
0 0 ProcessCreated 1001
1 0 ProcessCreated 1002
2 0 ProcessCreated 1003
3 0 ProcessCreated 1004
4 0 ProcessCreated 1005
.. ... ... ...
15 3389 ConnectionRequest 2010
16 3389 ConnectionRequest 2011
17 3389 ConnectionRequest 2012
18 3389 ConnectionRequest 2013
19 3389 ConnectionRequest 2014
[20 rows x 10 columns]
r1 := Project[r0]
Timestamp: r0.Timestamp
DeviceId: r0.DeviceId
DeviceName: r0.DeviceName
InitiatingProcessFileName: r0.InitiatingProcessFileName
FileName: r0.FileName
ProcessCommandLine: r0.ProcessCommandLine
RemoteIP: r0.RemoteIP
RemotePort: r0.RemotePort
ActionType: r0.ActionType
ReportId: r0.ReportId
_alerts: Array([StructColumn(names=['alert_name', 'alert_triggered'], values=['CDC_373_AppleScriptToPython', r0.InitiatingProcessFileName == 'osascript' & InValues(value=Lowercase(r0.FileName), options=['sh', 'bash']) & StringContains(haystack=Lowercase(r0.ProcessCommandLine), needle='python &')]), StructColumn(names=['alert_name', 'alert_triggered'], values=['CDC_011_ProcDumpLsass', r0.FileName == 'procdump.exe' & StringContains(haystack=Lowercase(r0.ProcessCommandLine), needle='lsass.exe')]), StructColumn(names=['alert_name', 'alert_triggered'], values=['CDC_214_NetshHelperDLL', StringContains(haystack=Lowercase(r0.ProcessCommandLine), needle='hklm\\software\\microsoft\\netsh')])])
r2 := Project[r1]
Timestamp: r1.Timestamp
DeviceId: r1.DeviceId
DeviceName: r1.DeviceName
InitiatingProcessFileName: r1.InitiatingProcessFileName
FileName: r1.FileName
ProcessCommandLine: r1.ProcessCommandLine
RemoteIP: r1.RemoteIP
RemotePort: r1.RemotePort
ActionType: r1.ActionType
ReportId: r1.ReportId
_alerts: ArrayFilter(r1._alerts, body=StructField(Argument(name='a', shape=<ibis.expr.datashape.Columnar object at 0x109e48b90>, dtype={'alert_name': string, 'alert_triggered': boolean}), field='alert_triggered'), param=Argument(name='a', shape=<ibis.expr.datashape.Columnar object at 0x109e48b90>, dtype={'alert_name': string, 'alert_triggered': boolean}))
r3 := Filter[r2]
ArrayLength(r2._alerts) > 0
r4 := Project[r3]
DeviceName: r3.DeviceName
FileName: r3.FileName
alert: Unnest(r3._alerts)
Project[r4]
DeviceName: r4.DeviceName
FileName: r4.FileName
alert_name: StructField(r4.alert, field='alert_name')SELECT
"t3"."DeviceName",
"t3"."FileName",
"t3"."alert"."alert_name" AS "alert_name"
FROM (
SELECT
"t2"."DeviceName",
"t2"."FileName",
UNNEST("t2"."_alerts") AS "alert"
FROM (
SELECT
*
FROM (
SELECT
"t0"."Timestamp",
"t0"."DeviceId",
"t0"."DeviceName",
"t0"."InitiatingProcessFileName",
"t0"."FileName",
"t0"."ProcessCommandLine",
"t0"."RemoteIP",
"t0"."RemotePort",
"t0"."ActionType",
"t0"."ReportId",
LIST_FILTER(
CAST([
{'alert_name': 'CDC_373_AppleScriptToPython', 'alert_triggered': (
"t0"."InitiatingProcessFileName" = 'osascript'
)
AND LOWER("t0"."FileName") IN ('sh', 'bash')
AND CONTAINS(LOWER("t0"."ProcessCommandLine"), 'python &')},
{'alert_name': 'CDC_011_ProcDumpLsass', 'alert_triggered': (
"t0"."FileName" = 'procdump.exe'
)
AND CONTAINS(LOWER("t0"."ProcessCommandLine"), 'lsass.exe')},
{'alert_name': 'CDC_214_NetshHelperDLL', 'alert_triggered': CONTAINS(LOWER("t0"."ProcessCommandLine"), 'hklm\software\microsoft\netsh')}
] AS STRUCT("alert_name" TEXT, "alert_triggered" BOOLEAN)[]),
__ibis_param_a__ -> (
__ibis_param_a__
)."alert_triggered"
) AS "_alerts"
FROM "ibis_pandas_memtable_fwbzej7h3rc4tcwr4dzokn6q6q" AS "t0"
) AS "t1"
WHERE
ARRAY_LENGTH("t1"."_alerts") > 0
) AS "t2"
) AS "t3"
Composition is concatenation. The registry is a Python list. Registering a new rule appends one element to it. The fused expression tree is rebuilt from the list on every compile, so the new rule joins the scan without touching anything downstream. The output shape stays identical regardless of how many rules are in the list.
The five-stage fused pipeline. Each stage is a single ibis operation: mutate builds the array of per-rule structs, array_filter drops non-firing entries, the row filter drops empty-array rows, unnest expands to one row per signal, and project shapes the output. Hover a stage for the ibis snippet. Because the whole tree compiles at once, the backend sees a single query with no intermediate writes.
Batch and streaming, one rule
Most detection rules are per-row predicates: one event arrives, the rule evaluates to true or false, an alert is written or not. Those stateless rules run identically on every backend ibis supports, batch or streaming. The more interesting case is the windowed rule: how many distinct remote IPs has this device connected to over RDP in the last 24 hours? That question requires state across a time window, and the answer differs depending on where the computation runs. A streaming engine like Apache Spark Structured Streaming, Apache Flink, or RisingWave has native watermarked tumbling windows. A batch engine like DuckDB or Snowflake does not. The naive response is a branch in the rule: an if streaming: ladder that maintains two code paths. The framework response is a single helper.
HighUniqueRemoteDesktopConnections is the canonical contextual detection in the codebase. It counts distinct destination IPs per device over a 24-hour tumbling window and fires when the count crosses a threshold. The full rule is about fifteen lines of ibis, the two load-bearing structural lines being window_size and watermark as class attributes.
The rule below shows the windowed detection with the key hover-annotated tokens:
class HighUniqueRemoteDesktopConnections(Contextual): """Trigger on devices that touch unusually many distinct RDP targets in a 24h window.""" window_size = ibis.interval(hours=24) watermark = ibis.interval(hours=4) def __init__(self, unique_connections_threshold: int = 10) -> None: self.config = Configuration( alert_name="CDC_168_DBX_HighRDPCount", source_table="cdc.DeviceNetworkEvents", mitre_attack_techniques=[MitreAttackTechnique.T1060], # ... ) self.unique_connections_threshold = unique_connections_threshold def detection_logic(self, events: Table) -> Table: rdp_requests = events.filter( (events.RemotePort == 3389) & (events.ActionType == "ConnectionRequest") & (events.RemoteIP != "127.0.0.1") ) aggregated = tumble_window_agg( rdp_requests, time_col="Timestamp", size=self.window_size, by=["DeviceId", "DeviceName"], number_of_unique_rdp_connections=rdp_requests.RemoteIP.approx_nunique(), ) return aggregated.filter( aggregated.number_of_unique_rdp_connections >= self.unique_connections_threshold )
The windowed RDP fan-out rule. Hover the numbered markers for what each structural element does. The rule contains no Spark imports and no streaming-specific API calls. Backend choice is visible only in the tumble_window_agg dispatch, which is a framework concern, not a rule concern. approx_nunique() is a deliberate choice imposed by Spark Structured Streaming’s restriction on exact distinct aggregations over windows.
The interactive cell below runs this rule against a synthetic set of RDP connection events. The events include one device making many distinct connections (triggering the rule) and one device staying below threshold. The cell shows the result through the DuckDB batch path, then renders the ibis SQL that DuckDB compiles to so the lowered form is visible.
r0 := InMemoryTable
data:
PandasDataFrameProxy:
Timestamp DeviceId DeviceName InitiatingProcessFileName \
0 2026-06-18 14:22:00 dev-001 WS-ENG-01 osascript
1 2026-06-18 14:23:00 dev-001 WS-ENG-01 osascript
2 2026-06-18 14:24:00 dev-002 WS-ENG-02 explorer.exe
3 2026-06-18 14:25:00 dev-003 WS-ENG-03 powershell.exe
4 2026-06-18 14:26:00 dev-lab-99 lab-test-01 osascript
.. ... ... ... ...
15 2026-06-18 14:00:00 dev-004 WS-OPS-01
16 2026-06-18 14:30:00 dev-004 WS-OPS-01
17 2026-06-18 15:00:00 dev-004 WS-OPS-01
18 2026-06-18 15:30:00 dev-004 WS-OPS-01
19 2026-06-18 16:00:00 dev-004 WS-OPS-01
FileName ProcessCommandLine RemoteIP \
0 sh sh -c 'echo abc | base64 -d | python &'
1 sh sh -c 'open /Applications/Safari.app'
2 cmd.exe cmd.exe /c dir
3 procdump.exe procdump.exe -ma lsass.exe out.dmp
4 sh sh -c 'echo xyz | python &'
.. ... ... ...
15 10.0.5.20
16 10.0.5.21
17 10.0.5.22
18 10.0.5.23
19 10.0.5.24
RemotePort ActionType ReportId
0 0 ProcessCreated 1001
1 0 ProcessCreated 1002
2 0 ProcessCreated 1003
3 0 ProcessCreated 1004
4 0 ProcessCreated 1005
.. ... ... ...
15 3389 ConnectionRequest 2010
16 3389 ConnectionRequest 2011
17 3389 ConnectionRequest 2012
18 3389 ConnectionRequest 2013
19 3389 ConnectionRequest 2014
[20 rows x 10 columns]
r1 := Filter[r0]
r0.RemotePort == 3389
r0.ActionType == 'ConnectionRequest'
Aggregate[r1]
groups:
window_day: TimestampTruncate(r1.Timestamp, unit=<IntervalUnit.DAY: 'D'>)
DeviceId: r1.DeviceId
DeviceName: r1.DeviceName
metrics:
unique_destinations: CountDistinct(r1.RemoteIP)
connection_count: CountStar(r1)SELECT
DATE_TRUNC('DAY', "t1"."Timestamp") AS "window_day",
"t1"."DeviceId",
"t1"."DeviceName",
COUNT(DISTINCT "t1"."RemoteIP") AS "unique_destinations",
COUNT(*) AS "connection_count"
FROM (
SELECT
*
FROM "ibis_pandas_memtable_fwbzej7h3rc4tcwr4dzokn6q6q" AS "t0"
WHERE
"t0"."RemotePort" = 3389 AND "t0"."ActionType" = 'ConnectionRequest'
) AS "t1"
GROUP BY
1,
2,
3
r0 := InMemoryTable
data:
PandasDataFrameProxy:
Timestamp DeviceId DeviceName InitiatingProcessFileName \
0 2026-06-18 14:22:00 dev-001 WS-ENG-01 osascript
1 2026-06-18 14:23:00 dev-001 WS-ENG-01 osascript
2 2026-06-18 14:24:00 dev-002 WS-ENG-02 explorer.exe
3 2026-06-18 14:25:00 dev-003 WS-ENG-03 powershell.exe
4 2026-06-18 14:26:00 dev-lab-99 lab-test-01 osascript
.. ... ... ... ...
15 2026-06-18 14:00:00 dev-004 WS-OPS-01
16 2026-06-18 14:30:00 dev-004 WS-OPS-01
17 2026-06-18 15:00:00 dev-004 WS-OPS-01
18 2026-06-18 15:30:00 dev-004 WS-OPS-01
19 2026-06-18 16:00:00 dev-004 WS-OPS-01
FileName ProcessCommandLine RemoteIP \
0 sh sh -c 'echo abc | base64 -d | python &'
1 sh sh -c 'open /Applications/Safari.app'
2 cmd.exe cmd.exe /c dir
3 procdump.exe procdump.exe -ma lsass.exe out.dmp
4 sh sh -c 'echo xyz | python &'
.. ... ... ...
15 10.0.5.20
16 10.0.5.21
17 10.0.5.22
18 10.0.5.23
19 10.0.5.24
RemotePort ActionType ReportId
0 0 ProcessCreated 1001
1 0 ProcessCreated 1002
2 0 ProcessCreated 1003
3 0 ProcessCreated 1004
4 0 ProcessCreated 1005
.. ... ... ...
15 3389 ConnectionRequest 2010
16 3389 ConnectionRequest 2011
17 3389 ConnectionRequest 2012
18 3389 ConnectionRequest 2013
19 3389 ConnectionRequest 2014
[20 rows x 10 columns]
r1 := Filter[r0]
r0.RemotePort == 3389
r0.ActionType == 'ConnectionRequest'
r2 := Aggregate[r1]
groups:
window_day: TimestampTruncate(r1.Timestamp, unit=<IntervalUnit.DAY: 'D'>)
DeviceId: r1.DeviceId
DeviceName: r1.DeviceName
metrics:
unique_destinations: CountDistinct(r1.RemoteIP)
connection_count: CountStar(r1)
Filter[r2]
r2.unique_destinations > 10SELECT
*
FROM (
SELECT
DATE_TRUNC('DAY', "t1"."Timestamp") AS "window_day",
"t1"."DeviceId",
"t1"."DeviceName",
COUNT(DISTINCT "t1"."RemoteIP") AS "unique_destinations",
COUNT(*) AS "connection_count"
FROM (
SELECT
*
FROM "ibis_pandas_memtable_fwbzej7h3rc4tcwr4dzokn6q6q" AS "t0"
WHERE
"t0"."RemotePort" = 3389 AND "t0"."ActionType" = 'ConnectionRequest'
) AS "t1"
GROUP BY
1,
2,
3
) AS "t2"
WHERE
"t2"."unique_destinations" > 10
The batch compile path on DuckDB. The 24-hour window becomes a date_trunc(‘day’, Timestamp) bucket, and the aggregation groups by device and bucket before filtering on the threshold. The output schema is identical to what the streaming path produces: window_start, window_end, DeviceName, DeviceId, and the count column. A cross-backend regression test in CI runs the same rule against a local PySpark session and asserts the two results agree.
The dispatch lives entirely in tumble_window_agg, a helper in ibis_utilities.py. On a streaming backend (identified by backend name matching {"flink", "pyspark", "risingwave"}), the function uses ibis’s native window_by(time_col).tumble(size=size).agg(...), which compiles to the streaming window aggregate the runtime expects. On DuckDB or any other batch backend it falls back to truncate(time_col, unit) + group_by + aggregate, producing the same column names and types. The rule code does not branch. The framework does.
The cell below shows the same rule compiled to the Spark streaming dialect using ibis.to_sql(dialect="pyspark"). The predicate is unchanged from the DuckDB version: the only difference is what appears at the aggregation layer.
window_by(Timestamp).tumble(size=24h).agg(...) with watermark='4h' attached to the streaming source.The streaming compile path. The Spark dialect emits window_by(Timestamp).tumble(size=INTERVAL 24 HOURS).agg(…) in place of the date-truncation bucket. The filter predicate above the aggregation is byte-for-byte equivalent to the DuckDB version. Swapping a DuckDB connection for a PySpark connection is the only change required to move a windowed rule from CI to a Spark cluster.
Where the watermark lives is a design choice that has detection-engineering consequences. In the prototype, watermark is a class attribute on the rule. A reviewer reading the rule file sees the window size and the late-event tolerance in the same place, as one coherent statement about the detection’s timing model. The runner reads both attributes before setting up the streaming query. A runner-level override is available for operational reasons (a source known to deliver events late), but the default is rule-owned, because late-event tolerance is a correctness concern for the rule writer: decide how long to wait for delayed events before closing the window and potentially missing the activity you are looking for.
The two compile paths share an output schema intentionally. window_start and window_end are present in both the batch fallback and the streaming window, so alert projections, downstream consumers, and test assertions do not need to know which path ran.
Fig 4: The streaming and batch dispatch paths for tumble_window_agg. A single call at the centre branches left to DuckDB (batch: truncate + group_by + aggregate) and right to Spark or Flink (streaming: window_by + tumble + agg). Both paths produce the same output schema and terminate at an identical alert-ready table. Hover each path for the lowered ibis expression.
A signals database
The signals envelopeEvery detection rule produces structured output. When a rule fires, the output is not just a notification. It is a row: the original event, the rule name, a severity level, and a set of observables pulled from the event (the device name, the identity, any process hashes, any IP addresses). We call this the signal envelope. Write those rows to a Delta or Iceberg table and the output of detection becomes a first-class queryable artifact, not an ephemeral alert stream.
The envelope shape is fixed across all rules. A rule is allowed to attach rule-specific metadata to a properties map, but the top-level columns are uniform: signal_id, ts, rule_name, severity, observables, alert_suppressed, alert_suppression_data, and origin_event. Keeping that contract stable is what makes everything downstream composable.
Here is a minimal envelope schema written as a Python dataclass:
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class AlertEnvelope:
signal_id: str
ts: datetime
rule_name: str
severity: int # 1 low .. 5 critical
observables: dict # device, identity, hashes, IPs
alert_suppressed: bool = False
alert_suppression_data: list = field(default_factory=list)
properties: dict = field(default_factory=dict)
# ...The envelope is what every rule writes, regardless of which engine evaluated it. The observables dict is intentionally loose - rules pull whichever fields are meaningful for their technique. The suppression fields are always present, even when nothing suppressed the alert, because downstream tooling should not need to check for key existence.
The cell below runs two rules over a synthetic event table and writes the resulting envelopes to a DuckDB table called signals.
r0 := InMemoryTable
data:
PandasDataFrameProxy:
Timestamp DeviceId DeviceName InitiatingProcessFileName \
0 2026-06-18 14:22:00 dev-001 WS-ENG-01 osascript
1 2026-06-18 14:23:00 dev-001 WS-ENG-01 osascript
2 2026-06-18 14:24:00 dev-002 WS-ENG-02 explorer.exe
3 2026-06-18 14:25:00 dev-003 WS-ENG-03 powershell.exe
4 2026-06-18 14:26:00 dev-lab-99 lab-test-01 osascript
.. ... ... ... ...
15 2026-06-18 14:00:00 dev-004 WS-OPS-01
16 2026-06-18 14:30:00 dev-004 WS-OPS-01
17 2026-06-18 15:00:00 dev-004 WS-OPS-01
18 2026-06-18 15:30:00 dev-004 WS-OPS-01
19 2026-06-18 16:00:00 dev-004 WS-OPS-01
FileName ProcessCommandLine RemoteIP \
0 sh sh -c 'echo abc | base64 -d | python &'
1 sh sh -c 'open /Applications/Safari.app'
2 cmd.exe cmd.exe /c dir
3 procdump.exe procdump.exe -ma lsass.exe out.dmp
4 sh sh -c 'echo xyz | python &'
.. ... ... ...
15 10.0.5.20
16 10.0.5.21
17 10.0.5.22
18 10.0.5.23
19 10.0.5.24
RemotePort ActionType ReportId
0 0 ProcessCreated 1001
1 0 ProcessCreated 1002
2 0 ProcessCreated 1003
3 0 ProcessCreated 1004
4 0 ProcessCreated 1005
.. ... ... ...
15 3389 ConnectionRequest 2010
16 3389 ConnectionRequest 2011
17 3389 ConnectionRequest 2012
18 3389 ConnectionRequest 2013
19 3389 ConnectionRequest 2014
[20 rows x 10 columns]
r1 := Project[r0]
Timestamp: r0.Timestamp
DeviceId: r0.DeviceId
DeviceName: r0.DeviceName
InitiatingProcessFileName: r0.InitiatingProcessFileName
FileName: r0.FileName
ProcessCommandLine: r0.ProcessCommandLine
RemoteIP: r0.RemoteIP
RemotePort: r0.RemotePort
ActionType: r0.ActionType
ReportId: r0.ReportId
_alerts: Array([StructColumn(names=['alert_name', 'alert_triggered'], values=['CDC_373_AppleScriptToPython', r0.InitiatingProcessFileName == 'osascript' & InValues(value=Lowercase(r0.FileName), options=['sh', 'bash']) & StringContains(haystack=Lowercase(r0.ProcessCommandLine), needle='python &')]), StructColumn(names=['alert_name', 'alert_triggered'], values=['CDC_011_ProcDumpLsass', r0.FileName == 'procdump.exe' & StringContains(haystack=Lowercase(r0.ProcessCommandLine), needle='lsass.exe')]), StructColumn(names=['alert_name', 'alert_triggered'], values=['CDC_214_NetshHelperDLL', StringContains(haystack=Lowercase(r0.ProcessCommandLine), needle='hklm\\software\\microsoft\\netsh')])])
r2 := Project[r1]
Timestamp: r1.Timestamp
DeviceId: r1.DeviceId
DeviceName: r1.DeviceName
InitiatingProcessFileName: r1.InitiatingProcessFileName
FileName: r1.FileName
ProcessCommandLine: r1.ProcessCommandLine
RemoteIP: r1.RemoteIP
RemotePort: r1.RemotePort
ActionType: r1.ActionType
ReportId: r1.ReportId
_alerts: ArrayFilter(r1._alerts, body=StructField(Argument(name='a', shape=<ibis.expr.datashape.Columnar object at 0x109e48b90>, dtype={'alert_name': string, 'alert_triggered': boolean}), field='alert_triggered'), param=Argument(name='a', shape=<ibis.expr.datashape.Columnar object at 0x109e48b90>, dtype={'alert_name': string, 'alert_triggered': boolean}))
r3 := Filter[r2]
ArrayLength(r2._alerts) > 0
r4 := Project[r3]
DeviceName: r3.DeviceName
FileName: r3.FileName
alert: Unnest(r3._alerts)
r5 := Project[r4]
DeviceName: r4.DeviceName
FileName: r4.FileName
alert_name: StructField(r4.alert, field='alert_name')
Project[r5]
DeviceName: r5.DeviceName
FileName: r5.FileName
alert_name: r5.alert_name
severity: 50
signal_timestamp: TimestampNow()SELECT
"t4"."DeviceName",
"t4"."FileName",
"t4"."alert_name",
50 AS "severity",
CAST(CURRENT_TIMESTAMP AS TIMESTAMP) AS "signal_timestamp"
FROM (
SELECT
"t3"."DeviceName",
"t3"."FileName",
"t3"."alert"."alert_name" AS "alert_name"
FROM (
SELECT
"t2"."DeviceName",
"t2"."FileName",
UNNEST("t2"."_alerts") AS "alert"
FROM (
SELECT
*
FROM (
SELECT
"t0"."Timestamp",
"t0"."DeviceId",
"t0"."DeviceName",
"t0"."InitiatingProcessFileName",
"t0"."FileName",
"t0"."ProcessCommandLine",
"t0"."RemoteIP",
"t0"."RemotePort",
"t0"."ActionType",
"t0"."ReportId",
LIST_FILTER(
CAST([
{'alert_name': 'CDC_373_AppleScriptToPython', 'alert_triggered': (
"t0"."InitiatingProcessFileName" = 'osascript'
)
AND LOWER("t0"."FileName") IN ('sh', 'bash')
AND CONTAINS(LOWER("t0"."ProcessCommandLine"), 'python &')},
{'alert_name': 'CDC_011_ProcDumpLsass', 'alert_triggered': (
"t0"."FileName" = 'procdump.exe'
)
AND CONTAINS(LOWER("t0"."ProcessCommandLine"), 'lsass.exe')},
{'alert_name': 'CDC_214_NetshHelperDLL', 'alert_triggered': CONTAINS(LOWER("t0"."ProcessCommandLine"), 'hklm\software\microsoft\netsh')}
] AS STRUCT("alert_name" TEXT, "alert_triggered" BOOLEAN)[]),
__ibis_param_a__ -> (
__ibis_param_a__
)."alert_triggered"
) AS "_alerts"
FROM "ibis_pandas_memtable_fwbzej7h3rc4tcwr4dzokn6q6q" AS "t0"
) AS "t1"
WHERE
ARRAY_LENGTH("t1"."_alerts") > 0
) AS "t2"
) AS "t3"
) AS "t4"
The signals table is written by the same ibis connection used to evaluate the rules. Each row is one matched (event, rule) pair. The alert_suppressed column defaults to false here. Suppression is applied in a later pass covered in the whitelisting section. The table is immediately queryable - the cell above renders the first few rows so you can see the exact shape before reading on.
A signals table makes risk-based alerting (RBA) another ibis rule. The standard RBA pattern is: aggregate signals per identity per time window, sum the severity scores, raise a case when the score crosses a threshold. Written as an ibis expression over the signals table, that aggregation uses the same primitives as any other rule - group_by, aggregate, filter, window. There is no separate scoring service, no webhook, no custom pipeline stage.
The following cell writes an RBA aggregator as an ibis expression. It groups by the identity observable over a 24-hour window, sums severity, and returns the identities whose score exceeds a configurable threshold.
r0 := InMemoryTable
data:
PandasDataFrameProxy:
Timestamp DeviceId DeviceName InitiatingProcessFileName \
0 2026-06-18 14:22:00 dev-001 WS-ENG-01 osascript
1 2026-06-18 14:23:00 dev-001 WS-ENG-01 osascript
2 2026-06-18 14:24:00 dev-002 WS-ENG-02 explorer.exe
3 2026-06-18 14:25:00 dev-003 WS-ENG-03 powershell.exe
4 2026-06-18 14:26:00 dev-lab-99 lab-test-01 osascript
.. ... ... ... ...
15 2026-06-18 14:00:00 dev-004 WS-OPS-01
16 2026-06-18 14:30:00 dev-004 WS-OPS-01
17 2026-06-18 15:00:00 dev-004 WS-OPS-01
18 2026-06-18 15:30:00 dev-004 WS-OPS-01
19 2026-06-18 16:00:00 dev-004 WS-OPS-01
FileName ProcessCommandLine RemoteIP \
0 sh sh -c 'echo abc | base64 -d | python &'
1 sh sh -c 'open /Applications/Safari.app'
2 cmd.exe cmd.exe /c dir
3 procdump.exe procdump.exe -ma lsass.exe out.dmp
4 sh sh -c 'echo xyz | python &'
.. ... ... ...
15 10.0.5.20
16 10.0.5.21
17 10.0.5.22
18 10.0.5.23
19 10.0.5.24
RemotePort ActionType ReportId
0 0 ProcessCreated 1001
1 0 ProcessCreated 1002
2 0 ProcessCreated 1003
3 0 ProcessCreated 1004
4 0 ProcessCreated 1005
.. ... ... ...
15 3389 ConnectionRequest 2010
16 3389 ConnectionRequest 2011
17 3389 ConnectionRequest 2012
18 3389 ConnectionRequest 2013
19 3389 ConnectionRequest 2014
[20 rows x 10 columns]
r1 := Project[r0]
Timestamp: r0.Timestamp
DeviceId: r0.DeviceId
DeviceName: r0.DeviceName
InitiatingProcessFileName: r0.InitiatingProcessFileName
FileName: r0.FileName
ProcessCommandLine: r0.ProcessCommandLine
RemoteIP: r0.RemoteIP
RemotePort: r0.RemotePort
ActionType: r0.ActionType
ReportId: r0.ReportId
_alerts: Array([StructColumn(names=['alert_name', 'alert_triggered'], values=['CDC_373_AppleScriptToPython', r0.InitiatingProcessFileName == 'osascript' & InValues(value=Lowercase(r0.FileName), options=['sh', 'bash']) & StringContains(haystack=Lowercase(r0.ProcessCommandLine), needle='python &')]), StructColumn(names=['alert_name', 'alert_triggered'], values=['CDC_011_ProcDumpLsass', r0.FileName == 'procdump.exe' & StringContains(haystack=Lowercase(r0.ProcessCommandLine), needle='lsass.exe')]), StructColumn(names=['alert_name', 'alert_triggered'], values=['CDC_214_NetshHelperDLL', StringContains(haystack=Lowercase(r0.ProcessCommandLine), needle='hklm\\software\\microsoft\\netsh')])])
r2 := Project[r1]
Timestamp: r1.Timestamp
DeviceId: r1.DeviceId
DeviceName: r1.DeviceName
InitiatingProcessFileName: r1.InitiatingProcessFileName
FileName: r1.FileName
ProcessCommandLine: r1.ProcessCommandLine
RemoteIP: r1.RemoteIP
RemotePort: r1.RemotePort
ActionType: r1.ActionType
ReportId: r1.ReportId
_alerts: ArrayFilter(r1._alerts, body=StructField(Argument(name='a', shape=<ibis.expr.datashape.Columnar object at 0x109e48b90>, dtype={'alert_name': string, 'alert_triggered': boolean}), field='alert_triggered'), param=Argument(name='a', shape=<ibis.expr.datashape.Columnar object at 0x109e48b90>, dtype={'alert_name': string, 'alert_triggered': boolean}))
r3 := Filter[r2]
ArrayLength(r2._alerts) > 0
r4 := Project[r3]
DeviceName: r3.DeviceName
FileName: r3.FileName
alert: Unnest(r3._alerts)
r5 := Project[r4]
DeviceName: r4.DeviceName
FileName: r4.FileName
alert_name: StructField(r4.alert, field='alert_name')
r6 := Project[r5]
DeviceName: r5.DeviceName
FileName: r5.FileName
alert_name: r5.alert_name
severity: 50
signal_timestamp: TimestampNow()
r7 := Aggregate[r6]
groups:
DeviceName: r6.DeviceName
metrics:
signal_count: CountStar(r6)
total_score: Sum(r6.severity)
r8 := Filter[r7]
r7.total_score > 49
Sort[r8]
desc r8.total_scoreSELECT
*
FROM (
SELECT
"t5"."DeviceName",
COUNT(*) AS "signal_count",
SUM("t5"."severity") AS "total_score"
FROM (
SELECT
"t4"."DeviceName",
"t4"."FileName",
"t4"."alert_name",
50 AS "severity",
CAST(CURRENT_TIMESTAMP AS TIMESTAMP) AS "signal_timestamp"
FROM (
SELECT
"t3"."DeviceName",
"t3"."FileName",
"t3"."alert"."alert_name" AS "alert_name"
FROM (
SELECT
"t2"."DeviceName",
"t2"."FileName",
UNNEST("t2"."_alerts") AS "alert"
FROM (
SELECT
*
FROM (
SELECT
"t0"."Timestamp",
"t0"."DeviceId",
"t0"."DeviceName",
"t0"."InitiatingProcessFileName",
"t0"."FileName",
"t0"."ProcessCommandLine",
"t0"."RemoteIP",
"t0"."RemotePort",
"t0"."ActionType",
"t0"."ReportId",
LIST_FILTER(
CAST([
{'alert_name': 'CDC_373_AppleScriptToPython', 'alert_triggered': (
"t0"."InitiatingProcessFileName" = 'osascript'
)
AND LOWER("t0"."FileName") IN ('sh', 'bash')
AND CONTAINS(LOWER("t0"."ProcessCommandLine"), 'python &')},
{'alert_name': 'CDC_011_ProcDumpLsass', 'alert_triggered': (
"t0"."FileName" = 'procdump.exe'
)
AND CONTAINS(LOWER("t0"."ProcessCommandLine"), 'lsass.exe')},
{'alert_name': 'CDC_214_NetshHelperDLL', 'alert_triggered': CONTAINS(LOWER("t0"."ProcessCommandLine"), 'hklm\software\microsoft\netsh')}
] AS STRUCT("alert_name" TEXT, "alert_triggered" BOOLEAN)[]),
__ibis_param_a__ -> (
__ibis_param_a__
)."alert_triggered"
) AS "_alerts"
FROM "ibis_pandas_memtable_fwbzej7h3rc4tcwr4dzokn6q6q" AS "t0"
) AS "t1"
WHERE
ARRAY_LENGTH("t1"."_alerts") > 0
) AS "t2"
) AS "t3"
) AS "t4"
) AS "t5"
GROUP BY
1
) AS "t6"
WHERE
"t6"."total_score" > 49
ORDER BY
"t6"."total_score" DESC
The RBA query is just another ibis expression, this time reading from the signals table rather than from a source telemetry table. Adding a new signal source means adding a rule that writes to the signals table - the RBA aggregator picks it up without modification. Adjusting the threshold or the window size is a one-line change to the expression, not a config-system edit.
A useful property of this pattern: RBA is version-controlled as code. Threshold changes appear in pull requests. The history of scoring policy changes is in git, not buried in a web console.
Reading signals backThe signals table is useful beyond RBA. Any tooling that can read Delta or Iceberg can consume it. A BI dashboard that counts signals by rule, severity, and device over rolling windows is a SQL query against the signals table. A SOAR integration that opens cases for identities above the RBA threshold is a reader of the same table. A triage UI that lets an analyst filter suppressed from non-suppressed signals is another reader. None of those consumers need access to the detection runtime or the source telemetry.
The topology below shows the data flow from source telemetry to signals to consumers.
Fig 5: The signals topology. Source telemetry enters from the top: endpoint logs (MDE), identity (Azure AD), warehouse data (Snowflake), and network flow (ClickHouse). Detection rules fan out over those sources and all write into the same signals table. From the signals table, four consumers read independently: the RBA aggregator, a triage UI, a SOAR integration, and a BI dashboard. None of those consumers couple to the detection runtime or to the source telemetry directly. Hover each node for one line on what it does.
The signals table also gives you a free audit log. Every signal ever emitted is a row. Retroactive analysis of rule performance (what fired on a given date, which rules produced the most suppressed signals, which rules never fired) is a SQL query, not a grep through alert archive logs. For detection engineering teams that need to demonstrate coverage or justify rule changes to stakeholders, having a queryable history is genuinely useful.
Iceberg vs DeltaBoth open table formats work here. For a multi-engine platform (DuckDB locally, Spark in production, Snowflake for ad-hoc, Trino federated), Apache Iceberg is the safer default. Its spec is engine-neutral, it is readable from all the engines listed above, and DuckDB’s iceberg_scan makes local development against a production snapshot practical. Delta Lake has tighter Databricks tooling and native Spark streaming source semantics, which matter if the team is already anchored in Databricks. Both are readable from DuckDB via iceberg_scan and delta_scan, so the local development loop works either way.
One practical note: writing signals from DuckDB directly into a transactional Iceberg or Delta table at scale is still primarily read-oriented in the current tooling. For a production signals sink, use the engine that owns the table (Spark, Trino, or Snowflake) to write, and DuckDB to read local snapshots for development. This is covered further in the limitations section.
Whitelisting and filters
Whitelists are the highest-churn surface in any detection platform. New benign processes appear after every software rollout. Lab hosts need to be excluded from production rules. Known-noisy users get filtered while an investigation is underway. In most platforms, those suppressions live in external config: YAML lookup tables, Splunk macros, environment variables passed to the job at run time. External config means the whitelist is reviewed on a different cycle than the rule, stored in a different place, and often invisible to whoever reviews a rule change. Treating the suppression as code, next to the rule, keeps it in the same review loop and the same version history.
In this design, a suppression is another boolean expression over the same table. The @suppression decorator registers it alongside the rule it belongs to. Common cases get concise one-line builders: known_device(devices=[...]), known_process(filenames=[...]), known_hash(hashes=[...]). Less common cases are just plain ibis predicates. The framework evaluates each suppression expression in the fused job alongside the rule predicates, so suppression is not a post-hoc filter on results - it is evaluated per event in the same scan.
The key decision in the design is what to do when a suppression matches. The answer here is to not drop the alert. The event still gets written to the signals table with alert_suppressed=true and an alert_suppression_data array that records which suppression matched and why. An alert that fires against a known-benign device is still evidence that the rule is working. It is also a record that a suppression is still needed, or that it has outlived its purpose.
This gives RBA scoring a concrete input. Suppressed signals get a lower weight or are excluded from the case score, but they accumulate in the same table as actionable signals. A periodic query over that table answers: which suppressions are still firing? Which have not fired in ninety days? Which benign processes appear in suppressed alerts but have no suppression entry yet? The signals table is the audit trail that keeps the whitelist list from growing indefinitely without review.
Suppression as codeThe code below shows the shape of a suppression decorator attached to a rule. Hover the numbered markers for what each token does.
from detectionkit import stateless, suppression, known_device, known_process from detectionkit.mitre import MitreAttackTechnique @stateless( number=412, source_table="cdc.DeviceProcessEvents", mitre=[MitreAttackTechnique.T1059], ) def lolbin_via_mshta(t): return ( (t.FileName.lower().isin(["mshta.exe", "cscript.exe", "wscript.exe"]) & t.ProcessCommandLine.lower().contains("http")) ) @suppression(rule_number=412, reason="IT management tooling - known SCCM distribution points") def suppress_sccm_hosts(t): return known_device(t, devices=["sccm-dist-01", "sccm-dist-02"]) @suppression(rule_number=412, reason="Software deployment - mshta used by package installer during rollout") def suppress_package_installer(t): return known_process(t, filenames=["sccm_installer.exe", "chocolatey.exe"])
A rule and two suppressions, all in the same file. The @suppression decorator binds a predicate to a specific rule by number, carries a human-readable reason, and is evaluated in the same expression tree as the rule itself. Adding a suppression is a pull request against the rule file, not a change to a separate config system.
The following cell runs both suppressions against synthetic events that include one SCCM host and one genuinely suspicious endpoint. Each row in the output shows alert_name, alert_suppressed, and the alert_suppression_data array so you can see which suppression matched and which alerts remain actionable.
r0 := InMemoryTable
data:
PandasDataFrameProxy:
Timestamp DeviceId DeviceName InitiatingProcessFileName \
0 2026-06-18 14:22:00 dev-001 WS-ENG-01 osascript
1 2026-06-18 14:23:00 dev-001 WS-ENG-01 osascript
2 2026-06-18 14:24:00 dev-002 WS-ENG-02 explorer.exe
3 2026-06-18 14:25:00 dev-003 WS-ENG-03 powershell.exe
4 2026-06-18 14:26:00 dev-lab-99 lab-test-01 osascript
.. ... ... ... ...
15 2026-06-18 14:00:00 dev-004 WS-OPS-01
16 2026-06-18 14:30:00 dev-004 WS-OPS-01
17 2026-06-18 15:00:00 dev-004 WS-OPS-01
18 2026-06-18 15:30:00 dev-004 WS-OPS-01
19 2026-06-18 16:00:00 dev-004 WS-OPS-01
FileName ProcessCommandLine RemoteIP \
0 sh sh -c 'echo abc | base64 -d | python &'
1 sh sh -c 'open /Applications/Safari.app'
2 cmd.exe cmd.exe /c dir
3 procdump.exe procdump.exe -ma lsass.exe out.dmp
4 sh sh -c 'echo xyz | python &'
.. ... ... ...
15 10.0.5.20
16 10.0.5.21
17 10.0.5.22
18 10.0.5.23
19 10.0.5.24
RemotePort ActionType ReportId
0 0 ProcessCreated 1001
1 0 ProcessCreated 1002
2 0 ProcessCreated 1003
3 0 ProcessCreated 1004
4 0 ProcessCreated 1005
.. ... ... ...
15 3389 ConnectionRequest 2010
16 3389 ConnectionRequest 2011
17 3389 ConnectionRequest 2012
18 3389 ConnectionRequest 2013
19 3389 ConnectionRequest 2014
[20 rows x 10 columns]
r1 := Project[r0]
Timestamp: r0.Timestamp
DeviceId: r0.DeviceId
DeviceName: r0.DeviceName
InitiatingProcessFileName: r0.InitiatingProcessFileName
FileName: r0.FileName
ProcessCommandLine: r0.ProcessCommandLine
RemoteIP: r0.RemoteIP
RemotePort: r0.RemotePort
ActionType: r0.ActionType
ReportId: r0.ReportId
_alerts: Array([StructColumn(names=['alert_name', 'alert_triggered', 'alert_suppression_data'], values=['CDC_373_AppleScriptToPython', r0.InitiatingProcessFileName == 'osascript' & InValues(value=Lowercase(r0.FileName), options=['sh', 'bash']) & StringContains(haystack=Lowercase(r0.ProcessCommandLine), needle='python &'), ArrayFilter(Array([StructColumn(names=['name'], values=['known_device_lab'])]), body=True, param=Argument(name='_s', shape=<ibis.expr.datashape.Scalar object at 0x109e48b60>, dtype={'name': string}))])])
r2 := Project[r1]
Timestamp: r1.Timestamp
DeviceId: r1.DeviceId
DeviceName: r1.DeviceName
InitiatingProcessFileName: r1.InitiatingProcessFileName
FileName: r1.FileName
ProcessCommandLine: r1.ProcessCommandLine
RemoteIP: r1.RemoteIP
RemotePort: r1.RemotePort
ActionType: r1.ActionType
ReportId: r1.ReportId
_alerts: ArrayFilter(r1._alerts, body=StructField(Argument(name='a', shape=<ibis.expr.datashape.Columnar object at 0x109e48b90>, dtype={'alert_name': string, 'alert_triggered': boolean, 'alert_suppression_data': array<struct<name: string>>}), field='alert_triggered'), param=Argument(name='a', shape=<ibis.expr.datashape.Columnar object at 0x109e48b90>, dtype={'alert_name': string, 'alert_triggered': boolean, 'alert_suppression_data': array<struct<name: string>>}))
r3 := Filter[r2]
ArrayLength(r2._alerts) > 0
r4 := Project[r3]
DeviceName: r3.DeviceName
ProcessCommandLine: r3.ProcessCommandLine
alert: Unnest(r3._alerts)
r5 := Project[r4]
DeviceName: r4.DeviceName
ProcessCommandLine: r4.ProcessCommandLine
alert: r4.alert
alert_suppressed: StringContains(haystack=Lowercase(r4.DeviceName), needle='lab-')
Project[r5]
DeviceName: r5.DeviceName
ProcessCommandLine: r5.ProcessCommandLine
alert_name: StructField(r5.alert, field='alert_name')
alert_suppressed: r5.alert_suppressedSELECT
"t3"."DeviceName",
"t3"."ProcessCommandLine",
"t3"."alert"."alert_name" AS "alert_name",
CONTAINS(LOWER("t3"."DeviceName"), 'lab-') AS "alert_suppressed"
FROM (
SELECT
"t2"."DeviceName",
"t2"."ProcessCommandLine",
UNNEST("t2"."_alerts") AS "alert"
FROM (
SELECT
*
FROM (
SELECT
"t0"."Timestamp",
"t0"."DeviceId",
"t0"."DeviceName",
"t0"."InitiatingProcessFileName",
"t0"."FileName",
"t0"."ProcessCommandLine",
"t0"."RemoteIP",
"t0"."RemotePort",
"t0"."ActionType",
"t0"."ReportId",
LIST_FILTER(
CAST([
{'alert_name': 'CDC_373_AppleScriptToPython', 'alert_triggered': (
"t0"."InitiatingProcessFileName" = 'osascript'
)
AND LOWER("t0"."FileName") IN ('sh', 'bash')
AND CONTAINS(LOWER("t0"."ProcessCommandLine"), 'python &'), 'alert_suppression_data': LIST_FILTER(
CAST([{'name': 'known_device_lab'}] AS STRUCT("name" TEXT)[]),
__ibis_param__s__ -> TRUE
)}
] AS STRUCT(
"alert_name" TEXT,
"alert_triggered" BOOLEAN,
"alert_suppression_data" STRUCT("name" TEXT)[]
)[]),
__ibis_param_a__ -> (
__ibis_param_a__
)."alert_triggered"
) AS "_alerts"
FROM "ibis_pandas_memtable_fwbzej7h3rc4tcwr4dzokn6q6q" AS "t0"
) AS "t1"
WHERE
ARRAY_LENGTH("t1"."_alerts") > 0
) AS "t2"
) AS "t3"
Two suppressions evaluated in the same scan as the rule. The SCCM event fires the rule but also matches the suppress_sccm_hosts predicate, so it lands in the signals table with alert_suppressed=true and the matched suppression name in alert_suppression_data. The suspicious endpoint event fires the rule and matches no suppression, so it lands as actionable. Filtering vs. marking is a downstream decision left to the RBA aggregator or triage UI.
Long suppression lists - hundreds of known hashes, a full inventory of IT management hosts, a rolling feed of known-good signed binaries - do not belong in source code. The known_device and known_hash builders accept either an inline list or a table reference. When given a table reference, the builder emits an ibis semi_join against that table. The join compiles to the same backends as the rest of the expression tree. DuckDB locally reads the lookup table from a parquet file or a DuckDB relation. Spark reads it from a Delta table. The suppression predicate stays one line in the code regardless of how many entries the list has.
This also means the lookup table itself is a data artifact with its own lineage. Updates to the known-good hash list are commits to a Delta table with a timestamp, not edits to a YAML file buried in a config repo. Querying the signals table for alert_suppressed=true events where the suppression reason is "known-good hash" and the hash is no longer in the current lookup table immediately surfaces stale entries.
Fig 6: The suppression flow. A single event is evaluated first by the rule predicate, then by each suppression predicate in the same scan pass. Both actionable and suppressed alerts are written to the signals table - nothing is dropped. The two output buckets are fed by the same arrow. Only the alert_suppressed flag differs. Hover each stage for what happens there.
SIEMs and other non-SQL engines
Not every engine in a detection engineering environment speaks SQL. Splunk speaks SPL. Microsoft Defender XDR and Microsoft Sentinel speak KQL. ibis compiles to neither today, so rules written against Splunk telemetry cannot be expressed directly as ibis expressions the way rules against a Delta table can.
The gap is real, but the bridge pattern from the incident response notebooks post covers it. A typed decorator runs the native query, handles authentication and async submission, and lands the result as an ibis table registered in the shared DuckDB connection. Once the SIEM data is sitting in DuckDB as an ibis table, the rest of the detection platform - composition, signals, suppression - operates on it exactly the same way it operates on a table that came from a Delta scan or a Snowflake query.
The bridge decoratorThe decorator pattern is covered in detail in the IR notebooks post, but the shape is worth restating in this context. You write a function that returns SPL, annotate it with a schema dict, and the decorator handles the async Splunk job lifecycle and casts the result into properly typed columns. The decorated function returns an ibis table, not a raw DataFrame, so the downstream ibis rule sees it as just another table, interchangeable with any other source.
The detection-engineering implication is that SIEM-native telemetry becomes one more composable source in the same pipeline. A rule that joins Splunk endpoint telemetry against a Snowflake vulnerability table and a DuckDB-registered threat-intel feed is the same kind of ibis expression as a rule that reads only from a Delta lakehouse table. The SIEM bridge is a data-source concern, not a rule-logic concern. The rule does not know where its table came from.
The splunk atomThe cell below shows the bridge atom: a mock @spl.df decorator (the same one used in the IR notebooks post) runs a Splunk-style query, and the result is registered in DuckDB as an ibis table. An ibis rule is then applied on top of it directly.
r0 := InMemoryTable
data:
PandasDataFrameProxy:
_time host dest_domain bytes_out
0 2026-06-18 14:22:00 WS-FINANCE-03 raw.githubusercontent.com 4200
1 2026-06-18 14:24:00 WS-OPS-01 api.github.com 1800
2 2026-06-18 14:26:00 lab-test-01 pypi.org 12500
Filter[r0]
r0.bytes_out > 5000SELECT
*
FROM "ibis_pandas_memtable_f63rpa2txnfinbcyoe5esxhh5y" AS "t0"
WHERE
"t0"."bytes_out" > 5000
The SIEM becomes another data source. The @spl.df decorator runs the SPL query and lands the result as an ibis table in the shared DuckDB connection. The ibis rule applied on top is the same predicate you would write against a Delta table - the source is abstracted away at the decorator boundary. SPL handles what only SPL can do (index routing, bloom-filter tokens, accelerated datamodels), and ibis handles the rest.
The one thing this pattern does not provide is a shared compile target. A rule written against Splunk data through the bridge lives in Python, not in SPL. The bridge runs native SPL at the source to do the initial filter and projection, then ibis takes over for join logic, composable predicates, and the fused multi-rule pipeline. If you need the full detection to run as a scheduled Splunk saved search, the rule has to be translated back into SPL by hand. That is the honest trade-off: you get composition and the signals infrastructure in exchange for giving up the ability to push the ibis expression natively into Splunk’s scheduler.
KQL follows the same bridge pattern. A @kql.df decorator wraps Advanced Hunting in Defender XDR or a Sentinel query in the same way. The result lands in DuckDB as an ibis table, and from that point the rule is identical to any other rule in the registry. Both SIEM bridges share auth, async job handling, and schema enforcement with their counterparts in the IR notebooks SDK, so they reuse code rather than duplicating it.
Testing, lightly
The inner loopThe key property of ibis on DuckDB for testing is speed. A detection rule that runs in two milliseconds locally has an inner loop short enough to put a TP test and a TN test directly next to the rule, check them on every commit, and get an answer before the CI job has even scheduled a container. No JVM, no cluster, no fixture setup beyond a list of Python dicts. The test is a function that hands two synthetic events to the rule and asserts one fires and the other does not.
Here is the full shape of a per-rule test fixture. The events are plain Python dicts. Ibis converts them to a DuckDB table at assertion time.
import ibis
from detection_ibis import apple_script_to_python
con = ibis.duckdb.connect()
tp_event = {
"InitiatingProcessFileName": "osascript",
"FileName": "sh",
"ProcessCommandLine": "sh -c 'python & ...'",
}
tn_event = {
"InitiatingProcessFileName": "bash",
"FileName": "python3",
"ProcessCommandLine": "python3 script.py",
}
for label, event, expect in [("TP", tp_event, True), ("TN", tn_event, False)]:
tbl = con.create_table(f"evt_{label}", [event], overwrite=True)
result = tbl.filter(apple_script_to_python(tbl)).count().execute()
assert (result > 0) == expect, f"{label} failed"One true positive, one true negative, one assert per case. The fixture is a flat list of dicts, so adding a new edge-case row is a one-line change. Because the whole thing runs on DuckDB in-process, this runs in under 50ms and requires nothing beyond a Python environment with ibis installed. The test file lives next to the rule file in the same directory.
When the same rule has to work on both DuckDB (local CI, development) and Spark (production lakehouse), a small cross-backend regression subset matters. The same fixture runs the rule against both connections and asserts the output sets match. This catches tumble_window_agg dispatch drift, dialect-level differences in string matching, and the occasional ibis function that landed on one backend before the other. The cross-backend run does not need to cover every rule - only the ones that are actually deployed to both a batch and a streaming target. The rest run DuckDB-only and that is sufficient.
The marimo island below shows a complete TP / TN test for the osascript rule from §3, running live against synthetic events.
A self-contained TP / TN harness for one detection rule. The cell runs the rule against a true-positive event (osascript spawning a shell that pipes to Python) and a true-negative event (a normal Python invocation) and reports PASS or FAIL for each case. This is what a detection PR looks like in this design: the rule and its acceptance test ship together, and CI executes both in under a second.
The catalog layer
Iceberg and Delta as the storage contractMost of the platform design above treats “the table” as an abstract ibis handle. In practice, two open formats dominate the lakehouse storage tier: Apache Iceberg and Delta Lake. Both provide ACID transactions, time travel, hidden partitioning, and schema evolution. Both are readable from DuckDB locally via iceberg_scan(...) and delta_scan(...). The choice between them is usually made at the infrastructure level, not the detection-engineering level - but it is worth knowing how they differ so you can read a production telemetry table locally without standing up a cluster.
DuckDB can scan either format directly from a local snapshot or from object storage. The code below shows how to bind a production Iceberg snapshot to an ibis table and run a rule against it with no Spark or Snowflake cluster involved.
import ibis
con = ibis.duckdb.connect()
# Point at a local snapshot exported from the lakehouse
events = con.read_parquet("snapshots/DeviceProcessEvents/**/*.parquet")
# Or scan the Iceberg table directly from S3 / Azure Blob
events = con.raw_sql(
"SELECT * FROM iceberg_scan('s3://telemetry-bucket/DeviceProcessEvents')"
).fetchdf()
# Bind as an ibis table and apply a rule
t = ibis.memtable(events)
matches = t.filter(apple_script_to_python(t))
matches.execute()Both iceberg_scan and delta_scan are DuckDB extensions installed on first use. Reading from object storage requires the relevant DuckDB secret (S3 credentials, Azure SAS token) set via CREATE SECRET. Scanning a snapshot locally - by exporting a date-bounded parquet slice from the lakehouse and referencing the path - removes the credential requirement entirely and is the safer pattern for developer laptops. The rule code is identical in both cases. Only the table source changes.
The catalog layer sits above the file format and governs what tables exist, where their data files are, and which engines can access them. Three catalogs are relevant here. Polaris is the open-source Iceberg catalog backed by Snowflake and a broad vendor coalition - its REST catalog API is now the closest thing the Iceberg ecosystem has to a standard. Unity Catalog is Databricks’ catalog, recently open-sourced, and cross-format: it manages both Delta and Iceberg tables under one namespace. Project Nessie is a catalog with git-like versioning semantics for Iceberg, useful when you want branch-per-detection-PR semantics on the data side.
For a detection engineering team, the catalog choice has one concrete implication: if your telemetry lives in an Iceberg-registered Polaris catalog and your signals table lives in a Unity-managed Delta table, a single ibis expression can span both. The source read goes through the Iceberg path, the signals write goes through the Delta write path in Spark or Trino, and a local delta_scan gives you the written signals back in DuckDB for development. Nothing in the rule code changes.
Fig 7: The catalog and engine landscape around a detection platform. Source telemetry sits in Iceberg or Delta tables, registered in one or more catalogs. Detection rules read from those tables via whichever engine holds the data. Signals are written back to a Delta or Iceberg sink. DuckDB provides a local read path for both formats, so the whole loop is accessible from a laptop without a running cluster.
Limitations
This design works well for the cases it targets, but several rough edges are worth naming directly.
Compile-layer gapswindow_by().tumble(). The tumble_window_agg helper dispatches to date-truncation plus group-by on those backends and to real tumbling windows on Spark, Flink, and RisingWave. The rule code stays clean, but the two paths can drift: a window-size change on one backend has to be verified against the other, because the aggregation semantics are subtly different at boundaries.is_streaming_backend(t) check uses a hard-coded set of backend names (flink, pyspark, risingwave) to decide whether to emit real watermarks. If ibis adds a new streaming backend, the dispatch falls through to the batch path silently rather than failing loudly.backend.supports_watermark) once ibis exposes backend feature flags. Until then, the name-based set is the pragmatic option and should be treated as a known debt item.approx_count_distinct directly. Others need a specific ibis approx_nunique call that may or may not lower correctly. Rare or specialized aggregations are still the most likely surface for cross-backend surprises.iceberg_scan and delta_scan. Writing into a transactional Iceberg or Delta table at scale still belongs to the engine that owns the table (Spark, Trino, Snowflake). For local development this is fine, but it means the signals-write step in a production job requires a different engine than the rules evaluation step if you want full ACID semantics on the signals table.NotImplementedError rather than a compile-time guarantee. The DuckDB path is usually ahead because the ibis team uses DuckDB as the reference backend.Summary
This post has walked through one way of expressing detection logic against a multi-engine, multi-format telemetry landscape. The core claim is that a detection rule is a boolean expression over a table, and that ibis can compile one such expression to whichever engine holds the data at query time - DuckDB locally, Spark or Flink for streaming, Snowflake or ClickHouse for batch, with thin bridges for SIEMs that speak SPL or KQL. The same primitives compose into a full detection platform: fused multi-rule scans, a queryable signals table, and suppression logic that marks rather than drops.
The key ideas, taken together:
- One expression, many engines. The predicate is written once as ibis. Swapping the connection swaps the compile target. Rules no longer drift independently per dialect.
- Batch and streaming share the same code.
tumble_window_agg(...)dispatches to real watermarks on Flink or Spark Structured Streaming, and to group-by aggregation on DuckDB or Snowflake. The rule author controls the window size and watermark. The backend controls which path activates. - The signals table is a queryable artifact. Detection output lands in Delta or Iceberg, not just an alert stream. Risk-based aggregation, triage UIs, and BI dashboards all read from the same table. Detection-of-detections is another ibis rule on top.
- Whitelists are code next to the rule. A
@suppressionexpression lives in the same repository and review loop as the detection it qualifies. Suppressed alerts are marked and written to the signals table withalert_suppressed=true. They are not dropped. - The local development story is DuckDB. Per-rule tests are millisecond-scale, with no cluster to stand up. The production story is whichever engine owns the data.
References and related work
- ibis - Python dataframe API that compiles one expression tree to many query backends. The substrate for everything in this post.
- DuckDB - in-process analytical query engine. Used here as the local development backend and the in-browser engine behind the interactive cells.
- Apache Iceberg - open table format with broad multi-engine support, time-travel queries, and hidden partitioning. One of the two storage formats discussed for the signals table.
- Delta Lake - open table format with ACID transactions and native Spark integration. The other signals-table candidate, preferred in Databricks-anchored environments.
- Apache Spark Structured Streaming - the reference streaming target for windowed ibis rules.
- Apache Flink - stateful stream processing engine, covered as an alternative streaming backend for
tumble_window_agg. - RisingWave - streaming SQL with materialised views, mentioned as an additional ibis streaming target.
- Snowflake - warehouse-scale batch backend. ibis compiles the same rule predicate to Snowflake SQL without changes.
- ClickHouse - real-time analytical engine, suited to network-flow detections at high ingest rates.
- Trino - federated query engine across Iceberg, Delta, and external sources. Mentioned for multi-catalog deployments.
- BigQuery - GCP warehouse backend. One of the ibis compile targets for batch detections.
- marimo - reactive Python notebook used for the interactive islands embedded in this post.
- marimo case study: DNB - earlier write-up of the same team’s work on notebook-driven incident response, published on the marimo blog.
- Anton Chuvakin on detection as code - the framing that detection rules belong in version-controlled code, not GUI editors or ad-hoc query strings.
- Sigma - YAML rule format that targets many SIEMs through translation. A different shape from this post: Sigma translates one rule to many dialects. Ibis compiles one expression to many backends. Both solve the drift problem, at different layers.
- Microsoft Sentinel notebooks - Jupyter-based threat hunting in Azure. Adjacent prior art on notebook-driven security operations.
- Incident response notebooks - sibling post on this blog covering the ibis and marimo substrate applied to the IR side, including the Splunk and Defender bridge decorators referenced in the SIEM section above.