Real-Time Security Detection with Streaming SQL: Credential Stuffing, DDoS, Lateral Movement, and Log Analytics

Real-time security detection with streaming SQL means authentication events, sFlow samples, HTTP access logs, and application logs are evaluated as they arrive in materialized views that count failures per account, distinct accounts per source, bytes per destination, requests per endpoint, distinct hosts per session, and errors per service, so brute force, credential stuffing, volumetric and application-layer DDoS, lateral movement, and service degradation each surface within seconds instead of after the next SIEM batch. This guide builds the detection layer on RisingWave, a PostgreSQL-compatible streaming database, starting with authentication attacks and extending to DDoS, lateral movement, and log analytics without an Elasticsearch indexing delay.

Why Detection Has to Be Continuous

The window for authentication attacks is measured in minutes. Fraud prevention research consistently shows attackers who validate stolen credentials begin exploiting accounts within 15 minutes of a positive login, and a credential stuffing campaign completes its bursts and moves on before a batch job runs. An attacker moving laterally across five machines in eight minutes finishes reconnaissance, and possibly exfiltration, before the next 15-minute scheduled query starts. A volumetric flood that saturates a link is observed by NetFlow 30 to 60 seconds after the fact, plus whatever the collector and the polling interval add. Logs in Elasticsearch are not searchable until they pass through parsing, tokenizing, and Lucene segment writes, typically 30 seconds to 2 minutes under production load and longer during the ingestion spikes that accompany incidents.

Batch detection also forces binary responses. Real-time detection enables graduated ones: a step-up MFA challenge the moment a pattern emerges, preserving the experience for legitimate users while stopping the attacker. RisingWave maintains each detection rule as an incrementally updated materialized view. When a new event arrives, only the affected window or session is recalculated; late events retract and correct the affected result automatically. Analysts write the rules in SQL, the SIEM or alert service reads pre-computed results over PostgreSQL, and sinks push alerts to Kafka, PagerDuty, Slack, or a webhook. RisingWave is open source under Apache 2.0, written in Rust, and checkpoints state to S3, so detection state survives restarts with exactly-once semantics.

Sources

Authentication events from the application, sFlow samples from the collector (sflowtool or GoFlow2), HTTP access logs from Filebeat or Fluentd, internal authentication logs from Windows Event IDs 4624 and 4648 or SSH, and structured application logs, each in its own Kafka topic.

CREATE SOURCE auth_events (
    event_id        VARCHAR,
    event_time      TIMESTAMPTZ,
    event_type      VARCHAR,        -- 'AUTH_SUCCESS' | 'AUTH_FAILURE'
    account_id      VARCHAR,
    source_ip       VARCHAR,
    user_agent      VARCHAR,
    ja3_hash        VARCHAR,        -- TLS fingerprint, if the edge records it
    failure_reason  VARCHAR         -- 'INVALID_PASSWORD' | 'ACCOUNT_LOCKED' | 'UNKNOWN_USER'
) WITH (connector = 'kafka', topic = 'auth-events', properties.bootstrap.server = 'kafka:9092', scan.startup.mode = 'latest')
FORMAT PLAIN ENCODE JSON;

CREATE SOURCE sflow_records (
    src_ip           VARCHAR,
    dst_ip           VARCHAR,
    src_port         INT,
    dst_port         INT,
    protocol         INT,
    bytes_sampled    BIGINT,
    packets_sampled  INT,
    sampling_rate    INT,
    ingestion_time   TIMESTAMPTZ
) WITH (connector = 'kafka', topic = 'sflow-records', properties.bootstrap.server = 'kafka:9092', scan.startup.mode = 'latest')
FORMAT PLAIN ENCODE JSON;

CREATE SOURCE http_access_logs (
    src_ip          VARCHAR,
    endpoint        VARCHAR,
    method          VARCHAR,
    status_code     INT,
    response_bytes  BIGINT,
    request_id      VARCHAR,
    request_time    TIMESTAMPTZ
) WITH (connector = 'kafka', topic = 'http-access-logs', properties.bootstrap.server = 'kafka:9092', scan.startup.mode = 'latest')
FORMAT PLAIN ENCODE JSON;

