list_reduce

Applies to VAST Cluster 5.5.

Functions › List, struct and map functions › Lambdas. Folds a list into one value: a lambda combines the result so far with each element in turn, from left to right, starting from the first element. array_reduce and reduce are the same function.

Syntax

list_reduce(l, (acc, x) -> expression)
list_reduce(l, (acc, x, i) -> expression)
array_reduce(l, (acc, x) -> expression)
reduce(l, (acc, x) -> expression)

Returns: a value of the element type of l.

Aliases: array_reduce, reduce.

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_reduce. 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, words, measurements and lists of lists. 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_reduce"'

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[], words VARCHAR[], vals DOUBLE[], nested BIGINT[][])")
run(f"INSERT INTO {TABLE} VALUES "
    "(1, [1, 2, 3], ['a', 'b', 'c'], [0.5, 2.5, -1.0], [[1, 2], [3]]), "
    "(2, [3, NULL, 3], ['b', NULL, 'B'], [2.0, NULL], [[], NULL, [4]]), "
    "(3, [], [], [], []), "
    "(4, NULL, NULL, NULL, NULL), "
    "(5, [-5, 0, -5, 10], ['\u00e9', 'Zo\u00eb', '\U0001F1EB\U0001F1F7', ''], [1e308, 1e308, -1e-308], [[5, NULL], [], [6, 7, 8]])")

[]
[]
[(5,)]

What it does

list_reduce(l, (acc, x) -> expression) starts with the first element of l as acc. For each following element x, it computes expression and makes the result the new acc. The last acc is the result:

• list_reduce([1, 2, 3], (acc, x) -> acc + x) computes (1 + 2) + 3 and returns 6.
• list_reduce(['a', 'b', 'c'], (acc, x) -> acc || x) returns 'abc'; with x || acc it returns 'cba'.
• A list of one element returns that element: list_reduce([7], ...) is 7.
• With (acc, x, i), i is the position of x: list_reduce([1, 2, 3], (acc, x, i) -> acc + x * i) is 1 + 2 * 2 + 3 * 3 = 14.

Two things to know:

• There's no initial value, so an empty list is an error. list_reduce on [] stops the whole query with Cannot perform list_reduce on an empty input list, even if only one row of the table has an empty list. To start from a value of your own, put it in front of the list with list_prepend: list_reduce(list_prepend(0, l), ...) returns 0 for an empty list. Or guard the call with CASE WHEN len(l) > 0 THEN list_reduce(...) END, or WHERE len(l) > 0.
• The result has the element type. The lambda's result is converted to it: on an INTEGER list, (acc, x) -> acc * 1.5 returns 3 for [1, 2, 3], and acc::VARCHAR || x::VARCHAR returns the INTEGER 123.

NULLs and edge cases

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

run("SELECT list_reduce(NULL::BIGINT[], (acc, x) -> acc + x), list_reduce([1, NULL, 2], (acc, x) -> acc + x), "
    "list_reduce([7], (acc, x) -> acc + x), list_reduce([3, 1, 2], (acc, x) -> greatest(acc, x))")
run("SELECT list_reduce(list_prepend(0, []::BIGINT[]), (acc, x) -> acc + x)")
run("SELECT list_reduce([]::BIGINT[], (acc, x) -> acc + x)")
run("SELECT list_reduce([1, 2], (acc, x) -> acc + x, 100)")
run("SELECT list_reduce([1, 2], x -> x)")
run("SELECT list_reduce([9223372036854775807, 1], (acc, x) -> acc + x)")
run("SELECT list_reduce(123, (acc, x) -> acc + x)")

[(None, None, 7, 3)]
[(0,)]
OSError: Cannot perform list_reduce on an empty input list
InternalError: Binder Error: failed to bind function, either: Invalid number of function arguments!
InternalError: Binder Error: list_reduce expects a function with 2 or 3 arguments
OSError: Overflow in addition of INT64 (9223372036854775807 + 1)!
InternalError: Binder Error: failed to bind function, either: Invalid LIST argument during lambda function binding!

The examples from What it does:

