Querying

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

This guide shows how to query VAST DataBase tables so that the cluster does the filtering. Most of it uses the Python SDK: choosing columns, writing predicates that push down, what to do with filters that don't, limiting results, tuning a query with QueryConfig, and reading results in parallel. Then it runs the same queries with Trino, Spark and the VAST Query Engine, and ends with the anti-patterns to avoid.

Run the code blocks in order in one Python session; the SQL blocks run in the Trino CLI. Below each block is the output you should see.

How a query runs

table.select() doesn't return rows directly. It returns a pyarrow.RecordBatchReader that streams record batches while you read it. Behind it, the SDK:

1. Turns your column list and predicate into one query request. A predicate it can't express raises NotImplementedError before anything is sent. The SDK never quietly filters on the client instead.
2. Divides the table into splits: disjoint subsets of the table's rows, interleaved in chunks across the whole table (read path). Unless you say otherwise, the number of splits is the table's row count divided by 4,000,000, with a minimum of 1.
3. Sends the splits to the cluster's data endpoints, one worker thread per endpoint, and passes each batch to your reader as it arrives.

The cluster applies the column list and the predicate, so only matching rows of the columns you asked for come back to your client. For what VAST DataBase evaluates, see supported pushdowns; the per-engine view is in pushdowns.

Set up

Create a table with 100,000 rows to query:

import os
import pyarrow as pa
import pyarrow.compute as pc
import vastdb
from vastdb import errors
from vastdb.config import QueryConfig

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

columns = pa.schema([
    ("id", pa.int64()),
    ("sensor", pa.utf8()),
    ("temperature", pa.float64()),
    ("note", pa.utf8()),
])
N = 100_000
data = pa.table({
    "id": list(range(N)),
    "sensor": [f"s{i % 10}" for i in range(N)],
    "temperature": [(i % 500) / 10 for i in range(N)],
    "note": [None if i % 7 else f"check-{i % 3}" for i in range(N)],
}, schema=columns)

with session.transaction() as tx:
    schema = tx.bucket(BUCKET).create_schema(SCHEMA, fail_if_exists=False)
    table = schema.create_table("readings", columns, fail_if_exists=False)
    table.insert(data)
    print("Rows:", table.select(columns=["id"]).read_all().num_rows)

Rows: 100000

Ask only for the columns you need

columns= is applied on the cluster. Columns you don't ask for aren't read or sent. Compare how much data comes back for all columns and for two fixed-width ones:

with session.transaction() as tx:
    t = tx.bucket(BUCKET).schema(SCHEMA).table("readings")
    everything = t.select().read_all()
    two = t.select(columns=["id", "temperature"]).read_all()

print(f"all columns: {everything.nbytes / 2**20:.1f} MiB")
print(f"id, temperature: {two.nbytes / 2**20:.1f} MiB")

all columns: 2.6 MiB
id, temperature: 1.5 MiB

select() with no columns reads every column, so name the columns in any query that runs more than once.

Write predicates that push down

Build predicates with Ibis expressions on the table, t["column"], or with the deferred _ from ibis. Each one is sent to the cluster with the query:

from ibis import _

with session.transaction() as tx:
    t = tx.bucket(BUCKET).schema(SCHEMA).table("readings")
    queries = {
        "comparison": t["temperature"] >= 49.5,
        "between": t["id"].between(100, 199),
        "isin": t["sensor"].isin(["s1", "s2"]),
        "startswith": t["note"].startswith("check-1"),
        "contains": t["note"].contains("-2"),
        "isnull": t["note"].isnull(),
        "not null": ~t["note"].isnull(),
        "AND of ORs": ((t["sensor"] == "s1") | (t["sensor"] == "s3")) & (t["temperature"] < 1),
        "deferred _": _.temperature < 0.5,
    }
    for name, predicate in queries.items():
        rows = t.select(columns=["id"], predicate=predicate).read_all().num_rows
        print(f"{name:>12}: {rows}")

  comparison: 1000
     between: 100
        isin: 20000
  startswith: 4762
    contains: 4762
      isnull: 85714
    not null: 14286
  AND of ORs: 400
  deferred _: 1000

The rules that decide whether a predicate can be sent:

