wavg

Applies to VAST Cluster 5.5.

Functions › Aggregate and analytic functions › Sums and averages. Returns a weighted mean: each value counts in proportion to its weight. weighted_avg is the same function.

Syntax

wavg(value, weight)
weighted_avg(value, weight)

Returns: DOUBLE.

Aliases: weighted_avg.

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_wavg. 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 course grades: a grade point score and the course's credits. Course 4 has no score yet, and course 5 has no credits:

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_wavg"'

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}.grades"
run(f"CREATE SCHEMA IF NOT EXISTS {SCHEMA}")
run(f"CREATE TABLE IF NOT EXISTS {TABLE} (id BIGINT, course VARCHAR, score DOUBLE, credits DOUBLE)")
run(f"INSERT INTO {TABLE} VALUES "
    "(1, 'Algebra', 3.5, 4.0), "
    "(2, 'Drawing', 2.0, 2.0), "
    "(3, 'Physics', 4.0, 1.0), "
    "(4, 'History', NULL, 3.0), "
    "(5, 'Seminar', 3.0, NULL)")

[]
[]
[(5,)]

What it does

wavg(value, weight) is an aggregate: it multiplies each value by its weight, adds up the products, and divides by the total of the weights. The plan shows exactly what it computes:

sum(value * weight) / sum(CASE WHEN value IS NOT NULL THEN weight ELSE 0 END)

So a row counts only when both its value and its weight are known. A row with a NULL value is left out of both sums, and so is a row with a NULL weight. This is why sum(value * weight) / sum(weight) isn't the same thing: it divides by the weights of the rows with no value too.

With every weight 1, wavg(value, 1) is the plain mean, avg. Negative weights are accepted: wavg(value, -weight) gives the same result as wavg(value, weight), because both sums change sign.

wavg is a macro, a shorthand the Query Engine expands into those two sums, not an aggregate of its own. That's why it can't take FILTER, DISTINCT or ORDER BY (see Limitations and errors).

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 wavg(score, credits), weighted_avg(score, credits) FROM {TABLE}")
run(f"SELECT wavg(score, credits) FROM {TABLE} WHERE id IN (4, 5)")
run(f"SELECT wavg(score, credits) FROM {TABLE} WHERE id > 100")
run(f"SELECT wavg(score, 0), wavg(score, 1), avg(score), wavg(score, -credits) FROM {TABLE}")
run(f"SELECT wavg(course, credits) FROM {TABLE}")
run(f"SELECT wavg(score) FROM {TABLE}")
run(f"SELECT wavg(score, credits) FILTER (WHERE id < 3) FROM {TABLE}")

[(3.142857142857143, 3.142857142857143)]
[(None,)]
[(None,)]
[(nan, 3.125, 3.125, 3.142857142857143)]
InternalError: Binder Error: No function matches the given name and argument types '*(VARCHAR, DOUBLE)'. You might need to add explicit type casts.
InternalError: Binder Error: Macro function wavg(value, weight) requires 2 positional arguments, but a single positional argument was provided.
InternalError: Invalid Input Error: Function "wavg" is a Macro Function. "DISTINCT", "FILTER", and "ORDER BY" are only applicable to aggregate functions.

Examples

The credit-weighted grade point average, next to the plain mean of the scores and the formula written out:

run(f"SELECT wavg(score, credits), avg(score), "
    f"sum(score * credits) / sum(credits) FILTER (WHERE score IS NOT NULL) FROM {TABLE}")

[(3.142857142857143, 3.125, 3.142857142857143)]

The weighted average of the first three courses only. wavg can't take FILTER, so restrict the rows in WHERE:

run(f"SELECT wavg(score, credits) FROM {TABLE} WHERE id <= 3")

[(3.142857142857143,)]

How it runs in VAST

The Query Engine runs on the cluster's CNodes. A wavg query has three steps, which EXPLAIN shows:

1. A scan in VAST DataBase (VAST_SCAN), which reads the value and weight columns and evaluates the WHERE conditions it can push down.
2. One aggregate step (UNGROUPED_AGGREGATE) in the Query Engine, which computes the two sums: of value * weight, and of the weights of the rows that have a value.
3. A PROJECTION step, which divides the first sum by the second.

plan(f"SELECT wavg(score, credits) FROM {TABLE} WHERE id <= 3")

┌───────────────────────────┐
│         PROJECTION        │
│    ────────────────────   │
│ (sum((score * credits)) / │
│  sum(CASE  WHEN ((score IS│
│  NOT NULL)) THEN (credits)│
│   ELSE CAST(0 AS DOUBLE)  │
│            END))          │
│                           │
└─────────────┬─────────────┘
┌─────────────┴─────────────┐
│    UNGROUPED_AGGREGATE    │
│    ────────────────────   │
│        Aggregates:        │
│   sum((score * credits))  │
│ sum(CASE  WHEN ((score IS │
│  NOT NULL)) THEN (credits)│
│        ELSE 0.0 END)      │
└─────────────┬─────────────┘
┌─────────────┴─────────────┐
│         VAST_SCAN         │
│    ────────────────────   │
│    Function: VAST_SCAN    │
│                           │
│        Projections:       │
│           score           │
│          credits          │
│                           │
└───────────────────────────┘

Make it faster

• Restrict the rows in WHERE. A comparison of a column with a constant is evaluated in VAST DataBase, so the rows that don't match never leave storage. It's also the only way to restrict wavg, since it can't take FILTER.
• Compute it with other aggregates in one query. wavg, avg, sum and the others in one SELECT list share one scan.
• Write the formula out to restrict different averages differently. sum(score * credits) FILTER (WHERE c) / sum(credits) FILTER (WHERE c AND score IS NOT NULL) gives wavg for the rows where c holds, and lets one query compute several of them.

Limitations and errors

• No FILTER, DISTINCT or ORDER BY. They fail with Function "wavg" is a Macro Function. "DISTINCT", "FILTER", and "ORDER BY" are only applicable to aggregate functions. (weighted_avg in the message when you use that name). Use WHERE, or write the formula out with FILTER on each sum.
• No window form. wavg(x, w) OVER (...) fails with Catalog Error: / is not an aggregate function.
• Numbers only. A string, a date, a BOOLEAN or a list fails with No function matches the given name and argument types, and the message names the multiplication inside the macro, such as '*(VARCHAR, DOUBLE)'.
• Integer overflow. The product value * weight is computed in the columns' type: two large BIGINT columns can fail with Overflow in multiplication of INT64 (...)!. Cast one of them to DOUBLE first.
• No GROUP BY. A weighted average per group fails with The operator 'HASH_GROUP_BY' type=9 is currently not supported.

run(f"SELECT id, wavg(score, credits) OVER (ROWS UNBOUNDED PRECEDING) FROM {TABLE}")
run(f"SELECT course, wavg(score, credits) FROM {TABLE} GROUP BY course")

InternalError: Catalog Error: / is not an aggregate function
InternalError: The operator 'HASH_GROUP_BY' type=9 is currently not supported.

Similar functions

run(f"SELECT wavg(score, credits), avg(score), wavg(score, 1), sum(score * credits) / sum(credits), "
    f"geomean(score) FROM {TABLE}")

[(3.142857142857143, 3.125, 3.125, 2.2, 3.0274001040350913)]

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, and the ones that aren't available.
• avg, sum and FILTER.
• Pushdowns and EXPLAIN.