list_aggregate

Applies to VAST Cluster 5.5.

Functions › List, struct and map functions › Aggregate a list. Runs an aggregate, named as text, over the elements of a list and returns one value per list. list_aggr, array_aggregate, array_aggr and aggregate are the same function.

Syntax

list_aggregate(l, name)
list_aggregate(l, name, argument, ...)
list_aggr(l, name)
array_aggregate(l, name)
array_aggr(l, name)
aggregate(l, name)

Returns: the aggregate's result type for the list's element type. For example 'sum' on a list of integers returns HUGEINT, which ADBC returns as DECIMAL(38, 0) and Python prints as Decimal.

Aliases: list_aggr, array_aggregate, array_aggr, aggregate.

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_lists_list_aggregate. 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 five rows of lists (numbers, measurements, words, flags and lists of lists) and the name of an aggregate, agg. Row 3 has empty lists and row 4 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_lists_list_aggregate"'

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}.samples"
run(f"CREATE SCHEMA IF NOT EXISTS {SCHEMA}")
run(f"CREATE TABLE IF NOT EXISTS {TABLE} (id BIGINT, nums BIGINT[], vals DOUBLE[], words VARCHAR[], flags BOOLEAN[], nested BIGINT[][], agg VARCHAR)")
run(f"INSERT INTO {TABLE} VALUES "
    "(1, [1, 2, 3], [0.5, 2.5, -1.0], ['a', 'b', 'c'], [true, false, true], [[1, 2], [3]], 'sum'), "
    "(2, [3, NULL, 3], [2.0, NULL], ['b', NULL, 'B'], [NULL, true], [[], NULL, [4]], 'max'), "
    "(3, [], [], [], [], [], 'sum'), "
    "(4, NULL, NULL, NULL, NULL, NULL, NULL), "
    "(5, [-5, 0, -5, 10], [1e308, 1e308, -1e-308], ['\u00e9', 'Zo\u00eb', '\U0001F1EB\U0001F1F7', ''], [false, false], [[5, NULL], [], [6, 7, 8]], 'min')")

[]
[]
[(5,)]

What it does

list_aggregate(l, 'name') runs the aggregate name over the elements of l, as if they were the rows of a group, and returns its result. list_aggregate([1, 2, 3], 'max') returns 3.

• Many aggregates work by name, among them 'sum', 'min', 'max', 'avg', 'count', 'median', 'mode', 'first', 'last', 'product', 'stddev_samp', 'bool_and', 'bool_or', 'bit_and', 'approx_count_distinct', 'histogram', 'list' and 'string_agg'.
• Extra arguments follow the name. list_aggregate(['a', 'b'], 'string_agg', '|') returns 'a|b'; without a separator, ',' is used. list_aggregate([1, 2, 3, 4], 'quantile_cont', 0.25) returns 1.75, and a list of fractions, [0.25, 0.75], returns [1.75, 3.25].
• It runs aggregates that aren't available on rows. median, mode and quantile_cont aren't available as aggregates over a table's rows (not available), but they work here on a list.
• The name and the extra arguments must be constants. Taking them from a column fails.

NULLs and edge cases

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

run("SELECT list_aggregate(NULL, 'sum'), list_aggregate([NULL, NULL]::BIGINT[], 'sum'), "
    "list_aggregate([], 'sum'), list_aggregate([], 'count')")
run("SELECT list_aggregate([1, NULL, 3], 'count'), list_aggregate([1, 2, 3], 'SUM')")
run("SELECT list_aggregate([1, 2], NULL)")
run("SELECT list_aggregate([1, 2], 'nope')")
run("SELECT list_aggregate([1, 2], 'abs')")
run("SELECT list_aggregate(['a'], 'sum')")
run(f"SELECT list_aggregate(123, 'sum') FROM {TABLE} LIMIT 1")

[(None, None, None, 0)]
[(2, Decimal('6'))]
InternalError: Catalog Error: Aggregate Function with name NULL does not exist!
InternalError: Catalog Error: Aggregate Function with name nope does not exist!
InternalError: Catalog Error: abs is not an aggregate function
InternalError: Binder Error: No matching aggregate function
InternalError: Binder Error: No function matches the given name and argument types 'list_aggregate(INTEGER_LITERAL, STRING_LITERAL)'. You might need to add explicit type casts.

Examples

The sum, median and number of elements of each row's numbers:

run(f"SELECT id, list_aggregate(nums, 'sum'), list_aggregate(nums, 'median'), list_aggregate(nums, 'count') "
    f"FROM {TABLE} ORDER BY id LIMIT 10")

[(1, Decimal('6'), 2.0, 3), (2, Decimal('6'), 3.0, 2), (3, None, None, 0), (4, None, None, None), (5, Decimal('0'), -2.5, 4)]

Join the words with a separator, take the largest word, and test whether any flag is true. The aliases work the same way:

run(f"SELECT id, list_aggregate(words, 'string_agg', ', '), list_aggr(words, 'max'), aggregate(flags, 'bool_or') "
    f"FROM {TABLE} ORDER BY id LIMIT 10")