• The predicate must be an AND of ORs, and every term inside one OR must use the same column.
• Each comparison is between a column and a literal value.
• ~ (NOT) works only directly around isnull().
• Only the operations listed in Predicates are supported. Functions such as upper() or endswith() aren't.

An AND of single-column OR groups pushes down; anything else raises an error you can rewrite.

Rewrite filters that don't push down

When a filter breaks those rules, select() raises NotImplementedError, and the message ends with the part it couldn't send:

with session.transaction() as tx:
    t = tx.bucket(BUCKET).schema(SCHEMA).table("readings")
    unsupported = {
        "OR across columns": (t["sensor"] == "s1") | (t["temperature"] > 49),
        "column vs column": t["id"] > t["temperature"],
        "NOT a comparison": ~(t["temperature"] > 1),
        "function": t["note"].endswith("-1"),
    }
    for name, predicate in unsupported.items():
        try:
            t.select(columns=["id"], predicate=predicate)
        except NotImplementedError as e:
            print(f"{name}: {str(e).splitlines()[-1]}")

OR across columns: Or(Equals(sensor, 's1'), Greater(temperature, 49)): r0.sensor == 's1' | r0.temperature > 49
column vs column: Greater(id, temperature): r0.id > r0.temperature
NOT a comparison: Not(Greater(temperature, 1)): Not(r0.temperature > 1)
function: EndsWith(note, '-1'): EndsWith(r0.note, end='-1')

Push down the part you can, and finish on the client with pyarrow.compute:

For example, notes ending in -1 are all notes that aren't null, so push down ~isnull() and apply ends_with in Arrow:

with session.transaction() as tx:
    t = tx.bucket(BUCKET).schema(SCHEMA).table("readings")
    candidates = t.select(columns=["id", "note"], predicate=~t["note"].isnull()).read_all()

ends_in_1 = candidates.filter(pc.ends_with(candidates["note"], "-1"))
print(candidates.num_rows, "rows pushed down,", ends_in_1.num_rows, "after the client-side filter")

14286 rows pushed down, 4762 after the client-side filter

Partitioned and sorted tables

On a partitioned table, the SDK first reads the table's partitions, keeping only the parts of your predicate that use partition source columns, and then queries each matching partition. A predicate on the partition columns therefore skips whole partitions; without one, every partition is scanned. This is partition pruning; see database partitioning.

On a sorted table, the SDK also asks the cluster for a row estimate that takes your predicate into account, and uses it to choose fewer splits. A predicate on the sort columns is what makes a sorted table pay off. For choosing between these structures, see data modeling.

Limit the rows

limit_rows stops the query once that many rows have been returned. The SDK then tells its worker threads to stop:

with session.transaction() as tx:
    t = tx.bucket(BUCKET).schema(SCHEMA).table("readings")
    sample = t.select(columns=["id", "sensor"], predicate=t["sensor"] == "s4", limit_rows=10).read_all()

print(sample.num_rows, "rows, all from", set(sample["sensor"].to_pylist()))

10 rows, all from {'s4'}

Which rows you get isn't defined, because the splits run in parallel. limit_rows is for samples and "is there any?" checks, not for top-N queries. Rows don't come back in any guaranteed order in general, so sort on the client when order matters.

Read results inside the transaction

The reader streams from the cluster, so it only works while its transaction is open. Reading it after the with block ends raises errors.MissingTransaction:

with session.transaction() as tx:
    t = tx.bucket(BUCKET).schema(SCHEMA).table("readings")
    reader = t.select(columns=["id"], predicate=t["id"] < 10)

try:
    reader.read_all()
except errors.MissingTransaction as e:
    print("Too late:", type(e).__name__)

Too late: MissingTransaction

To keep memory flat on a large result, process the batches as they arrive instead of calling read_all():

total, count = 0.0, 0
with session.transaction() as tx:
    t = tx.bucket(BUCKET).schema(SCHEMA).table("readings")
    for batch in t.select(columns=["temperature"], predicate=t["sensor"] == "s0"):
        total += pc.sum(batch["temperature"]).as_py()
        count += batch.num_rows

print(f"{count} rows, mean temperature {total / count:.2f}")

10000 rows, mean temperature 24.50