CREATE SOURCE internal_auth_events (
    src_host     VARCHAR,
    dest_host    VARCHAR,
    account      VARCHAR,
    auth_method  VARCHAR,     -- 'NTLM', 'KERBEROS', 'SSH', 'RDP'
    status       VARCHAR,     -- 'success', 'failure'
    ts           TIMESTAMPTZ
) WITH (connector = 'kafka', topic = 'auth_logs', properties.bootstrap.server = 'kafka:9092', scan.startup.mode = 'latest')
FORMAT PLAIN ENCODE JSON (timestamptz.handling.mode = 'utc_without_suffix');

CREATE SOURCE log_events (
    event_time    TIMESTAMPTZ,
    service_name  VARCHAR,
    log_level     VARCHAR,
    message       VARCHAR,
    host          VARCHAR,
    trace_id      VARCHAR
) WITH (connector = 'kafka', topic = 'application-logs', properties.bootstrap.server = 'kafka:9092', scan.startup.mode = 'latest')
FORMAT PLAIN ENCODE JSON;

CREATE TABLE asset_sensitivity (hostname VARCHAR PRIMARY KEY, sensitivity VARCHAR);   -- 'critical', 'high', 'medium', 'low'
CREATE TABLE service_accounts (account VARCHAR PRIMARY KEY, owner VARCHAR, purpose VARCHAR);
CREATE TABLE ip_allowlist (source_ip VARCHAR PRIMARY KEY, reason VARCHAR);

Brute Force vs Credential Stuffing

The two are often confused and require different detection units. A brute force attack concentrates on one account with many password guesses; the signal is many failures on one account in a short window, and a per-account rate limit catches most of it. Credential stuffing replays username and password pairs from previous breaches, one or two tries per account across thousands of accounts, through rotating proxies or botnets. Per-account limits are blind to it. You have to detect the campaign through the attacker's shared infrastructure signal: source IP, subnet, user agent, or TLS fingerprint.

Brute force is a count per account in a five-minute window. TUMBLE gives clock-aligned non-overlapping windows; HOP gives a five-minute window advancing every minute so an attack straddling a boundary is still caught.

CREATE MATERIALIZED VIEW brute_force_alerts AS
SELECT account_id, COUNT(*) AS failed_attempts, MIN(event_time) AS first_attempt, MAX(event_time) AS last_attempt,
       COUNT(DISTINCT source_ip) AS source_ips, window_start, window_end
FROM HOP(auth_events, event_time, INTERVAL '1 minute', INTERVAL '5 minutes')
WHERE event_type = 'AUTH_FAILURE'
GROUP BY account_id, window_start, window_end
HAVING COUNT(*) > 5;

Credential stuffing changes the GROUP BY from the account to the shared signal and counts distinct accounts. ARRAY_AGG carries the affected accounts so the response can protect them directly. A subnet view catches campaigns spread across a /24, and grouping by JA3 or user agent catches IP rotation.

CREATE MATERIALIZED VIEW credential_stuffing_alerts AS
SELECT source_ip, user_agent, COUNT(DISTINCT account_id) AS accounts_targeted, COUNT(*) AS total_failures,
       COUNT(*) FILTER (WHERE failure_reason = 'UNKNOWN_USER') AS unknown_user_failures,
       ARRAY_AGG(DISTINCT account_id) AS affected_accounts, window_start, window_end
FROM TUMBLE(auth_events, event_time, INTERVAL '10 minutes')
WHERE event_type = 'AUTH_FAILURE'
GROUP BY source_ip, user_agent, window_start, window_end
HAVING COUNT(DISTINCT account_id) > 10;

CREATE MATERIALIZED VIEW subnet_stuffing_alerts AS
SELECT SPLIT_PART(source_ip, '.', 1) || '.' || SPLIT_PART(source_ip, '.', 2) || '.' || SPLIT_PART(source_ip, '.', 3) || '.0/24' AS ip_subnet,
       COUNT(DISTINCT account_id) AS accounts_targeted, COUNT(DISTINCT source_ip) AS source_ips, COUNT(*) AS total_failures, window_start, window_end
FROM TUMBLE(auth_events, event_time, INTERVAL '10 minutes')
WHERE event_type = 'AUTH_FAILURE'
GROUP BY 1, window_start, window_end
HAVING COUNT(DISTINCT account_id) > 20;

