QuestDB Balance Query

This page documents the QuestDB SQL query used to compute the medical fluid balance over a given time window. The query is structured as a series of Common Table Expressions (CTEs), each building on the previous one. The sections below walk through each CTE in order.

---

Full Query Reference

Show full query

WITH buckets AS (
    SELECT
        generate_series AS bucket_ts,
        CASE
            WHEN to_timezone(generate_series, '${timezone}') >=
                 timestamp_floor('1d', to_timezone(generate_series, '${timezone}')) + ${resetOffsetUs}L
            THEN timestamp_floor('1d', to_timezone(generate_series, '${timezone}')) + ${resetOffsetUs}L
            ELSE timestamp_floor('1d', to_timezone(generate_series, '${timezone}')) + ${resetOffsetUs}L - 86400000000L
        END AS balance_day,
        greatest(
            generate_series - ${options.intervalUs}L,
            cast('${startStr}' AS timestamp)
        ) AS rate_period_start
    FROM generate_series(
        cast('${startStr}' AS timestamp),
        cast('${endAdjustedStr}' AS timestamp),
        '${options.interval}'
    )
),
discrete AS (
    SELECT
        b.bucket_ts AS bucket_ts,
        coalesce(e.in_val, 0.0) AS discrete_ml_input,
        coalesce(e.out_val, 0.0) AS discrete_ml_output
    FROM buckets b
    LEFT JOIN (
        SELECT
            timestamp_floor('${options.interval}', timestamp, cast('${startStr}' AS timestamp)) AS bucket_ts,
            sum(CASE WHEN value > 0 THEN value ELSE 0.0 END) AS in_val,
            sum(CASE WHEN value < 0 THEN -value ELSE 0.0 END) AS out_val
        FROM balance_events
        WHERE encounterId = '${options.encounterId}' AND balanceType = '${options.balanceType}'
          AND timestamp >= cast('${startStr}' AS timestamp)
          AND timestamp <  cast('${endStr}' AS timestamp) ${fhirReferenceFilter}
        GROUP BY 1
    ) e ON e.bucket_ts = b.bucket_ts
),
seed_rates AS (
    (
        SELECT
            fhirResourceId,
            timestamp AS ts,
            value AS rate_ml_per_h
        FROM balance_rates
        WHERE encounterId = '${options.encounterId}' AND balanceType = '${options.balanceType}'
          AND timestamp < cast('${startStr}' AS timestamp) ${fhirReferenceFilter}
        LATEST ON timestamp PARTITION BY fhirResourceId
    )
    WHERE rate_ml_per_h <> 0
),
range_rates AS (
    SELECT
        fhirResourceId,
        timestamp AS ts,
        value AS rate_ml_per_h
    FROM balance_rates
    WHERE encounterId = '${options.encounterId}' AND balanceType = '${options.balanceType}'
      AND timestamp >= cast('${startStr}' AS timestamp)
      AND timestamp <  cast('${endStr}' AS timestamp) ${fhirReferenceFilter}
),
rate_events AS (
    SELECT * FROM seed_rates
    UNION ALL
    SELECT * FROM range_rates
),
rate_segments_raw AS (
    SELECT
        fhirResourceId,
        ts AS seg_start,
        lead(ts) OVER (
            PARTITION BY fhirResourceId
            ORDER BY ts
        ) AS next_ts,
        rate_ml_per_h
    FROM rate_events
),
rate_segments AS (
    SELECT
        fhirResourceId,
        seg_start,
        coalesce(next_ts, cast('${endStr}' AS timestamp)) AS seg_end,
        rate_ml_per_h
    FROM rate_segments_raw
),
infusions AS (
    SELECT
        b.bucket_ts AS bucket_ts,
        sum(
            CASE WHEN rs.rate_ml_per_h > 0 THEN
                rs.rate_ml_per_h * greatest(0L, datediff('s', greatest(rs.seg_start, b.rate_period_start), least(rs.seg_end, b.bucket_ts))) / 3600.0
            ELSE 0.0 END
        ) AS infusion_ml_input,
        sum(
            CASE WHEN rs.rate_ml_per_h < 0 THEN
                (-rs.rate_ml_per_h) * greatest(0L, datediff('s', greatest(rs.seg_start, b.rate_period_start), least(rs.seg_end, b.bucket_ts))) / 3600.0
            ELSE 0.0 END
        ) AS infusion_ml_output
    FROM buckets b
    JOIN rate_segments rs
      ON rs.seg_start < b.bucket_ts
     AND rs.seg_end > b.rate_period_start
    GROUP BY b.bucket_ts
),
history_discrete AS (
    SELECT
        coalesce(sum(CASE WHEN value > 0 THEN value ELSE 0.0 END), 0.0) AS discrete_opening_input,
        coalesce(sum(CASE WHEN value < 0 THEN -value ELSE 0.0 END), 0.0) AS discrete_opening_output
    FROM balance_events
    WHERE encounterId = '${options.encounterId}' AND balanceType = '${options.balanceType}'
      AND timestamp < cast('${startStr}' AS timestamp) ${fhirReferenceFilter}
),
history_rate_segments_raw AS (
    SELECT
        fhirResourceId,
        timestamp AS seg_start,
        lead(timestamp) OVER (
            PARTITION BY fhirResourceId
            ORDER BY timestamp
        ) AS next_ts,
        value AS rate_ml_per_h
    FROM balance_rates
    WHERE encounterId = '${options.encounterId}' AND balanceType = '${options.balanceType}'
      AND timestamp < cast('${startStr}' AS timestamp) ${fhirReferenceFilter}
),
history_rate_segments AS (
    SELECT
        fhirResourceId,
        seg_start,
        coalesce(next_ts, cast('${startStr}' AS timestamp)) AS seg_end,
        rate_ml_per_h
    FROM history_rate_segments_raw
),
history_infusions AS (
    SELECT
        coalesce(sum(CASE WHEN rate_ml_per_h > 0 THEN rate_ml_per_h * datediff('s', seg_start, seg_end) / 3600.0 ELSE 0.0 END), 0.0) AS infusion_opening_input,
        coalesce(sum(CASE WHEN rate_ml_per_h < 0 THEN (-rate_ml_per_h) * datediff('s', seg_start, seg_end) / 3600.0 ELSE 0.0 END), 0.0) AS infusion_opening_output
    FROM history_rate_segments
),
opening_balance AS (
    SELECT
        hd.discrete_opening_input + hi.infusion_opening_input AS opening_input,
        hd.discrete_opening_output + hi.infusion_opening_output AS opening_output,
        (hd.discrete_opening_input + hi.infusion_opening_input) - (hd.discrete_opening_output + hi.infusion_opening_output) AS opening_balance
    FROM history_discrete hd
    CROSS JOIN history_infusions hi
),
final_rows AS (
    SELECT
        b.bucket_ts AS ts,
        b.balance_day AS balance_day,
        coalesce(d.discrete_ml_input, 0.0)  AS discrete_ml_input,
        coalesce(d.discrete_ml_output, 0.0) AS discrete_ml_output,
        coalesce(i.infusion_ml_input, 0.0)  AS infusion_ml_input,
        coalesce(i.infusion_ml_output, 0.0) AS infusion_ml_output,
        (coalesce(d.discrete_ml_input, 0.0) + coalesce(i.infusion_ml_input, 0.0) - coalesce(d.discrete_ml_output, 0.0) - coalesce(i.infusion_ml_output, 0.0)) AS current_total_delta
    FROM buckets b
    LEFT JOIN discrete d ON b.bucket_ts = d.bucket_ts
    LEFT JOIN infusions i ON b.bucket_ts = i.bucket_ts
),
running_rows AS (
    SELECT
        ts,
        discrete_ml_input,
        discrete_ml_output,
        infusion_ml_input,
        infusion_ml_output,
        current_total_delta,

        sum(current_total_delta) OVER (PARTITION BY balance_day ORDER BY ts) AS current_total_acc,
        sum(discrete_ml_input + infusion_ml_input) OVER (PARTITION BY balance_day ORDER BY ts) AS current_acc_input,
        sum(discrete_ml_output + infusion_ml_output) OVER (PARTITION BY balance_day ORDER BY ts) AS current_acc_output,

        sum(current_total_delta) OVER (ORDER BY ts) AS running_total_delta,
        sum(discrete_ml_input + infusion_ml_input) OVER (ORDER BY ts) AS running_total_input,
        sum(discrete_ml_output + infusion_ml_output) OVER (ORDER BY ts) AS running_total_output
    FROM final_rows
)
SELECT
    cast(r.ts AS timestamp) AS bucket_ts,
    -- Current Accumulators
    round(r.current_acc_input, 3) AS current_acc_input,
    round(r.current_acc_output, 3) AS current_acc_output,
    round(r.current_total_acc, 3) AS current_total_acc,

    -- Deltas
    round(r.discrete_ml_input + r.infusion_ml_input, 3) AS current_delta_input,
    round(r.discrete_ml_output + r.infusion_ml_output, 3) AS current_delta_output,
    round(r.current_total_delta, 3) AS current_total_delta,

    -- Absolute Totals
    round(ob.opening_input + r.running_total_input, 3) AS total_input,
    round(ob.opening_output + r.running_total_output, 3) AS total_output,
    round((ob.opening_input + r.running_total_input) - (ob.opening_output + r.running_total_output), 3) AS total_balance
