SQL functions

Pondra’s SQL is Apache DataFusion’s, so every function in DataFusion’s lists works here: strings, math, dates, arrays, regexp_*, window functions and aggregates. This page lists what Pondra adds. For the functions you make yourself, see CREATE FUNCTION.

Spark’s functions

Spark SQL’s functions whose names DataFusion doesn’t have work as Spark computes them (from Apache DataFusion’s datafusion-spark), so a PySpark job’s SQL keeps its words: format_string, pmod, parse_url, try_parse_url, sha1, sha2, crc32, xxhash64, bin, bit_count, bit_get, shiftleft, shiftright, luhn_check, soundex, elt, str_to_map, json_tuple, make_dt_interval, date_add, date_sub, last_day, next_day, from_utc_timestamp, to_utc_timestamp, collect_list, collect_set, try_sum and more (SHOW FUNCTIONS lists them all).

SELECT format_string('%s x %d', 'tea', 3) AS said, pmod(-7, 3) AS m, parse_url('https://example.com/a?q=1', 'HOST') AS host;

said     m  host
tea x 3  2  example.com

Where both have a name and differ (Spark’s floor of a DOUBLE is a BIGINT; its substring from a negative start counts from the end), DataFusion’s is the one you get, unless the session speaks Spark: after SET datafusion.sql_parser.dialect = 'spark' (or 'databricks'), Spark’s are, and so are its EXTRACT and SUBSTRING. Spark’s are also there under names of their own (spark_floor, spark_substring, …), which a query in Spark’s grammar, spark_sql('…'), calls where it says floor:

SELECT floor(2.5) AS pondra, spark_floor(2.5) AS spark, s.* FROM spark_sql('SELECT floor(-2.5) AS f, substring("Spark SQL", -3) AS s') s;

pondra  spark  f   s
2.0     2      -3  SQL

JSON

A JSON document is a VARCHAR column, or a JSON (or VARIANT) column, which holds JSON text. Each function takes the document, then a path: keys for objects and positions for arrays. A path that isn’t there gives NULL.

Function	Returns
json_get_str(doc, path…)	the string at the path
json_get_int(doc, path…)	the whole number
json_get_float(doc, path…)	the number as a DOUBLE
json_get_bool(doc, path…)	the boolean
json_get_array(doc, path…)	the array
json_get_json(doc, path…)	the value as JSON text
json_get(doc, path…)	the value, in whatever type it has
json_as_text(doc, path…)	the value as text, whatever its type
json_contains(doc, path…)	true if the path exists
json_length(doc, path…)	the length of the array or object
json_object_keys(doc, path…)	the keys of the object
doc -> path	the same as json_get; chain it: doc -> 'a' -> 'b'
doc ->> path	the same as json_as_text

CREATE TABLE events (id BIGINT, payload VARCHAR);
INSERT INTO events VALUES
  (1, '{"user": {"name": "ann"}, "amount": 4.5, "tags": ["a", "b"]}'),
  (2, '{"user": {"name": "bo"}, "amount": 2.0, "tags": []}');

SELECT id,
       json_get_str(payload, 'user', 'name') AS name,
       payload ->> 'amount' AS amount,
       json_length(payload, 'tags') AS tags
FROM events
WHERE json_contains(payload, 'user')
ORDER BY id;

id  name  amount  tags
1   ann   4.5     2
2   bo    2.0     0

Caution

The typed getters don’t convert: json_get_str on a number is NULL, and json_get_int on 4.5 is NULL. Use json_get_float for any number, or ->> to read anything as text and CAST it. -> and json_get give a value that shows as {int=7}; use it to go deeper into a document, and finish with ->> or a typed getter.

Vectors

A vector is a FLOAT[] column. The distance functions take two vectors of the same length and give a DOUBLE. Vectors of different lengths are an error. If either side is NULL or holds a NULL, the answer is NULL, and so is a cosine of a vector of zeros, which has no direction. Each also answers to DuckDB’s names, so SQL written for DuckDB runs as it is.

Function	Meaning	Also called
cosine_similarity(a, b)	1 for the same direction, 0 for unrelated, -1 for opposite	array_cosine_similarity, list_cosine_similarity
cosine_distance(a, b)	1 - cosine_similarity(a, b): 0 for the same direction	array_cosine_distance, list_cosine_distance
l2_distance(a, b)	Euclidean distance: 0 for the same vector	array_distance, list_distance
dot_product(a, b)	the sum of the products of each pair of numbers	inner_product, array_inner_product, list_inner_product

An array literal, [1.0, 0.0, 0.0], is a vector, so “most like this” is an ORDER BY:

CREATE TABLE docs (id BIGINT, title VARCHAR, v FLOAT[]);
INSERT INTO docs VALUES
  (1, 'tea', [1.0, 0.0, 0.0]),
  (2, 'cake', [0.0, 1.0, 0.0]),
  (3, 'coffee', [0.9, 0.1, 0.0]);

SELECT title, cosine_similarity(v, [1.0, 0.0, 0.0]) AS score
FROM docs ORDER BY score DESC LIMIT 2;