CREATE MATERIALIZED VIEW fingerprint_stuffing_alerts AS
SELECT ja3_hash, COUNT(DISTINCT account_id) AS accounts_targeted, COUNT(DISTINCT source_ip) AS source_ips, COUNT(*) AS total_failures, window_start, window_end
FROM TUMBLE(auth_events, event_time, INTERVAL '10 minutes')
WHERE event_type = 'AUTH_FAILURE' AND ja3_hash IS NOT NULL
GROUP BY ja3_hash, window_start, window_end
HAVING COUNT(DISTINCT account_id) > 10 AND COUNT(DISTINCT source_ip) > 5;

The most dangerous event in a stuffing campaign is the success. Joining successes to the campaign view in the same window gives the accounts that must be forced through MFA or a password reset right now.

CREATE MATERIALIZED VIEW compromised_account_candidates AS
SELECT a.account_id, a.source_ip, a.event_time AS success_time, c.accounts_targeted, c.total_failures
FROM auth_events a
JOIN credential_stuffing_alerts c ON a.source_ip = c.source_ip AND a.event_time >= c.window_start AND a.event_time < c.window_end + INTERVAL '10 minutes'
WHERE a.event_type = 'AUTH_SUCCESS';

A SIEM polling brute_force_alerts with WHERE window_end > NOW() - INTERVAL '1 minute' gets instant results from pre-computed state with no re-aggregation.

DDoS: Volumetric Floods from sFlow and HTTP Floods from Access Logs

sFlow samples 1 in N packets at wire speed and exports immediately, so traffic data arrives within seconds. NetFlow accumulates per-flow records in a cache and exports on expiry or a timer, 30 to 60 seconds later. For detection that drives automated mitigation, sFlow is the right input; NetFlow's richer flow metadata belongs in trending and post-incident forensics. Three attack classes map to SQL: volumetric floods appear as a byte spike per destination; amplification attacks appear as a high ratio of response to request bytes per source; application-layer HTTP floods are invisible in byte volume and appear in access logs as request rate per source and endpoint.

CREATE MATERIALIZED VIEW dst_bytes_10s AS
SELECT window_start, window_end, dst_ip, SUM(bytes_sampled * sampling_rate) AS estimated_bytes,
       SUM(packets_sampled * sampling_rate) AS estimated_packets, COUNT(*) AS sample_count, COUNT(DISTINCT src_ip) AS distinct_sources
FROM TUMBLE(sflow_records, ingestion_time, INTERVAL '10 seconds')
GROUP BY window_start, window_end, dst_ip;

CREATE MATERIALIZED VIEW dst_baseline_24h AS
SELECT dst_ip, AVG(estimated_bytes) AS avg_bytes_10s, MAX(estimated_bytes) AS max_bytes_10s
FROM dst_bytes_10s
WHERE window_end > NOW() - INTERVAL '24 hours'
GROUP BY dst_ip;

CREATE MATERIALIZED VIEW volumetric_flood_alerts AS
SELECT c.window_start, c.window_end, c.dst_ip, c.estimated_bytes, c.estimated_packets, c.distinct_sources, b.avg_bytes_10s,
       c.estimated_bytes / NULLIF(b.avg_bytes_10s, 0) AS deviation_ratio
FROM dst_bytes_10s c
LEFT JOIN dst_baseline_24h b ON c.dst_ip = b.dst_ip
WHERE c.estimated_bytes > 1000000000                                    -- 1 GB in 10 seconds, absolute floor
   OR c.estimated_bytes > 5 * COALESCE(b.avg_bytes_10s, 1e18);           -- or five times the destination's baseline

CREATE MATERIALIZED VIEW amplification_alerts AS
SELECT window_start, window_end, dst_ip AS victim_ip, src_port AS amplifier_port,
       SUM(bytes_sampled * sampling_rate) AS inbound_bytes, SUM(packets_sampled * sampling_rate) AS inbound_packets,
       SUM(bytes_sampled * sampling_rate) / NULLIF(SUM(packets_sampled * sampling_rate), 0) AS avg_packet_bytes, COUNT(DISTINCT src_ip) AS reflector_count
FROM TUMBLE(sflow_records, ingestion_time, INTERVAL '10 seconds')
WHERE protocol = 17 AND src_port IN (53, 123, 1900, 11211, 389)        -- DNS, NTP, SSDP, memcached, LDAP
GROUP BY window_start, window_end, dst_ip, src_port
HAVING COUNT(DISTINCT src_ip) > 50 AND SUM(bytes_sampled * sampling_rate) > 100000000;