FROM running_rows r
CROSS JOIN opening_balance ob
ORDER BY r.ts

---

Variables

The query uses template variables that are injected at runtime before execution.

Variable	Type	Description
${timezone}	string	IANA timezone identifier (e.g. Europe/Berlin) used to convert UTC timestamps for day boundary calculations
${resetOffsetUs}	long (µs)	Microsecond offset from midnight at which the daily balance resets (e.g. 21600000000 = 06:00 local time)
${startStr}	timestamp string	Start of the query window (inclusive), ISO 8601 format
${endStr}	timestamp string	End of the query window (exclusive), ISO 8601 format
${endAdjustedStr}	timestamp string	End timestamp adjusted for bucket alignment (typically endStr minus one interval step)
${options.interval}	string	Bucket size as a QuestDB interval literal, e.g. '5m', '1h'
${options.intervalUs}	long (µs)	Numeric microsecond equivalent of options.interval, used for arithmetic
${options.encounterId}	string	FHIR Encounter ID to filter all events and rates to a specific patient encounter
${options.balanceType}	string	Balance category identifier (e.g. fluid type) to scope the calculation
${fhirReferenceFilter}	SQL fragment	Optional additional SQL filter fragment appended to WHERE clauses to filter for very specific FHIR Resource IDs only

---