Tune a query with QueryConfig

Pass a vastdb.config.QueryConfig as config=. These are the fields that matter most, and what the SDK does with them on the client:

Client-side parallelism comes from endpoints, not splits. With the default settings, if the SDK discovers only one endpoint, the splits are processed one after another on one thread. To read with more threads, give data_endpoints more entries:

endpoint = os.environ["AWS_S3_ENDPOINT_URL"]
config = QueryConfig(num_splits=8, data_endpoints=[endpoint] * 4, query_id="docs")

with session.transaction() as tx:
    t = tx.bucket(BUCKET).schema(SCHEMA).table("readings")
    result = t.select(columns=["id"], predicate=t["temperature"] > 40, config=config).read_all()

print(result.num_rows, "rows")

19800 rows

To spread queries over more CNodes, ask your platform team which virtual IPs serve the database, and list them in data_endpoints. You can also set a session-wide default with SessionConfig(data_endpoints=[...]); see SessionConfig. For how splits and endpoints affect throughput, see performance.

Read splits in parallel yourself

select_splits() takes the same arguments as select(), but returns one reader per split instead of merging them. Use it when you want to hand splits to your own workers:

from concurrent.futures import ThreadPoolExecutor

with session.transaction() as tx:
    t = tx.bucket(BUCKET).schema(SCHEMA).table("readings")
    readers = t.select_splits(columns=["id"], predicate=t["sensor"] == "s9",
                              config=QueryConfig(num_splits=4))
    with ThreadPoolExecutor(max_workers=4) as pool:
        parts = list(pool.map(lambda r: r.read_all(), readers))

print(len(readers), "splits,", sum(p.num_rows for p in parts), "rows")

4 splits, 10000 rows

How rows are spread across the splits depends on how the data is laid out. On a small table, one split can hold every row and the others return nothing. The readers belong to the transaction, so read them all before it ends. select_splits() isn't available for partitioned tables.

Query with Trino, Spark or the VAST Query Engine

The same table can be queried with SQL. The examples below use the database (bucket) vastdb; use yours instead. The big difference from the SDK: a SQL engine never refuses a filter. Whatever it can't hand to VAST DataBase, it evaluates itself on the rows that come back. Results are the same either way. What changes is how many rows cross the network, so check which parts of a query pushed down.

Trino

The table is vast."<database>/<schema>".<table>. This query groups readings of 49 degrees or more whose note ends in -1:

SELECT sensor, count(*) AS readings, round(avg(temperature), 2) AS avg_temp
FROM vast."vastdb/guide_querying".readings
WHERE temperature >= 49 AND note LIKE '%-1'
GROUP BY sensor
ORDER BY sensor;

 sensor | readings | avg_temp 
--------+----------+----------
 s0     |       10 |     49.0 
 s1     |        9 |     49.1 
 s2     |        9 |     49.2 
 s3     |       10 |     49.3 
 s4     |       10 |     49.4 
 s5     |        9 |     49.5 
 s6     |        9 |     49.6 
 s7     |       10 |     49.7 
 s8     |       10 |     49.8 
 s9     |        9 |     49.9 
(10 rows)

To see what VAST DataBase evaluates, run EXPLAIN:

EXPLAIN
SELECT sensor FROM vast."vastdb/guide_querying".readings
WHERE temperature >= 49 AND note LIKE '%-1';

Look at the ScanFilterProject step. Its table = ... part lists the predicates that VAST DataBase evaluates, column by column; here, for temperature, SortedRangeSet[type=double, ranges=1, {[49.0,<max>)}]. Its filterPredicate is what Trino evaluates itself: here $like(note, LikePattern '[%-1]') (Trino 478, VAST Cluster 5.5.0-SP6). Trino pushes LIKE only in the form '%substring%', so it evaluates LIKE '%-1' on the rows that pass the temperature filter.

An OR across two columns isn't pushed by default. For sensor = 's1' OR temperature > 49, the table part says ALL, which means no predicate, and the whole OR is in filterPredicate. The session properties that change this are in Trino, and the full rules in pushdowns.

Spark