HTTP floods follow the same shape on a different source, and the allowlist join removes high-traffic service accounts. A low-and-slow flood spread across many sources is caught at the endpoint level rather than per source.

CREATE MATERIALIZED VIEW http_flood_alerts AS
SELECT l.window_start, l.window_end, l.src_ip, l.endpoint, COUNT(*) AS request_count, COUNT(DISTINCT l.request_id) AS unique_requests,
       COUNT(*) FILTER (WHERE l.status_code >= 400) AS error_responses
FROM TUMBLE(http_access_logs, request_time, INTERVAL '1 minute') l
LEFT JOIN ip_allowlist w ON l.src_ip = w.source_ip
WHERE w.source_ip IS NULL
GROUP BY l.window_start, l.window_end, l.src_ip, l.endpoint
HAVING COUNT(*) > 500;

CREATE MATERIALIZED VIEW endpoint_rate_1m AS
SELECT window_start, window_end, endpoint, COUNT(*) AS request_count, COUNT(DISTINCT src_ip) AS distinct_sources
FROM TUMBLE(http_access_logs, request_time, INTERVAL '1 minute')
GROUP BY window_start, window_end, endpoint;

CREATE MATERIALIZED VIEW endpoint_flood_alerts AS
SELECT c.window_start, c.window_end, c.endpoint, c.request_count, c.distinct_sources, b.avg_requests_1m, c.request_count / NULLIF(b.avg_requests_1m, 0) AS deviation_ratio
FROM endpoint_rate_1m c
JOIN (SELECT endpoint, AVG(request_count) AS avg_requests_1m FROM endpoint_rate_1m WHERE window_end > NOW() - INTERVAL '1 hour' GROUP BY endpoint) b
  ON c.endpoint = b.endpoint
WHERE c.request_count > 5 * b.avg_requests_1m AND c.request_count > 1000;

With a NetFlow pipeline taking 90 seconds from traffic to alert, 60 of export and 30 of polling, RisingWave removes the analytics half and lands around 60 to 65 seconds. With sFlow it lands in the low seconds, which is what automated mitigation triggers need.

Lateral Movement: Session Windows Over Internal Authentication

After compromising one host, an attacker pivots by authenticating to one host after another, harvesting credentials along the way. Each hop is an ordinary successful logon, and no single event is suspicious. The pattern is the same account reaching many distinct destination hosts within minutes. Session windows fit this exactly: the window extends as long as activity continues and closes after a gap of inactivity, so a burst of hops is one window regardless of clock alignment. Enriching with asset sensitivity and excluding service accounts turns the raw count into a usable alert.

CREATE MATERIALIZED VIEW lateral_movement_alerts AS
WITH session_agg AS (
    SELECT account, src_host, window_start, window_end,
           COUNT(DISTINCT dest_host) AS unique_dest_hosts, COUNT(*) AS auth_events,
           ARRAY_AGG(DISTINCT dest_host) AS dest_host_list
    FROM SESSION((SELECT * FROM internal_auth_events WHERE status = 'success'), ts, INTERVAL '5 minutes')
    GROUP BY account, src_host, window_start, window_end
)
SELECT sa.account, sa.src_host, sa.window_start, sa.window_end, sa.unique_dest_hosts, sa.auth_events, sa.dest_host_list,
       (SELECT MAX(CASE a.sensitivity WHEN 'critical' THEN 4 WHEN 'high' THEN 3 WHEN 'medium' THEN 2 ELSE 1 END)
        FROM asset_sensitivity a WHERE a.hostname = ANY(sa.dest_host_list)) AS max_asset_sensitivity_rank
FROM session_agg sa
LEFT JOIN service_accounts s ON s.account = sa.account
WHERE s.account IS NULL
  AND sa.auth_events >= 3
  AND (sa.unique_dest_hosts >= 5
       OR (SELECT COUNT(*) FROM asset_sensitivity a WHERE a.hostname = ANY(sa.dest_host_list) AND a.sensitivity = 'critical') >= 2);

CREATE MATERIALIZED VIEW credential_spray_internal AS
SELECT src_host, window_start, window_end, COUNT(DISTINCT account) AS accounts_tried, COUNT(DISTINCT dest_host) AS hosts_tried, COUNT(*) AS failures
FROM TUMBLE(internal_auth_events, ts, INTERVAL '10 minutes')
WHERE status = 'failure'
GROUP BY src_host, window_start, window_end
HAVING COUNT(DISTINCT account) >= 10;