Overview

The query operates on two primary tables:

• balance_events — discrete, point-in-time fluid in/out entries (e.g. a bolus, a drainage measurement)
• balance_rates — continuous rate-based entries (e.g. an IV infusion in ml/h)

The result is a time series of buckets, each containing current-period deltas, daily accumulators, and absolute totals (including a historical opening balance).

---

Section 1: buckets — Time Grid Generation

buckets AS (
    SELECT
        generate_series AS bucket_ts,
        CASE
            WHEN to_timezone(generate_series, '${timezone}') >=
                 timestamp_floor('1d', to_timezone(generate_series, '${timezone}')) + ${resetOffsetUs}L
            THEN timestamp_floor('1d', to_timezone(generate_series, '${timezone}')) + ${resetOffsetUs}L
            ELSE timestamp_floor('1d', to_timezone(generate_series, '${timezone}')) + ${resetOffsetUs}L - 86400000000L
        END AS balance_day,
        greatest(
            generate_series - ${options.intervalUs}L,
            cast('${startStr}' AS timestamp)
        ) AS rate_period_start
    FROM generate_series(
        cast('${startStr}' AS timestamp),
        cast('${endAdjustedStr}' AS timestamp),
        '${options.interval}'
    )
)

Explanation

buckets generates the uniform time grid that all subsequent CTEs are built on. Each row corresponds to a bucket timestamp (bucket_ts) produced by generate_series, stepping from startStr to endAdjustedStr at the configured interval.

balance_day — Balance day with a configurable reset time

The balance day does not start at UTC midnight but at a configured local time (resetOffsetUs) — for example 06:00 Europe/Berlin. The CASE block first converts the bucket timestamp into the target timezone (to_timezone) and compares it against the reset point of the current calendar day. If the bucket already falls on or after the reset, it belongs to the current balance day; if not, it still belongs to the previous balance day (reset point minus 24 h).

The key detail is that the offset is expressed as a fixed microsecond value relative to local midnight. This keeps the reset anchored at "06:00 local time" both in summer (CEST, UTC+2) and in winter (CET, UTC+1), so DST transitions do not shift the reset time.

rate_period_start — Integration start for infusion rates

