Pushdowns

Applies to VAST Cluster 5.5. Examples use the VAST DataBase Python SDK 2.1.0.

A pushdown is work that a client or query engine hands to VAST DataBase, so that the cluster's CNodes do it next to the data. Only the result comes back. This page lists what VAST DataBase can take over in VAST Cluster 5.5, and how each engine uses it.

What VAST DataBase can push down

See Supported Pushdowns.

COUNT(*) is the only aggregation pushed down this way. With a filter, as in SELECT COUNT(*) FROM my_table WHERE my_column > 100;, the cluster applies the filter and returns the matching rows, and the engine counts them. No other aggregation is pushed down.

Anything the cluster can't do, the engine does itself, after it gets the filtered data back. In this example, both WHERE conditions are pushed down and the engine computes the average:

SELECT AVG(col1) FROM vast.schema.mytable WHERE col1 > 100 AND col3 < 50;

Supported predicates

See Supported Pushdowns.

Put together, a filter pushes down when it's an AND of conditions, and each condition is about one column. A condition can OR several tests on that same column. An OR across two different columns doesn't fit that shape.

By engine

For how to write filters and choose columns so that more of the work is pushed down, see querying.

Python SDK

The SDK's predicate rules match the shape above: an AND of OR groups, with one column per OR group, and comparisons against literals only. They are listed in full in Predicates. In short:

• Supported: comparisons including !=, isnull(), ~isnull(), isin(), between(), contains(), startswith(), & and |.
• An OR group must use one column. Comparing two columns isn't supported.
• ~ (NOT) works only around isnull().
• You can filter on a struct sub-field that's a leaf, for example t["location"]["lat"], and on a fixed-size list with a flat element type. You can't filter on a list, a map or a whole struct.

columns= takes top-level column names. A struct column is returned whole.

Pushdown errors and how to rewrite the predicate are in the error reference.

Trino

What the connector pushes down is in Trino. Two things to know when you check it:

• EXPLAIN shows the pushed-down predicates in the TableScan, for plain, sorted and partitioned tables alike. A condition that Trino evaluates itself, such as one on upper(region), appears as a filterPredicate in a ScanFilterProject step instead. Row filters from row and column security, applied through impersonation, appear in the TableScan too.
• Dynamic filtering reaches VAST DataBase. In a join, Trino collects the join keys from the smaller (build) side and adds them to the scan of the larger (probe) side, so VAST DataBase returns only the rows that can match. In a join of a 20,000,000-row table with a dimension table filtered to 5 keys, the probe scan read 107 rows; with dynamic filtering turned off (SET SESSION enable_dynamic_filtering = false) it read all 20,000,000, and the query took two to seven times as long (two runs each). Same result either way. Those timings are single-endpoint loopback results, so compare them only with each other.

Example: a pushed-down filter

This example uses the connection environment variables from the quickstart. It creates a small table with a struct column:

import os

import pyarrow as pa
import vastdb

session = vastdb.connect()
BUCKET = os.environ["VASTDB_BUCKET"]

columns = pa.schema([
    ("id", pa.int64()),
    ("city", pa.utf8()),
    ("temperature", pa.float64()),
    ("location", pa.struct([("lat", pa.float64()), ("lon", pa.float64())])),
])
rows = pa.table({
    "id": [1, 2, 3, 4, 5],
    "city": ["Lisbon", "Oslo", "Cairo", "Lima", "Perth"],
    "temperature": [21.5, 4.0, 30.2, 18.9, 25.1],
    "location": [
        {"lat": 38.7, "lon": -9.1},
        {"lat": 59.9, "lon": 10.7},
        {"lat": 30.0, "lon": 31.2},
        {"lat": -12.0, "lon": -77.0},
        {"lat": -31.9, "lon": 115.9},
    ],
}, schema=columns)

with session.transaction() as tx:
    schema = tx.bucket(BUCKET).create_schema("ref_pushdowns", fail_if_exists=False)
    table = schema.create_table("cities", columns, fail_if_exists=False)
    table.insert(rows)
print("Inserted", rows.num_rows, "rows")

Inserted 5 rows

This filter is an AND of three conditions. The first ORs two tests on city, and the third uses the struct leaf location.lat. All of it runs on the cluster, and only two columns come back:

with session.transaction() as tx:
    table = tx.bucket(BUCKET).schema("ref_pushdowns").table("cities")
    predicate = (
        ((table["city"] == "Lima") | (table["city"].startswith("P")))
        & (table["temperature"] > 15)
        & (table["location"]["lat"] < 0)
    )
    result = table.select(columns=["city", "temperature"], predicate=predicate).read_all()

print(result.sort_by("city").to_pylist())

[{'city': 'Lima', 'temperature': 18.9}, {'city': 'Perth', 'temperature': 25.1}]

An OR across two columns can't be pushed down. The SDK raises NotImplementedError before it sends anything. The message starts with the table's definition and ends with the part of the predicate it couldn't send, so the example prints only the last line:

with session.transaction() as tx:
    table = tx.bucket(BUCKET).schema("ref_pushdowns").table("cities")
    try:
        table.select(predicate=(table["city"] == "Oslo") | (table["temperature"] > 25)).read_all()
    except NotImplementedError as e:
        print(type(e).__name__, "-", str(e).splitlines()[-1])

NotImplementedError - Or(Equals(city, 'Oslo'), Greater(temperature, 25)): r0.city == 'Oslo' | r0.temperature > 25

To get those rows, run one query per column and combine the results, or push down what you can and filter the rest in Arrow.

Clean up:

with session.transaction() as tx:
    schema = tx.bucket(BUCKET).schema("ref_pushdowns")
    schema.table("cities").drop()
    schema.drop()
print("Dropped")

Dropped

See also: Optimize queries, Profile a query and Indexing and data skipping.