Five distinct hosts in a five-minute session flags attackers and also administrators running deployment scripts, monitoring agents polling dozens of servers, and batch jobs touching storage nodes. Four adjustments keep the signal: tune the gap (five minutes is tight, ten still gives early detection with less noise), exclude service accounts through the join, weight by sensitivity so two critical hosts alert regardless of count, and require a minimum event count so a misconfigured client spraying connection attempts does not fire. dest_host_list gives the analyst the exact hop sequence. The companion credential_spray_internal view catches the password-spray phase that usually precedes lateral movement, one source trying many accounts.

Log Analytics Without Elasticsearch

Elasticsearch stores raw documents and builds an inverted index for later search, with indexing latency of 30 seconds to 2 minutes and substantial storage for every tokenized field. RisingWave does the opposite: it maintains incrementally updated aggregations as events flow through, so an error rate per service per minute updates the moment the event arrives, and a view tracking 1,000 services holds 1,000 rows rather than billions of log lines. The trade is intentional. RisingWave is the right tool for continuous aggregation, alerting, and dashboards over structured fields; Elasticsearch remains the right tool for full-text search over arbitrary historical content and investigations nobody anticipated when the views were defined.

CREATE MATERIALIZED VIEW error_rate_per_service AS
SELECT service_name, window_start, window_end,
       COUNT(*) AS total_events, COUNT(*) FILTER (WHERE log_level = 'ERROR') AS error_count,
       ROUND(COUNT(*) FILTER (WHERE log_level = 'ERROR') * 100.0 / NULLIF(COUNT(*), 0), 2) AS error_rate_pct,
       COUNT(DISTINCT host) AS hosts_reporting
FROM TUMBLE(log_events, event_time, INTERVAL '1 minute')
GROUP BY service_name, window_start, window_end;

CREATE MATERIALIZED VIEW error_rate_alerts AS
SELECT cur.service_name, cur.window_start, cur.error_rate_pct, prev.error_rate_pct AS prior_error_rate_pct, cur.error_count
FROM error_rate_per_service cur
JOIN error_rate_per_service prev ON cur.service_name = prev.service_name AND prev.window_start = cur.window_start - INTERVAL '1 minute'
WHERE cur.error_rate_pct > 5 AND cur.error_rate_pct > 3 * GREATEST(prev.error_rate_pct, 0.1) AND cur.total_events > 100;

CREATE MATERIALIZED VIEW host_health_5m AS
SELECT host, window_start, window_end,
       COUNT(*) FILTER (WHERE log_level IN ('ERROR', 'FATAL')) AS severe_events,
       COUNT(*) FILTER (WHERE message LIKE '%OOM%' OR message LIKE '%OutOfMemory%') AS memory_pressure_events,
       COUNT(*) FILTER (WHERE message LIKE '%connection refused%' OR message LIKE '%timeout%') AS network_error_events
FROM TUMBLE(log_events, event_time, INTERVAL '5 minutes')
GROUP BY host, window_start, window_end;

CREATE MATERIALIZED VIEW privilege_escalation_sequences AS
SELECT a.host, a.trace_id, a.event_time AS sudo_time, b.event_time AS change_time, b.message AS change_message
FROM log_events a
JOIN log_events b ON a.host = b.host AND a.trace_id = b.trace_id AND b.event_time BETWEEN a.event_time AND a.event_time + INTERVAL '2 minutes'
WHERE a.message LIKE '%sudo%' AND (b.message LIKE '%useradd%' OR b.message LIKE '%authorized_keys%' OR b.message LIKE '%crontab%');

CREATE MATERIALIZED VIEW audit_counts_daily AS
SELECT window_start::DATE AS day, service_name,
       COUNT(*) FILTER (WHERE message LIKE '%data_export%') AS data_exports,
       COUNT(*) FILTER (WHERE message LIKE '%access_granted%') AS access_events,
       COUNT(*) FILTER (WHERE message LIKE '%auth_attempt%') AS auth_attempts
FROM TUMBLE(log_events, event_time, INTERVAL '1 day')
GROUP BY window_start::DATE, service_name;