The table is ndb.<database>.<schema>.<table>. Where your platform team has set up the connector, getOrCreate() picks up its settings; to try it locally, see Try it from a local PySpark session. The same query as a DataFrame:

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.getOrCreate()
readings = spark.table(f"ndb.`{BUCKET}`.{SCHEMA}.readings")
hot = readings.filter((F.col("temperature") >= 49) & F.col("note").endswith("-1"))
(hot.groupBy("sensor")
    .agg(F.count("*").alias("readings"), F.round(F.avg("temperature"), 2).alias("avg_temp"))
    .orderBy("sensor")
    .show())

+------+--------+--------+
|sensor|readings|avg_temp|
+------+--------+--------+
|    s0|      10|    49.0|
|    s1|       9|    49.1|
|    s2|       9|    49.2|
|    s3|      10|    49.3|
|    s4|      10|    49.4|
|    s5|       9|    49.5|
|    s6|       9|    49.6|
|    s7|      10|    49.7|
|    s8|      10|    49.8|
|    s9|       9|    49.9|
+------+--------+--------+

The query plan shows the split. pushed_down_predicates in the VAST scan is what VAST DataBase evaluates, and a Filter above the scan is what Spark evaluates:

import re

plan = spark.sql(f"EXPLAIN SELECT sensor FROM ndb.`{BUCKET}`.{SCHEMA}.readings "
                 "WHERE temperature >= 49 AND note LIKE '%-1'").first()[0]
print("VAST DataBase:", re.search(r"pushed_down_predicates=(\[.*\]\])", plan).group(1))
print("Spark:", re.findall(r"Filter (.*)", plan))

VAST DataBase: [[temperature IS NOT NULL], [note IS NOT NULL], [temperature >= 49.0]]
Spark: ['EndsWith(note#53, -1)']

The Spark connector pushes no LIKE at all, nor startswith() or contains(), so filter on other columns as well where you can. See Spark.

VAST Query Engine

The Query Engine runs SQL inside the cluster, so only the result comes back. Connect with the ADBC driver; QE_ENDPOINT is the Query Engine's endpoint from your platform team:

import adbc_driver_manager.dbapi as dbapi
import adbc_driver_vastdb

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,
)
TABLE = f'"{BUCKET}/{SCHEMA}".readings'

with conn.cursor() as cur:
    cur.execute(f"SELECT count(*), round(avg(temperature), 2) FROM {TABLE} "
                "WHERE temperature >= 49 AND note LIKE '%-1'")
    print("readings, avg_temp:", cur.fetchone())
    cur.execute(f"SELECT id, temperature FROM {TABLE} WHERE sensor = 's3' "
                "ORDER BY temperature DESC, id LIMIT 3")
    print("hottest s3 readings:", cur.fetchall())

readings, avg_temp: (95, 49.45)
hottest s3 readings: [(493, 49.3), (993, 49.3), (1493, 49.3)]

In VAST Cluster 5.5 the Query Engine has no GROUP BY, and accepts ORDER BY only together with LIMIT:

import adbc_driver_manager

with conn.cursor() as cur:
    try:
        cur.execute(f"SELECT sensor, count(*) FROM {TABLE} GROUP BY sensor")
    except adbc_driver_manager.DatabaseError as e:
        # The server's own message is inside the driver's error text.
        print(type(e).__name__ + ":", re.search(r'\]: (.*?)"', str(e)).group(1))

InternalError: The operator 'HASH_GROUP_BY' type=9 is currently not supported.

To group, use Trino or Spark, or read the columns with the SDK and group them with PyArrow (table.group_by()). The SQL you can use is in the SQL reference.

Anti-patterns

Clean up

conn.close()
with session.transaction() as tx:
    schema = tx.bucket(BUCKET).schema(SCHEMA)
    schema.table("readings").drop()
    schema.drop()
    print("Dropped", SCHEMA)

Dropped guide_querying

Next steps

• Pushdowns: what each engine pushes down to VAST DataBase.
• Trino, Spark and VAST Query Engine (ADBC): connecting each engine.
• Predicates and select in the SDK reference.
• Arrow, pandas and Polars: hand results to DataFrame libraries and DuckDB.
• Data modeling: table structures that make predicates cheaper.
• See also: Optimize queries, Profile a query and Indexing and data skipping.