Defines how far back infusion rates are integrated for a given bucket. Normally this is bucket_ts − intervalUs (the start of the current bucket interval). For the very first bucket of the query window, greatest(...) clamps the value to startStr so that no data from before the query window is included.

---

Section 2: discrete — Discrete Events per Bucket

discrete AS (
    SELECT
        b.bucket_ts AS bucket_ts,
        coalesce(e.in_val, 0.0) AS discrete_ml_input,
        coalesce(e.out_val, 0.0) AS discrete_ml_output
    FROM buckets b
    LEFT JOIN (
        SELECT
            timestamp_floor('${options.interval}', timestamp, cast('${startStr}' AS timestamp)) AS bucket_ts,
            sum(CASE WHEN value > 0 THEN value ELSE 0.0 END) AS in_val,
            sum(CASE WHEN value < 0 THEN -value ELSE 0.0 END) AS out_val
        FROM balance_events
        WHERE encounterId = '${options.encounterId}' AND balanceType = '${options.balanceType}'
          AND timestamp >= cast('${startStr}' AS timestamp)
          AND timestamp <  cast('${endStr}' AS timestamp) ${fhirReferenceFilter}
        GROUP BY 1
    ) e ON e.bucket_ts = b.bucket_ts
)

Explanation

discrete reads point-in-time fluid entries from balance_events and maps them onto the time grid from buckets.

timestamp_floor — bucket assignment

Each event's timestamp is floored to the nearest bucket boundary using timestamp_floor, aligned to startStr as the origin. This ensures the bucket boundaries match exactly with those generated in Section 1.

Aggregation — summing events per bucket

Multiple events that fall into the same bucket are summed up via GROUP BY on the floored bucket timestamp. Each bucket therefore yields a single aggregated input and output value.

Sign convention — input vs. output

Positive value entries are treated as fluid input (in_val), negative entries as fluid output (out_val). Output values are negated (-value) so both columns are always ≥ 0.

LEFT JOIN on buckets — preserving empty buckets

The aggregated events are left-joined back onto the buckets grid so that every bucket is present in the result, even if no events occurred in that interval. COALESCE(..., 0.0) fills any missing buckets with zero.

---

Section 3: Rate Segments — seed_rates, range_rates, rate_events, rate_segments_raw, rate_segments

This group of CTEs converts the raw rate entries from balance_rates into contiguous time segments, each with a known start, end, and rate value.

3a. seed_rates — Carry-over Rates from Before the Window

seed_rates AS (
    (
        SELECT
            fhirResourceId,
            timestamp AS ts,
            value AS rate_ml_per_h
        FROM balance_rates
        WHERE encounterId = '${options.encounterId}' AND balanceType = '${options.balanceType}'
          AND timestamp < cast('${startStr}' AS timestamp) ${fhirReferenceFilter}
        LATEST ON timestamp PARTITION BY fhirResourceId
    )
    WHERE rate_ml_per_h <> 0
)

Explanation

seed_rates handles the case where an infusion was started before the query window and is still active at startStr. Without this CTE, such a rate would be completely ignored in the calculation.

LATEST ON timestamp PARTITION BY fhirResourceId

This is QuestDB-specific syntax that returns only the most recent row per fhirResourceId from all entries before startStr — i.e. the last known rate change for each infusion.

WHERE rate_ml_per_h <> 0

Filters out entries where the rate is 0. A rate of 0 means the infusion was stopped and does not need to be carried forward into the query window.

3b. range_rates — Rates Within the Query Window

range_rates AS (
    SELECT
        fhirResourceId,
        timestamp AS ts,
        value AS rate_ml_per_h
    FROM balance_rates
    WHERE encounterId = '${options.encounterId}' AND balanceType = '${options.balanceType}'
      AND timestamp >= cast('${startStr}' AS timestamp)
      AND timestamp <  cast('${endStr}' AS timestamp) ${fhirReferenceFilter}
)

Explanation

range_rates captures all rate entries from balance_rates that fall within the query window (>= startStr and < endStr). This includes any rate starts, changes, or stops recorded during that period.

Together with seed_rates, it forms the complete rate event timeline for the query window.

3c. rate_events — Combined Rate Timeline

rate_events AS (
    SELECT * FROM seed_rates
    UNION ALL
    SELECT * FROM range_rates
)

Explanation

rate_events combines seed_rates (carry-over rates from before the query window) and range_rates (rate changes within the window) into a single chronological rate timeline via UNION ALL.