Grafana adds RisingWave as a standard PostgreSQL data source and uses window_start as the time field. Compliance counters serve audit queries instantly without scanning raw files, and the raw log topic retained in Kafka or archived to Iceberg through a sink covers the ad hoc retrieval that materialized views were never meant for.

Routing Alerts and Graduated Response

CREATE SINK auth_attack_sink AS
SELECT account_id AS subject, 'BRUTE_FORCE' AS alert_type, failed_attempts AS metric, window_end AS alert_time FROM brute_force_alerts
UNION ALL
SELECT source_ip, 'CREDENTIAL_STUFFING', accounts_targeted, window_end FROM credential_stuffing_alerts
UNION ALL
SELECT account_id, 'POSSIBLE_COMPROMISE', 1, success_time FROM compromised_account_candidates
WITH (connector = 'kafka', topic = 'security-alerts.auth', properties.bootstrap.server = 'kafka:9092') FORMAT PLAIN ENCODE JSON;

CREATE SINK ddos_sink AS
SELECT dst_ip AS target, 'VOLUMETRIC_FLOOD' AS alert_type, estimated_bytes AS metric, window_end AS alert_time FROM volumetric_flood_alerts
UNION ALL
SELECT victim_ip, 'AMPLIFICATION', inbound_bytes, window_end FROM amplification_alerts
UNION ALL
SELECT endpoint, 'HTTP_FLOOD', request_count, window_end FROM endpoint_flood_alerts
WITH (connector = 'kafka', topic = 'security-alerts.ddos', properties.bootstrap.server = 'kafka:9092') FORMAT PLAIN ENCODE JSON;

CREATE SINK lateral_movement_sink FROM lateral_movement_alerts
WITH (connector = 'kafka', topic = 'security-alerts.lateral', properties.bootstrap.server = 'kafka:9092', primary_key = 'account,src_host,window_start')
FORMAT UPSERT ENCODE JSON;

CREATE SINK service_health_sink FROM error_rate_alerts
WITH (connector = 'kafka', topic = 'ops-alerts.error-rate', properties.bootstrap.server = 'kafka:9092') FORMAT PLAIN ENCODE JSON;

CREATE SINK auth_alert_history FROM credential_stuffing_alerts
WITH (connector = 'jdbc', jdbc.url = 'jdbc:postgresql://siem-db:5432/security', table.name = 'stuffing_alert_history', type = 'append-only', force_append_only = 'true');

The auth alert topic drives graduated response: a stuffing campaign from a source triggers MFA step-up for every account in affected_accounts, a possible compromise triggers a forced reset, and a brute force target gets a temporary per-account throttle rather than a lockout that an attacker could weaponize against your own users. The DDoS topic feeds the mitigation controller, and the lateral movement topic opens an incident with the hop list attached.

Streaming SQL vs Flink and vs Batch SIEM Rules

Teams already running Flink can coexist with RisingWave on the same Kafka topics and let analysts own detection rules in SQL without involving Flink developers. Streaming SQL complements the SIEM rather than replacing it: it covers the detect-fast half, and the SIEM keeps case management and long-horizon forensics.

Frequently Asked Questions

How do I pick thresholds for brute force and stuffing views? Start from your own traffic: run the views without HAVING for a day, look at the distribution of failures per account and distinct accounts per source, and set thresholds above the legitimate tail. Then tighten the window or raise the threshold based on the false positive rate the response team reports.

Does the sFlow sampling rate make byte estimates unreliable? Multiplying sampled bytes by the sampling rate gives an unbiased estimate, and at flood volumes the sample count per 10-second window is large enough that the estimate is accurate to a few percent. Low-volume destinations are noisy, which is why the alert has an absolute floor as well as a baseline ratio.

Can lateral movement detection use failed authentications too? Yes. The credential_spray_internal view counts failures by source host. A source that fails against ten accounts and then succeeds and hops to five hosts is two views agreeing, and joining them on src_host and time gives a high-confidence alert.

What does RisingWave not do that Elasticsearch does? Full-text search over arbitrary historical content. Keep raw logs in Kafka with retention or archive them to Iceberg via a sink, and use RisingWave for the metrics, alerts, and counters you can define in advance.

How is state kept bounded? Windowed views retain state only for open windows plus allowed lateness, defined by a watermark on the source timestamp. Baseline views use a temporal filter such as NOW() - INTERVAL '24 hours'. Unwindowed aggregations grow with history and belong in a demo, not in production.