FILTER (WHERE ...)

Applies to VAST Cluster 5.5.

Functions › Aggregate and analytic functions › Restricting an aggregate. FILTER (WHERE condition) after an aggregate restricts that one aggregate to the rows where the condition is true. The other aggregates in the query still see every row, so one query can count, sum or average several subsets in one scan.

Syntax

aggregate(arguments) FILTER (WHERE condition)
aggregate(arguments) FILTER (condition)

Returns: the aggregate's own result type.

Set up the examples

The examples connect with the ADBC driver, as in VAST Query Engine (ADBC), and work in a schema called docs_sql_fn_aggregate_filter_clause. Set QE_ENDPOINT, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY and VASTDB_BUCKET (SQL reference), and run the blocks in order in one Python session. Below each block is the output you should see.

This block connects, defines the run() and plan() helpers, and creates a table of eight orders. Order 7 has a region only, and order 8 has no values:

import os
import re

import adbc_driver_manager
import adbc_driver_manager.dbapi as dbapi
import adbc_driver_vastdb

BUCKET = os.environ["VASTDB_BUCKET"]
SCHEMA = f'"{BUCKET}/docs_sql_fn_aggregate_filter_clause"'

conn = dbapi.connect(
    driver=adbc_driver_vastdb.get_driver_path(),
    db_kwargs={
        "vast.db.endpoint": os.environ["QE_ENDPOINT"],
        "vast.db.access_key": os.environ["AWS_ACCESS_KEY_ID"],
        "vast.db.secret_key": os.environ["AWS_SECRET_ACCESS_KEY"],
    },
    autocommit=True,
)

def server_message(e):
    """The server's message in a driver error, without the request ID or the echoed statement."""
    text = re.sub(r"\[req=0x[0-9a-f]+\]: ", "", str(e))
    found = (re.search(r'message: "(.*?)", request_id', text, re.S)
             or re.search(r"description: (.*)\)", text, re.S))
    message = found.group(1) if found else text.strip().splitlines()[-1]
    message = re.sub(r"^Plan query: '.*?' failed with: ", "", message, flags=re.S)
    message = re.split(r"\\n|\n", message)[0].replace('\\"', '"')
    return message.replace(BUCKET, "<bucket>")

def run(sql):
    """Run one statement. Print the rows it returns, or the error."""
    try:
        with conn.cursor() as cur:
            cur.execute(sql)
            rows = cur.fetchall() if cur.description else []
    except (adbc_driver_manager.Error, OSError) as e:
        print(f"{type(e).__name__}: {server_message(e)}")
    else:
        print(rows)

def plan(sql):
    """Print the plan without its ~N Rows lines: the table's row count, which can still be 0 just after an insert."""
    with conn.cursor() as cur:
        cur.execute("EXPLAIN " + sql)
        steps = cur.fetchall()[0][1]
    print("\n".join(line for line in steps.splitlines() if "Rows" not in line))

TABLE = f"{SCHEMA}.orders"
run(f"CREATE SCHEMA IF NOT EXISTS {SCHEMA}")
run(f"CREATE TABLE IF NOT EXISTS {TABLE} "
    "(id BIGINT, region VARCHAR, amount BIGINT, paid BOOLEAN, item VARCHAR)")
run(f"INSERT INTO {TABLE} VALUES "
    "(1, 'north', 10, true, 'apple'), (2, 'north', -20, false, 'eclair'), "
    "(3, 'south', 0, true, ''), (4, 'south', 10, true, 'Apple'), "
    "(5, 'west', 30, false, 'melon'), (6, 'west', 5, true, 'zebra cake'), "
    "(7, 'west', NULL, NULL, NULL), (8, NULL, NULL, NULL, NULL)")

[]
[]
[(8,)]

What it does

For each row, the Query Engine evaluates the condition, and passes the row to the aggregate only if the condition is true. A row where the condition is false or NULL is left out of that aggregate, and only that one: each aggregate in the SELECT list has its own FILTER, or none.

The result is what the aggregate returns on the rows that pass. If no row passes, that's the aggregate's result on no rows: 0 for count and approx_count_distinct, NULL for the others, count_if included.

FILTER works on the aggregates count, count_if, approx_count_distinct, sum, avg, min, max, any_value, bool_and, bool_or, product and the standard deviation and variance functions, and on the running form of an aggregate used as a window function (OVER (ROWS UNBOUNDED PRECEDING)).