[(1, 'a, b, c', 'c', True), (2, 'b, B', 'b', True), (3, None, None, None), (4, None, None, None), (5, 'é, Zoë, 🇫🇷, ', '🇫🇷', False)]

Aggregates with extra arguments, and some that return lists or maps:

run("SELECT list_aggregate([1, 2, 3, 4], 'quantile_cont', 0.25), list_aggregate([1, 2, 3, 4], 'quantile_cont', [0.25, 0.75]), "
    "list_aggregate([1, 2, 3, 4], 'quantile_disc', 0.5), list_aggregate(['a', 'b'], 'string_agg')")
run("SELECT list_aggregate([1, 2, 2], 'mode'), list_aggregate([1, 2, 2], 'histogram'), list_aggregate([1, 2], 'list'), "
    "list_aggregate([3, 6], 'bit_and'), list_aggregate([1, 2, 3], 'stddev_samp')")

[(1.75, [1.75, 3.25], 2, 'a,b')]
[(2, [(1, 1), (2, 2)], [1, 2], 2, 1.0)]

The rows whose largest number is above 3:

run(f"SELECT id FROM {TABLE} WHERE list_aggregate(nums, 'max') > 3 ORDER BY id LIMIT 10")

[(5,)]

How it runs in VAST

The Query Engine runs on the cluster's CNodes, and a query on a table starts with a scan in VAST DataBase (VAST_SCAN). The scan reads a list column whole, and list_aggregate() always runs in the Query Engine. It aggregates within each row's list, so it's a scalar function: it returns one value per row and needs no GROUP BY.

• In the SELECT list, a projection step aggregates each row's list after the scan. The plan has no FILTER step.
• In WHERE, the scan sends the list for every row, and a FILTER step computes the aggregate and keeps the rows that match. No rows are skipped in storage.

The plan of the WHERE example shows the FILTER step above the scan:

plan(f"SELECT id FROM {TABLE} WHERE list_aggregate(nums, 'max') > 3")

┌───────────────────────────┐
│         PROJECTION        │
│    ────────────────────   │
│             #1            │
│                           │
└─────────────┬─────────────┘
┌─────────────┴─────────────┐
│           FILTER          │
│    ────────────────────   │
│ (list_aggregate(nums, 'max│
│          ') > 3)          │
│                           │
└─────────────┬─────────────┘
┌─────────────┴─────────────┐
│         VAST_SCAN         │
│    ────────────────────   │
│    Function: VAST_SCAN    │
│                           │
│        Projections:       │
│            nums           │
│             id            │
│                           │
└───────────────────────────┘

Make it faster

• Add a condition on a plain column. A comparison of a column with a constant, such as id > 1, is evaluated in VAST DataBase, so the aggregate runs only on the rows that pass it. The plan for id > 1 alone has no FILTER step:

plan(f"SELECT id FROM {TABLE} WHERE id > 1")

┌───────────────────────────┐
│         VAST_SCAN         │
│    ────────────────────   │
│    Function: VAST_SCAN    │
│      Projections: id      │
│                           │
└───────────────────────────┘

• Store the value if you filter on it often. Compute the sum or the largest element when you write the rows and keep it in its own column: max_num > 3 is pushed down, list_aggregate(nums, 'max') > 3 isn't.

Limitations and errors

• The name must be a constant. Taking it from a column fails with Invalid Input Error: Aggregate function name must be a constant. The separator of 'string_agg' must be a constant too.
• Not every aggregate works. 'count_star' fails with No matching aggregate function, and 'arg_max' with Aggregate function arg_max(NULL, 1) is not supported for list_aggr: extra arguments were not removed during bind. An argument the aggregate doesn't take, such as list_aggregate([1, 2], 'sum', 1), fails with No matching aggregate function.
• The aggregate must accept the element type. 'sum' on a list of text fails with No matching aggregate function.
• ORDER BY needs a LIMIT. Without one, the query fails with The operator 'ORDER_BY' type=1 is currently not supported for this usage.

run(f"SELECT id, list_aggregate(nums, agg) FROM {TABLE} ORDER BY id LIMIT 10")
run("SELECT list_aggregate([1, NULL, 3], 'count_star')")
run("SELECT list_aggregate([1, 2], 'arg_max', 1)")

InternalError: Invalid Input Error: Aggregate function name must be a constant
InternalError: Binder Error: No matching aggregate function
InternalError: Invalid Input Error: Aggregate function arg_max(NULL, 1) is not supported for list_aggr: extra arguments were not removed during bind

Similar functions

The list aggregates are shorter ways to write the common names, with the same results:

run("SELECT list_sum(l), list_aggregate(l, 'sum'), list_min(l), list_aggregate(l, 'min'), "
    "list_count(l), list_aggregate(l, 'count') FROM (SELECT [3, 1, NULL, 3, 10] AS l)")
run("SELECT list_string_agg(['a', 'b']), list_aggregate(['a', 'b'], 'string_agg', ',')")

[(Decimal('17'), Decimal('17'), 1, 1, 4, 4)]
[('a,b', 'a,b')]

Clean up:

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

[]
[]

See also

• List, struct and map functions: all the list aggregates.
• Aggregate functions: aggregates over the rows of a table.
• Where filters run and pushdowns.
• EXPLAIN: how to read a plan.