UNION ALL is used instead of UNION because no deduplication is needed — both sources cover disjoint time ranges.

3d. rate_segments_raw & rate_segments — Segment Windows

rate_segments_raw AS (
    SELECT
        fhirResourceId,
        ts AS seg_start,
        lead(ts) OVER (
            PARTITION BY fhirResourceId
            ORDER BY ts
        ) AS next_ts,
        rate_ml_per_h
    FROM rate_events
),
rate_segments AS (
    SELECT
        fhirResourceId,
        seg_start,
        coalesce(next_ts, cast('${endStr}' AS timestamp)) AS seg_end,
        rate_ml_per_h
    FROM rate_segments_raw
)

Explanation

balance_rates only stores the point in time when a rate changes — not the duration over which it applies. To calculate infusion volume, that duration must be made explicit. This is critical because a rate can start or end at any point within a bucket: if an infusion starts at 10:23 or stops at 10:23 inside a 15-minute bucket (10:15–10:30), only the active portion of that segment should contribute to the bucket's volume. Without explicit seg_start/seg_end boundaries, this precise clipping would not be possible.

rate_segments_raw and rate_segments solve this by converting the flat list of rate events into concrete time segments, each with a known start, end, and rate value.

lead() OVER (PARTITION BY fhirResourceId ORDER BY ts)

For each rate event, lead() looks up the timestamp of the next event for the same fhirResourceId. This turns a list of rate changes into a sequence of segments: the current event becomes seg_start, and the next event for the same resource becomes seg_end.

COALESCE(next_ts, endStr) — closing the last open segment

The last event of an infusion has no successor, so lead() returns NULL. COALESCE closes this open end by falling back to endStr.

The result is a set of non-overlapping time segments per fhirResourceId, each carrying a known rate — the foundation for the volume calculation in Section 4.

---

Section 4: infusions — Infusion Volume per Bucket

infusions AS (
    SELECT
        b.bucket_ts AS bucket_ts,
        sum(
            CASE WHEN rs.rate_ml_per_h > 0 THEN
                rs.rate_ml_per_h * greatest(0L, datediff('s', greatest(rs.seg_start, b.rate_period_start), least(rs.seg_end, b.bucket_ts))) / 3600.0
            ELSE 0.0 END
        ) AS infusion_ml_input,
        sum(
            CASE WHEN rs.rate_ml_per_h < 0 THEN
                (-rs.rate_ml_per_h) * greatest(0L, datediff('s', greatest(rs.seg_start, b.rate_period_start), least(rs.seg_end, b.bucket_ts))) / 3600.0
            ELSE 0.0 END
        ) AS infusion_ml_output
    FROM buckets b
    JOIN rate_segments rs
      ON rs.seg_start < b.bucket_ts
     AND rs.seg_end > b.rate_period_start
    GROUP BY b.bucket_ts
)

Explanation

infusions calculates the actual infusion volume delivered within each bucket's time window.

JOIN condition — overlap filter

rs.seg_start < b.bucket_ts AND rs.seg_end > b.rate_period_start is a range overlap check: only rate segments that actually overlap with the bucket's time window [rate_period_start, bucket_ts) are joined.

Overlap clipping — greatest / least

greatest(rs.seg_start, b.rate_period_start) and least(rs.seg_end, b.bucket_ts) clip the segment to the bucket's boundaries, so only the overlapping portion is used for the calculation.

datediff('s', start, end) — duration in seconds

Computes the length of the clipped overlap in seconds. This is the effective duration over which the rate was active within the bucket.

greatest(0L, datediff(...)) — guard against negative durations

Ensures the duration is never negative, which can occur at boundary conditions due to floating-point or timestamp edge cases.

Volume formula — rate_ml_per_h × seconds / 3600.0

Multiplies the rate (ml/h) by the overlap duration in seconds, then divides by 3600 to convert to ml.

Sign convention — input vs. output

Positive rates contribute to infusion_ml_input, negative rates (negated) contribute to infusion_ml_output.

GROUP BY b.bucket_ts

Sums contributions from all overlapping rate segments into a single input and output value per bucket.

---

Section 5: Opening Balance — history_discrete, history_rate_segments_*, history_infusions, opening_balance

This group calculates the cumulative fluid balance from before the query window, which serves as the starting point for the absolute totals.

5a. history_discrete — Historical Discrete Events