NULLs and edge cases

The same cases, as the cluster returns them. They use the table and helpers from Set up the examples:

run(f"SELECT count(*) FILTER (WHERE paid), count(*) FILTER (WHERE NOT paid), count(*) FROM {TABLE}")
run(f"SELECT count(*) FILTER (WHERE false), sum(amount) FILTER (WHERE false) FROM {TABLE}")
run(f"SELECT count(*) FILTER (paid) FROM {TABLE}")
run(f"SELECT count(*) FILTER (WHERE amount) FROM {TABLE}")
run(f"SELECT count(*) FILTER (WHERE item) FROM {TABLE}")
run(f"SELECT count(*) FILTER (WHERE paid) FROM {TABLE} WHERE amount >= 0")

[(4, 2, 8)]
[(0, None)]
[(4,)]
[(5,)]
OSError: Could not convert string 'apple' to BOOL
[(4,)]

Examples

Count the paid and unpaid orders, and sum the paid amounts, in one query:

run(f"SELECT count(*) FILTER (WHERE paid) AS paid_orders, "
    f"count(*) FILTER (WHERE NOT paid) AS unpaid_orders, "
    f"sum(amount) FILTER (WHERE paid) AS paid_total FROM {TABLE}")

[(4, 2, Decimal('25'))]

Different aggregates with different conditions. The condition can use any column, and functions:

run(f"SELECT avg(amount) FILTER (WHERE region = 'north'), max(amount) FILTER (WHERE region = 'west'), "
    f"sum(amount) FILTER (WHERE item LIKE 'a%'), sum(amount) FILTER (WHERE length(item) > 5) FROM {TABLE}")

[(-5.0, 30, Decimal('10'), Decimal('-15'))]

Count the rows per region. There's no GROUP BY in the Query Engine; for a few known values, one FILTER per value gives the same counts in one row:

run(f"SELECT count(*) FILTER (WHERE region = 'north') AS north, "
    f"count(*) FILTER (WHERE region = 'south') AS south, "
    f"count(*) FILTER (WHERE region = 'west') AS west, "
    f"count(*) FILTER (WHERE region IS NULL) AS no_region FROM {TABLE}")

[(2, 2, 3, 1)]

A running sum of the paid amounts, in the order the rows reach the Query Engine:

run(f"SELECT id, sum(amount) FILTER (WHERE paid) OVER (ROWS UNBOUNDED PRECEDING) FROM {TABLE}")

[(1, Decimal('10')), (2, Decimal('10')), (3, Decimal('10')), (4, Decimal('20')), (5, Decimal('20')), (6, Decimal('25')), (7, Decimal('25')), (8, Decimal('25'))]

How it runs in VAST

The Query Engine runs on the cluster's CNodes. An aggregate query has two steps, which EXPLAIN shows:

1. A scan in VAST DataBase (VAST_SCAN) reads the columns the query uses, and evaluates the WHERE conditions it can push down. A pushed-down condition doesn't appear in the plan.
2. An aggregate step (UNGROUPED_AGGREGATE) in the Query Engine computes the aggregates. A FILTER condition is part of this step: the plan prints it next to its aggregate, as Filter: ....

So a FILTER condition is never pushed down: the scan sends every row, with the condition's columns, and the aggregate step evaluates the condition. With WHERE amount > 0, the scan evaluates the condition and the plan doesn't show it; with FILTER (WHERE amount > 0), the condition is in the aggregate step:

plan(f"SELECT sum(amount) FROM {TABLE} WHERE amount > 0")
plan(f"SELECT sum(amount) FILTER (WHERE amount > 0) FROM {TABLE}")

┌───────────────────────────┐
│    UNGROUPED_AGGREGATE    │
│    ────────────────────   │
│        Aggregates:        │
│        sum(amount)        │
└─────────────┬─────────────┘
┌─────────────┴─────────────┐
│         VAST_SCAN         │
│    ────────────────────   │
│    Function: VAST_SCAN    │
│    Projections: amount    │
│                           │
└───────────────────────────┘
┌───────────────────────────┐
│    UNGROUPED_AGGREGATE    │
│    ────────────────────   │
│        Aggregates:        │
│ sum(amount) FILTER (WHERE │
│   (amount > 0)) Filter:   │
│        (amount > 0)       │
└─────────────┬─────────────┘
┌─────────────┴─────────────┐
│         VAST_SCAN         │
│    ────────────────────   │
│    Function: VAST_SCAN    │
│    Projections: amount    │
│                           │
└───────────────────────────┘