run("SELECT list_reduce([1, 2, 3], (acc, x) -> acc + x), list_reduce(['a', 'b', 'c'], (acc, x) -> acc || x), "
    "list_reduce(['a', 'b', 'c'], (acc, x) -> x || acc), list_reduce([1, 2, 3], (acc, x, i) -> acc + x * i)")
run("SELECT list_reduce([1, 2, 3], (acc, x) -> acc * 1.5), list_reduce([1, 2, 3], (acc, x) -> acc::VARCHAR || x::VARCHAR)")

[(6, 'abc', 'cba', 14)]
[(3, 123)]

Examples

On the whole table, list_reduce fails, because row 3's lists are empty:

run(f"SELECT id, list_reduce(nums, (acc, x) -> acc + x) FROM {TABLE} ORDER BY id LIMIT 10")

OSError: Cannot perform list_reduce on an empty input list

Skip the empty lists with WHERE len(...) > 0. Join the words with -, and concatenate the lists of lists into one list:

run(f"SELECT id, list_reduce(nums, (acc, x) -> acc + x), reduce(words, (acc, x) -> acc || '-' || x), "
    f"array_reduce(nested, (acc, x) -> list_concat(acc, x)) FROM {TABLE} WHERE len(nums) > 0 ORDER BY id LIMIT 10")

[(1, 6, 'a-b-c', [1, 2, 3]), (2, None, None, [4]), (5, 0, 'é-Zoë-🇫🇷-', [5, None, 6, 7, 8])]

Keep every row, with a guard or a starting value. With CASE, an empty or NULL list returns NULL; with list_prepend(0, ...), it returns 0:

run(f"SELECT id, CASE WHEN len(nums) > 0 THEN list_reduce(nums, (acc, x) -> acc + x) END, "
    f"list_reduce(list_prepend(0, nums), (acc, x) -> acc + x) FROM {TABLE} ORDER BY id LIMIT 10")

[(1, 6, 6), (2, None, None), (3, None, 0), (4, None, 0), (5, 0, 0)]

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_reduce() and its lambda always run in the Query Engine:

• In the SELECT list, a projection step folds each row's list after the scan.
• In WHERE, the scan sends the list for every row, and a FILTER step computes the condition. len(nums) > 0 is such a condition: it runs in the Query Engine too, before the projection.

The plan of the WHERE len(...) > 0 example shows the FILTER step above the scan:

plan(f"SELECT id, list_reduce(nums, (acc, x) -> acc + x) FROM {TABLE} WHERE len(nums) > 0")

┌───────────────────────────┐
│         PROJECTION        │
│    ────────────────────   │
│             id            │
│     list_reduce(nums)     │
│                           │
└─────────────┬─────────────┘
┌─────────────┴─────────────┐
│           FILTER          │
│    ────────────────────   │
│      (len(nums) > 0)      │
│                           │
└─────────────┬─────────────┘
┌─────────────┴─────────────┐
│         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 lambda 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      │
│                           │
└───────────────────────────┘

• Use a list aggregate for sums and the like. list_sum, and list_aggregate skip NULL elements and return NULL for an empty list instead of failing, so they need no guard.

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

[(1, Decimal('6'), 3), (2, Decimal('6'), 3), (3, None, None), (4, None, None), (5, Decimal('0'), 10)]

Limitations and errors

• No initial-value argument. A third argument fails with Invalid number of function arguments!. Use list_prepend(initial, l), as in What it does.
• Empty lists stop the query. Cannot perform list_reduce on an empty input list; guard the call as shown in Examples.
• Two or three lambda parameters. One fails with list_reduce expects a function with 2 or 3 arguments, four with This lambda function only supports up to three lambda parameters!.

Similar functions

run("SELECT list_reduce(l, (acc, x) -> acc + x), list_reduce(list_filter(l, x -> x IS NOT NULL), (acc, x) -> acc + x), "
    "list_sum(l), list_aggregate(l, 'sum') FROM (SELECT [3, 1, NULL, 3, 10] AS l)")

[(None, 17, Decimal('17'), Decimal('17'))]

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 functions, and how lambdas work.
• list_transform and list_filter: the other lambda functions.
• Where filters run and pushdowns.
• EXPLAIN: how to read a plan.