history_discrete AS (
    SELECT
        coalesce(sum(CASE WHEN value > 0 THEN value ELSE 0.0 END), 0.0) AS discrete_opening_input,
        coalesce(sum(CASE WHEN value < 0 THEN -value ELSE 0.0 END), 0.0) AS discrete_opening_output
    FROM balance_events
    WHERE encounterId = '${options.encounterId}' AND balanceType = '${options.balanceType}'
      AND timestamp < cast('${startStr}' AS timestamp) ${fhirReferenceFilter}
)

Explanation

history_discrete sums all discrete fluid entries from balance_events that occurred before the query window (timestamp < startStr). The result is a single scalar row representing the cumulative historical input and output — analogous to discrete in Section 2, but without bucket assignment, since only the overall total is needed here.

The sign convention is identical to Section 2: positive values count as input, negative values (negated) as output. COALESCE(..., 0.0) ensures the result is 0 if no historical events exist.

5b. history_rate_segments_raw & history_rate_segments — Historical Rate Segments

history_rate_segments_raw AS (
    SELECT
        fhirResourceId,
        timestamp AS seg_start,
        lead(timestamp) OVER (
            PARTITION BY fhirResourceId
            ORDER BY timestamp
        ) AS next_ts,
        value AS rate_ml_per_h
    FROM balance_rates
    WHERE encounterId = '${options.encounterId}' AND balanceType = '${options.balanceType}'
      AND timestamp < cast('${startStr}' AS timestamp) ${fhirReferenceFilter}
),
history_rate_segments AS (
    SELECT
        fhirResourceId,
        seg_start,
        coalesce(next_ts, cast('${startStr}' AS timestamp)) AS seg_end,
        rate_ml_per_h
    FROM history_rate_segments_raw
)

Explanation

history_rate_segments_raw and history_rate_segments are analogous to rate_segments_raw and rate_segments in Section 3d, but operate on the historical period before startStr.

lead() converts the historical rate events into time segments with explicit seg_start and seg_end boundaries. The key difference is that COALESCE(next_ts, startStr) closes the last open segment at startStr rather than endStr — because the historical period ends exactly there.

5c. history_infusions — Historical Infusion Volume

history_infusions AS (
    SELECT
        coalesce(sum(CASE WHEN rate_ml_per_h > 0 THEN rate_ml_per_h * datediff('s', seg_start, seg_end) / 3600.0 ELSE 0.0 END), 0.0) AS infusion_opening_input,
        coalesce(sum(CASE WHEN rate_ml_per_h < 0 THEN (-rate_ml_per_h) * datediff('s', seg_start, seg_end) / 3600.0 ELSE 0.0 END), 0.0) AS infusion_opening_output
    FROM history_rate_segments
)

Explanation

history_infusions is analogous to infusions in Section 4, but simplified — no bucket assignment or overlap clipping is needed for the historical period.

Each historical rate segment's volume is calculated as rate_ml_per_h × datediff('s', seg_start, seg_end) / 3600.0. The segments are already precisely bounded to the historical period (Section 5b), so the full segment duration is used directly.

The result is a single scalar row with the cumulative historical infusion input and output. COALESCE(..., 0.0) ensures the result is 0 if no historical rate segments exist.

5d. opening_balance — Combined Opening Balance

opening_balance AS (
    SELECT
        hd.discrete_opening_input + hi.infusion_opening_input AS opening_input,
        hd.discrete_opening_output + hi.infusion_opening_output AS opening_output,
        (hd.discrete_opening_input + hi.infusion_opening_input) - (hd.discrete_opening_output + hi.infusion_opening_output) AS opening_balance
    FROM history_discrete hd
    CROSS JOIN history_infusions hi
)

Explanation

opening_balance combines history_discrete and history_infusions into a single scalar row via CROSS JOIN. The cross join is correct here because both CTEs produce exactly one row each, so the result is always exactly one row.

It computes opening_input, opening_output, and opening_balance by adding the discrete and infusion components together.

This single row is later joined onto every bucket in Section 7 via another CROSS JOIN, providing the historical baseline for the absolute totals.

---

Section 6: final_rows & running_rows — Aggregation and Running Totals

6a. final_rows — Per-Bucket Join