title   score
tea     1.0
coffee  0.9938837341719244

There is no vector index yet, so a search reads every row.

Models

ai_complete and ai_embed call an OpenAI-compatible endpoint: your own Ollama or vLLM, or a hosted one. Pondra carries no model.

Function	Returns
ai_complete(prompt [, model])	the model’s answer, as text
ai_embed(text [, model])	the embedding, as a FLOAT[] to store in a column

The node reads where to call from PONDRA_AI_URL (default http://127.0.0.1:11434/v1), PONDRA_AI_KEY (if the endpoint wants a key), and the default models PONDRA_AI_MODEL (llama3.2) and PONDRA_EMBED_MODEL (nomic-embed-text). See environment variables. Each distinct text in a batch is asked once, embeddings many to a request; a node has at most PONDRA_AI_CALLS requests in flight, and tries one again after a 429, a 5xx or a lost connection. A row whose call still fails is NULL, so one bad row doesn’t lose the query, and pondra_ai_requests_total in /metrics counts the requests answered, failed and tried again.

SELECT id, ai_complete('Summarize in five words: ' || body) AS summary FROM tickets;
UPDATE tickets SET v = ai_embed(body) WHERE v IS NULL;

Caution

With no endpoint listening, every row is NULL and the query still succeeds. If a model function answers NULL for every row, check PONDRA_AI_URL first.

Files

A file is an object in the lake, next to the tables: an image, a PDF, a model. Upload it with PUT /files/<path> (see HTTP), then list and read it in SQL. Files are stored under files/, and a path that exists can’t be written again.

Function	Returns
files()	a table of every file: path, size, written
files('dir/')	the same, for the paths that start with the prefix. files('files/dir/') is the same
file_read(path)	the file’s bytes as BINARY, or NULL if there is no such file. file_read('files/dir/a.txt') is the same as file_read('dir/a.txt')
byte_length(bytes)	the number of bytes in a BINARY value, as a BIGINT

curl -s -X PUT http://localhost:8080/files/docs/hello.txt --data-binary 'hello world'

SELECT path, size, byte_length(file_read(path)) AS bytes, CAST(file_read(path) AS VARCHAR) AS text
FROM files('docs/');

path                  size  bytes  text
files/docs/hello.txt  11    11     hello world

A file’s path in the answer starts with files/, the folder it is kept in. A prefix or a path you give may start with files/ or not.

Binary data

A BINARY value is what file_read gives, and what a BINARY column holds (BYTEA, VARBINARY and BLOB are the same type). These functions take it:

Function	Returns
byte_length(b)	the number of bytes
substr(b, start, len), substring(b FROM start FOR len)	that part of the bytes, counting from 1
sha256(b)	the SHA-256 digest, as BINARY
md5(b)	the MD5 digest, as hex text
encode(b, 'hex' | 'base64' | 'base64pad')	the bytes as text
decode(text, 'hex' | 'base64' | 'base64pad')	the text as bytes
CAST(b AS VARCHAR)	the bytes read as UTF-8 text

SELECT md5(file_read(path)) AS md5, encode(sha256(file_read(path)), 'hex') AS sha256
FROM files('files/docs/');

md5                               sha256
5eb63bbbe01eeed093cb22bb8f5acdc3  b94d27b9934d3e08a52e52d7da7dabfac484efe37a5380ee9088f7ace2efcde9

SELECT CAST(substr(file_read(path), 1, 5) AS VARCHAR) AS word, byte_length(substring(file_read(path) FROM 7)) AS rest
FROM files('files/docs/');

word   rest
hello  5

Caution

CAST(x AS BYTEA) makes bytes. BINARY, VARBINARY and BLOB name a column’s type in CREATE TABLE, but a CAST takes only BYTEA. Other string functions don’t take bytes: CAST(b AS VARCHAR) first, then use the string function.

Dates and times

DataFusion’s to_timestamp turns text or a number into a time. It answers a TIMESTAMP, a wall-clock time with no zone, as DataFusion, Spark and DuckDB’s strptime do. to_timestamp_seconds, to_timestamp_millis, to_timestamp_micros and to_timestamp_nanos answer the same. For text that names a zone, the answer is UTC’s time of day:

SELECT to_timestamp('2020-09-08T13:42:29+02:00') AS ts;

ts
2020-09-08T11:42:29

A TIMESTAMPTZ column, and now(), are still TIMESTAMPTZ, an instant in UTC. Comparing a TIMESTAMPTZ column with to_timestamp(…) works:

CREATE TABLE pings (id BIGINT, at TIMESTAMPTZ);
INSERT INTO pings VALUES (1, '2020-09-08 09:00:00+00'), (2, '2020-09-08 13:00:00+02');
SELECT id FROM pings WHERE at < to_timestamp('2020-09-08T12:00:00+02:00') ORDER BY id;

id
1

Files outside the lake

Table functions read files in place, and give a table. Their names, and the DuckDB names that mean the same, are on the names page.

Function	Reads
read_parquet(url…)	Parquet files
read_csv(url…)	CSV files
read_json(url…)	newline-delimited JSON
read_delta(url)	a Delta Lake table
read_iceberg(url)	an Iceberg table

A URL is a path, a glob (sales/*.csv), a list (['a.csv', 'b.csv']) or an s3://, gs://, az:// or https:// address. Options come by name, after the URLs:

Function	Options
read_parquet, read_json	hive_partitioning, union_by_name, columns, hive_types
read_csv	header (default true: the first line names the columns), delim (or sep, delimiter), quote, escape, comment, new_line, hive_partitioning, union_by_name, columns, hive_types
read_delta	version
read_iceberg	version, snapshot_from_id, snapshot_from_timestamp, allow_moved_paths

• columns => {'id': 'BIGINT', 'name': 'VARCHAR'} names the columns and their types. A CSV’s are matched by position, JSON’s and Parquet’s by name.
• hive_types => {'day': 'DATE'} names the keys of the key=value folders and their types. It turns on hive_partitioning.
• comment => '#' (read_csv) skips the lines that start with it.
• new_line => '\r' (read_csv) is only needed for files whose lines end in \r.

An option Pondra doesn’t know is an error that lists the ones it does. A bare quoted name in FROM (FROM 'sales.csv') picks the reader from the file’s extension.

To keep a name for files instead of spelling the reader out each time, see CREATE EXTERNAL TABLE.

A path on the node’s machine is read only by the program that started the node: the shell, pondra.local() or pondra sql. A query over HTTP is refused.

printf 'id,name\n1,tea\n2,cake\n' > sales.csv

import pondra
db = pondra.local("scratch")
print(db.sql("SELECT * FROM read_csv('sales.csv', header => true) ORDER BY id").rows())
db.close()

[{'id': 1, 'name': 'tea'}, {'id': 2, 'name': 'cake'}]

A file with a comment line and its own delimiter, read with the columns and types you name:

printf '# exported by the till\nid;name\n1;tea\n2;cake\n' > till.csv
pondra sql --dir scratch "SELECT * FROM read_csv('till.csv', delim => ';', comment => '#', columns => {'id': 'BIGINT', 'name': 'VARCHAR'})"

+----+------+
| id | name |
+----+------+
| 1  | tea  |
| 2  | cake |
+----+------+

Secrets

Function	Returns
secrets()	a table of the lake’s secrets: their names, types and scopes, never their values

SELECT * FROM secrets();

Routines

Function	Meaning
pondra.start('name', arg…)	run a procedure without waiting, and answer with its run’s id

CALL p(args) waits for the procedure to finish. pondra.start returns at once with an id you can find in pondra.runs (see system columns). Add AS name to name the column. As a statement, START CALL p(args) does the same.

CREATE TABLE stamps (n BIGINT);
CREATE PROCEDURE stamp() LANGUAGE sql AS $$ INSERT INTO stamps VALUES (1); $$;
SELECT pondra.start('stamp') AS run_id;
START CALL stamp();

Code inside a Python function or procedure has more: pondra.sql, pondra.call and pondra.secret. They are in the Python reference.

Exact sums

sum over DOUBLE gives the same answer whatever the order the rows arrive in, and on any number of nodes: the exact total, rounded once. Ordinary floating-point addition can differ in its last digits from one run to the next. Sums of integers and decimals, and sum(DISTINCT …), are DataFusion’s own.

SELECT sum(x) AS total FROM (VALUES (0.1), (0.2), (0.3)) AS t(x);

total
0.6

Functions you bring

A function can also run in a process of your own, as an Arrow Flight server: Pondra sends it the rows as one batch, and reads one column back. Register it with POST /functions/<name>, with the server’s address, the argument types and the return type. GET /functions lists them, and DELETE /functions/<name> takes one away. The definition is kept in the lake, so every node has the function as soon as your POST is answered, and loses it as soon as you DELETE it.

import json, threading, urllib.request
import pyarrow as pa
import pyarrow.flight as fl

class Shout(fl.FlightServerBase):
    def do_exchange(self, context, descriptor, reader, writer):
        column = reader.read_all().column(0).to_pylist()
        answer = pa.table({"out": [None if v is None else v.upper() for v in column]})
        writer.begin(answer.schema)
        writer.write_table(answer)

server = Shout("grpc://127.0.0.1:0")   # port 0: any free port
threading.Thread(target=server.serve, daemon=True).start()

body = {"flight": f"http://127.0.0.1:{server.port}", "args": ["Utf8"], "returns": "Utf8"}
urllib.request.urlopen(urllib.request.Request("http://localhost:8080/functions/shout", data=json.dumps(body).encode(), method="POST"))

db = pondra.connect("http://localhost:8080")
print(db.sql("SELECT shout('tea') AS s").rows())

[{'s': 'TEA'}]

The arguments arrive as columns arg0, arg1, and so on. The types are Arrow’s names: Utf8, Int64, Float64, Binary, Float32[]. A function returns one column with as many values as it got rows. An optional "rows": n in the body sets how many rows go in each call.