Several FILTER clauses share one scan and one aggregate step. The plan of the paid and unpaid counts reads only paid:

plan(f"SELECT count(*) FILTER (WHERE paid), count(*) FILTER (WHERE NOT paid) FROM {TABLE}")

┌───────────────────────────┐
│    UNGROUPED_AGGREGATE    │
│    ────────────────────   │
│        Aggregates:        │
│ count_star() FILTER (WHERE│
│     paid) Filter: paid    │
│ count_star() FILTER (WHERE│
│  (NOT paid)) Filter: (NOT │
│            paid)          │
└─────────────┬─────────────┘
┌─────────────┴─────────────┐
│         VAST_SCAN         │
│    ────────────────────   │
│    Function: VAST_SCAN    │
│     Projections: paid     │
│                           │
└───────────────────────────┘

Make it faster

• Use WHERE for a condition every aggregate shares. sum(amount) FILTER (WHERE amount > 0) and sum(amount) ... WHERE amount > 0 return the same value, but only the WHERE form is evaluated in VAST DataBase, so fewer rows leave storage. Keep FILTER for conditions that differ between aggregates.
• Combine them. Put the shared part in WHERE and the differences in FILTER: WHERE amount >= 0 is pushed down, and each FILTER runs only on the rows that pass it.
• One query, many subsets. Each FILTER adds work to the aggregate step, not another scan, so eight filtered counts in one query read the table once.

Limitations and errors

• Aggregates only. FILTER on a scalar function fails with Function "abs" is a Scalar Function. "DISTINCT", "FILTER", and "ORDER BY" are only applicable to aggregate functions.
• Not on geomean or wavg. geomean, geometric_mean, wavg and weighted_avg fail with Function "geomean" is a Macro Function. "DISTINCT", "FILTER", and "ORDER BY" are only applicable to aggregate functions. (with the name you used). Restrict the rows with WHERE instead.
• No subquery in the condition. FILTER (WHERE amount > (SELECT 0)) fails with a The operator '...' is currently not supported. error. Compute the value first and write it as a constant.
• No aggregate in the condition. FILTER (WHERE sum(amount) > 0) fails with aggregate function calls cannot be nested.
• No DISTINCT. count(DISTINCT amount) FILTER (WHERE paid) fails with The operator 'UNGROUPED_AGGREGATE' type=8 is currently not supported for this usage., as count(DISTINCT ...) does without FILTER.
• The condition must convert to BOOLEAN. A VARCHAR condition fails with Could not convert string ... to BOOL. Write a comparison, such as item <> ''.
• As a window function, only the running form. sum(amount) FILTER (WHERE paid) OVER () fails with The operator 'WINDOW' type=6 is currently not supported.; OVER (ROWS UNBOUNDED PRECEDING) runs. See window functions.

run(f"SELECT abs(amount) FILTER (WHERE paid) FROM {TABLE}")
run(f"SELECT count(*) FILTER (WHERE sum(amount) > 0) FROM {TABLE}")
run(f"SELECT count(DISTINCT amount) FILTER (WHERE paid) FROM {TABLE}")
run(f"SELECT sum(amount) FILTER (WHERE paid) OVER () FROM {TABLE}")

InternalError: Invalid Input Error: Function "abs" is a Scalar Function. "DISTINCT", "FILTER", and "ORDER BY" are only applicable to aggregate functions.
InternalError: Binder Error: aggregate function calls cannot be nested
InternalError: The operator 'UNGROUPED_AGGREGATE' type=8 is currently not supported for this usage.
InternalError: The operator 'WINDOW' type=6 is currently not supported.

Similar functions

The four forms on the orders:

run(f"SELECT count(*) FILTER (WHERE amount > 0), count_if(amount > 0), "
    f"sum(CASE WHEN amount > 0 THEN 1 ELSE 0 END) FROM {TABLE}")
run(f"SELECT count(*) FROM {TABLE} WHERE amount > 0")

[(4, Decimal('4'), Decimal('4'))]
[(4,)]

Clean up:

run(f"DROP TABLE IF EXISTS {TABLE}")
run(f"DROP SCHEMA IF EXISTS {SCHEMA}")
conn.close()

[]
[]

See also

• Aggregate and analytic functions: all the aggregates.
• count_if and count.
• SELECT, EXPLAIN and pushdowns.