final_rows AS (
    SELECT
        b.bucket_ts AS ts,
        b.balance_day AS balance_day,
        coalesce(d.discrete_ml_input, 0.0)  AS discrete_ml_input,
        coalesce(d.discrete_ml_output, 0.0) AS discrete_ml_output,
        coalesce(i.infusion_ml_input, 0.0)  AS infusion_ml_input,
        coalesce(i.infusion_ml_output, 0.0) AS infusion_ml_output,
        (coalesce(d.discrete_ml_input, 0.0) + coalesce(i.infusion_ml_input, 0.0) - coalesce(d.discrete_ml_output, 0.0) - coalesce(i.infusion_ml_output, 0.0)) AS current_total_delta
    FROM buckets b
    LEFT JOIN discrete d ON b.bucket_ts = d.bucket_ts
    LEFT JOIN infusions i ON b.bucket_ts = i.bucket_ts
)

Explanation

final_rows joins discrete and infusions onto the bucket grid from buckets using LEFT JOIN. This ensures every bucket is present in the result, even if neither discrete events nor infusions occurred during that interval.

COALESCE(..., 0.0) null-guards all four volume columns. current_total_delta is then computed as (discrete_input + infusion_input) − (discrete_output + infusion_output) — the net fluid balance for that bucket.

6b. running_rows — Window Functions for Cumulative Totals

running_rows AS (
    SELECT
        ts,
        discrete_ml_input,
        discrete_ml_output,
        infusion_ml_input,
        infusion_ml_output,
        current_total_delta,

        sum(current_total_delta) OVER (PARTITION BY balance_day ORDER BY ts) AS current_total_acc,
        sum(discrete_ml_input + infusion_ml_input) OVER (PARTITION BY balance_day ORDER BY ts) AS current_acc_input,
        sum(discrete_ml_output + infusion_ml_output) OVER (PARTITION BY balance_day ORDER BY ts) AS current_acc_output,

        sum(current_total_delta) OVER (ORDER BY ts) AS running_total_delta,
        sum(discrete_ml_input + infusion_ml_input) OVER (ORDER BY ts) AS running_total_input,
        sum(discrete_ml_output + infusion_ml_output) OVER (ORDER BY ts) AS running_total_output
    FROM final_rows
)

Explanation

running_rows computes cumulative sums over all buckets using window functions in two distinct scopes:

PARTITION BY balance_day ORDER BY ts — daily accumulators

Accumulates values within each balance day, resetting at the configurable reset time (as defined by balance_day in Section 1). Produces current_total_acc, current_acc_input, and current_acc_output.

ORDER BY ts (global, no partition) — running totals

Accumulates values across the entire query window without any reset. Produces running_total_delta, running_total_input, and running_total_output. These are added to the opening balance in Section 7 to compute the absolute totals.

---

Section 7: Final SELECT — Output Columns

SELECT
    cast(r.ts AS timestamp) AS bucket_ts,
    -- Current Accumulators
    round(r.current_acc_input, 3) AS current_acc_input,
    round(r.current_acc_output, 3) AS current_acc_output,
    round(r.current_total_acc, 3) AS current_total_acc,

    -- Deltas
    round(r.discrete_ml_input + r.infusion_ml_input, 3) AS current_delta_input,
    round(r.discrete_ml_output + r.infusion_ml_output, 3) AS current_delta_output,
    round(r.current_total_delta, 3) AS current_total_delta,

    -- Absolute Totals
    round(ob.opening_input + r.running_total_input, 3) AS total_input,
    round(ob.opening_output + r.running_total_output, 3) AS total_output,
    round((ob.opening_input + r.running_total_input) - (ob.opening_output + r.running_total_output), 3) AS total_balance
FROM running_rows r
CROSS JOIN opening_balance ob
ORDER BY r.ts

Explanation

The final SELECT exposes three groups of output columns:

Current Accumulators (current_acc_input, current_acc_output, current_total_acc)

Cumulative values since the last daily reset, derived from the PARTITION BY balance_day window functions in Section 6b.

Deltas (current_delta_input, current_delta_output, current_total_delta)

The net fluid change within this individual bucket only — discrete events plus infusion volume for that interval.

Absolute Totals (total_input, total_output, total_balance)

Opening balance (Section 5) added to the global running totals from Section 6b. The single scalar opening_balance row is applied to every bucket via CROSS JOIN.

round(..., 3) rounds all values to 3 decimal places. ORDER BY r.ts returns the buckets in chronological order.