Module Talon.Query

The type for queries.

of_table t is the query of the rows of the table t. It costs O(1).

of_source s is the query of the rows of the source s, which it reads when it runs. It costs O(1).

schema q is the names and types of the columns of q's rows. The verb that made q resolved it, so it costs O(1) and reads no data.

select os q is q's rows with exactly the columns os, in order:

select
  Expr.[ keep Sel.(names [ "carrier" ]); "late" := delay > float 15. ]

It keeps q's row count and order. Each output reads q's columns, never another output. It streams in O(n), except that Expr.over, Expr.rank and Expr.window over the input's rows need the whole input: the verb then blocks.

Its problems are those of binding os, and two outputs of one name.

derive os q is q's rows with each of q's columns and the outputs os: an output replaces the column of its name in place, with the output's type, and the others follow q's columns, in order. It keeps q's row count and order. Each output reads q's columns, never another output. It streams in O(n), except that Expr.over, Expr.rank and Expr.window over the input's rows need the whole input: the verb then blocks.

Its problems are those of binding os, two outputs of one name, and an output of Expr.within named as a column of q.

filter p q is the rows of q on which p is true, in order, so a row on which p is null is dropped. Its schema is q's. p is a bool column: an OCaml value, as const f $ x is, takes that type. It streams in O(n), except that Expr.over, Expr.rank and Expr.window over the input's rows need the whole input: the verb then blocks.

Over traced columns (Compiled, mapped and differentiated runs) it raises when p reads one.

Its problems are those of binding p, and an extension's values, even when they read as bool.

sort ks q is q's rows ordered by the keys ks in turn, each by talon's total order in its direction, nulls last unless Order.nulls_first (see Order). It is stable: rows equal on every key keep q's order. Its schema is q's. It blocks, in O(n log n).

Its problems are a key that names no column, a key that names the column of an earlier key, a key on a column whose type holds an extension type, which has no order, and a key on an extension column, which orders only through its declaration.

slice ~offset ~length q is q's rows at the positions offset to offset + length - 1 that q has, in order, a negative offset counting from the end: slice ~offset:0 ~length:10 q is q's first ten rows, and slice ~offset:(-10) ~length:10 q its last ten. Its schema is q's. With offset >= 0 it stops reading at the last row it keeps; with a negative offset it holds q's last -offset rows.

Its problem is a negative length.

aggregate ~by os q is one row per group of q's rows that have the same keys in the columns by, by key identity (null is one key, and NaN is one), in order of first appearance. Each row holds the columns by, then the outputs os reduced over the group's rows in q's order. ~by:[] makes one group of every row, so one row, even when q has none.

It takes O(n) expected time. A stream of several batches holds O(groups

1. largest batch) rows, a state of one row per group, where each output reduces with rows, count, sum, mean, var, std, min, max, first, last, only, arg_min or arg_max of an operand that reads its own row alone and holds no per_list. Its values and failures do not depend on the batches. Otherwise it blocks, holding the columns by and those that os read: for any other reduction, a sum of durations, an OCaml function or value ($, const, of_option), an extension value, a Expr.within, an output that can fail outside a reduction, or a traced batch (Compiled, mapped and differentiated runs). Over traced columns it raises unless by is empty. With a var or std it raises Invalid_argument for a traced batch that follows two untraced batches.

Its problems are a key that names no column or is named twice, those of binding os, two outputs of one name, and an output of a key's name.

join ?kind ?each_left ?each_right ~on right left pairs left's rows with the rows of right that on matches, written left |> join ~on right. kind says which rows are kept (Join.kind) and defaults to Join.kind.Inner. each_left and each_right are the numbers of matches each row of left and of right must have (Join.count), Join.count.Any by default. Its rows and columns are those Join describes.

A join on Join.all blocks on right and streams left, and the others block on both inputs. The cost of an equality or range join is in Algorithms.

Its problems are, for each atom of on in order, a column missing on its side and the atom's own (Join.eq, Join.lt, Join.overlaps). Then, in a Join.kind.Full join, a left column that is the left of two equality atoms, and the joined columns that have one name, as Join states.

Raises Invalid_argument over traced columns (Compiled, mapped and differentiated runs) on an equality or range join. The others run, their counts included.

append rest q is q's rows, then rest's, written q |> append rest. The two have the same columns, matched by name. Its schema is q's names, in order, each at the type where the column's two types meet (Type.common): an int32 column and an int64 one append as int64, a categorical and a string as string. Each side's values widen to it exactly. Appending other columns is Kit.union, which derives the missing ones as nulls. It costs O(1) over tables of one schema and streams over sources.

Its problems are a column that only q has, one that only rest has, and each column whose two types do not meet, such as int8 and uint8.

unnest ?keep_empty os q is the rows of the frames of the lists that the outputs os compute (Lists), concatenated in q's row order, each row's positions in list order. Its columns are q's with os, placed as derive places them, an output holding its list's element and every other column repeating the row's value.

A row without positions gives no row. With ~keep_empty:true (default false) it gives one row with every output null, so a null list and an empty one merge, a padded null reads as a null element, and Expr.rows counts that row as one. Lengths that differ, a null beside a non-empty list included, are a data error at the row.

It streams, and blocks where derive would, in O(n + m) for m positions, plus the size of each repeated list or text column once per position of its row. optimize narrows the repeated columns.

Its problems are those of binding os, no output, two outputs of one name, an output that is not a list, and an output of Expr.within named as a column of q.

Raises Invalid_argument over traced columns under Rune.jit, and under Rune.vmap where the lists' offsets or validity are traced (Compiled, mapped and differentiated runs).

fold q ~init f runs q: it reads q's sources, computes q's rows in batches and folds f over them, in order, from init. Each batch has at least one row. The batches f sees depend on the batches of q's inputs; their rows in order do not. It optimizes q first (optimize).

Error e where the run fails: a source's error, or one found in the data, such as a cast that loses a value, only with two values or a join assertion. A run fails as evaluating its optimized plan one row at a time would, at the first row, in that order, where a step fails, whether an error or an exception. Which failure is reported depends on the plan and its inputs' values in order, never on batches or cores. e names the step, as pp prints it, and the row of the step's input. Over traced columns (Compiled, mapped and differentiated runs) a step that can fail on a value, such as a cast that may lose one, raises at the trace.

A function of q's expressions may be called on rows past a limit of q as written: the optimized plan may read them, as a filter it moves below a sort reads every row. A failure on such a row, an error or an exception, ends the run where the optimized plan, evaluated one row at a time, reads the row, whatever the batches, and is never met elsewhere. Every reader fold opens is closed when it returns, also when f, a function of q's expressions or a source raises; the exception then propagates.

run q is q's rows as one batch: fold into a table, then one copy into a single batch. Its columns are canonical (Column), so their Column.layouts do not depend on batches, except that a column whose validity is traced (Compiled, mapped and differentiated runs) keeps it, even with no null, and that a record's fields under its nulls are unspecified.

values e q is e on each row of q, decoded to OCaml: values Expr.(const mk $ carrier $ option origin) q. An extension value is decoded with its declaration's dec. q reads only the columns e reads.

Error e as for run, and where e is null on a row, or a value is outside what OCaml reads (Column.values): read nullable values through Expr.option. e then names the step values followed by e.

Raises Invalid_argument with a report like a verb's, named values, if e does not bind against q's schema, and over traced columns or under Rune.grad (Compiled, mapped and differentiated runs), as nx refuses to read their arrays out.

optimize q is the plan that running q runs: q with the same schema and the same rows, rewritten to read and compute less. Running a query optimizes it first, so optimize serves to print the plan that runs and to compare plans. It reads no data, it calls the sources' pushdown, and optimize (optimize q) is optimize q.

Results. The rewrites change no value and no null of a result, and add no failure: an operation that can fail meets only rows that q shows it. They may remove failures: a value that no row and no column of the result reads is not computed, so a failure, or an exception from a user function, that only such a value meets does not happen. Of the failures that remain, a run reports the one at the earliest row of the optimized plan.

Constants. Arithmetic, comparisons, not, Expr.is_in and the text operations other than Expr.Str.split and Expr.Str.concat, of literals, are evaluated as a run evaluates them, and become the literal they compute. One whose evaluation fails, or gives a value OCaml's type does not hold (an int64 past max_int), stays and evaluates at run time. &&, ||, is_null, if_, coalesce, and store of a literal fold alike. a && false, a || true, a && true, a || false, an if_ on a literal and a coalesce with literals simplify alike. A filter whose predicate is true goes, and one with a conjunct false or null becomes slice ~offset:0 ~length:0, since a row passes only if every conjunct is true.

Predicates. A filter's predicate splits into its conjuncts, the operands of its &&s. A conjunct moves toward the sources when it is row-local, its value at a row depending on that row alone: outside the bodies of its Expr.per_lists, it has no Expr.over, Expr.shift, Expr.rank or Expr.window. It moves past each step that keeps the rows it sees and the values it reads:

• a sort;
• a filter whose predicate is row-local, with which it merges;
• a select or a derive whose outputs are row-local, when each column the conjunct reads is a column of the step's input kept unchanged, under its name or another (select ["y" := x]), or an output that is a literal or null, and the renaming hands no Expr.per_list body's read of a column to one of its outputs. Such an output is read as its value and the conjunct folds, which simplifies a predicate over constant columns: over a fold of files with partition columns derived as literals, a conjunct that folds to false or null empties that branch before its source is read, and false || x becomes x, which the source can use;
• an aggregate with keys, when the conjunct reads only keys and none of their types holds floats, since -0. and 0. are one key with two values;
• an append, into both inputs;
• a join whose ~each_left and ~each_right are Any, on a condition other than Join.position: into the left input when the conjunct reads only left columns and the join is Inner, Left, Semi or Anti, and into the right input when it reads only right columns and the join is Inner.

It stops at a slice, at an unnest, at a Full join, at a join with an assertion, at a step with an output or predicate that is not row-local, and at a table. Pushdown never crosses a join with an assertion: a conjunct entering one input changes the matches of the other input's rows, and so could make that side's assertion fail on rows the plan shows it. A conjunct that can fail, one with a cast that narrows, Str.parse, temporal arithmetic or parsing, of_option or $, or a Expr.per_list of several lists or with Expr.only or a duration Expr.sum, passes only the steps that show it the rows they receive: a sort, a select or a derive, an aggregate's keys and an append. Below a filter or a join it would meet rows that q never shows it.

When it reaches a source, a conjunct that compares a column with a literal, tests it with Expr.is_in or Expr.is_null, or combines such tests with &&, || and not, is a Source.Pred.t, and it is offered to the source's pushdown: an Exact conjunct leaves the plan and goes into the source's request, an Inexact one goes into the request and stays, and an Unsupported one stays. The conjuncts that stay at one place form filters in plan order, a conjunct that can fail in a filter above those it must not pass.

Slices. A slice moves below a select or a derive whose outputs are row-local, and above the conjuncts that move. Two slices from the start, offset >= 0, merge into one. A slice from the start of an append limits both its inputs to offset + length rows. A slice from the start directly above a source gives it a limit of offset + length rows. A conjunct that stays is a filter between them, so a source with an Inexact or Unsupported conjunct gets no limit. A slice from the end gets one too when the source states its rows and is handed no conjunct, the slice then counting from the start. A slice from the start directly above a sort runs as a selection of its first offset + length rows, which sorts only the rows whose first key is at most the offset + lengthth row's: sort then slice is a top-k.

Projections. Each source reads only the columns that a step reads or that the result has, each select, derive and aggregate drops the outputs that nothing reads, and an unnest repeats only the columns read above it, though it keeps every output, which sets its rows. A select that keeps its input's columns unchanged, in order, goes, and so does a derive left without outputs, and a select that only keeps columns below a select or an aggregate, which read their input by name. Where an input of an append has columns that the append does not take, select [keep (names …)] keeps those it takes.

Sharing. Subplans that are the same steps with equal arguments over physically the same tables and sources become one value, which a run runs once; equal subexpressions of one step are one expression, which it computes once. Each place that reads a source is a read of its own, with the columns and the conjuncts of that place, so places with equal requests are one read. A step that a plan reaches more than once runs once: a run computes it as far as its furthest reader reads, and holds its rows, of the columns its readers read, from its slowest reader to its furthest.

The guide's pipeline, optimized, reads two of the CSV file's columns, which answers Unsupported to every conjunct:

query → carrier string, mean_delay float64, flights int64, name string
sort [desc "mean_delay"]
└ join ~on:(keys ["carrier"]) ~each_left:One
  ├ aggregate ~by:["carrier"] ["mean_delay" := mean dep_delay;
  │                            "flights" := rows]
  │ └ filter (dep_delay > 15.)
  │   └ csv "flights.csv" (19 columns) ~columns:["dep_delay"; "carrier"]
  └ parquet "carriers.parquet" (2 columns)

equal q0 q1 is true iff q0 and q1 are the same plan: the same steps with equal arguments, expressions compared by their identity, tables by key identity row by row, and sources physically, with equal requests (see pp). Equal queries have equal schemas.

pp ppf q formats q's plan, which reads no data: the line query → and q's schema as Schema.pp formats it, then q's last step, with the steps it reads below it in a tree:

query → carrier string, mean_delay float64, flights int64, name string
sort [desc "mean_delay"]
└ join ~on:(keys ["carrier"]) ~each_left:One
  ├ aggregate ~by:["carrier"] ["mean_delay" := mean dep_delay;
  │                            "flights" := rows]
  │ └ filter (dep_delay > 15.)
  │   └ csv "flights.csv" (19 columns)
  └ parquet "carriers.parquet" (2 columns)

A step formats as the call of the verb that made it, without its input or module paths, its expressions as Expr.pp formats them, its keys as they are written inside Order.( … ) and its condition as it is written inside Join.( … ):

• outputs as "name" := e, with selectors resolved, and a run of columns kept unchanged under their names as one keep (names ["a"; "b"]);
• ~kind, ~each_left and ~each_right after ~on, when they are not their defaults;
• a table as table (4 columns, 16 rows), and a source as its name and its number of columns, then its number of rows when it states one: parquet "carriers.parquet" (2 columns, 1491 rows);
• a source that is asked for less than all of it, as an optimized plan asks (optimize), then as its request: ~columns when it reads fewer than all its columns, ~filters when it is handed conjuncts, and ~limit when it has one: parquet "f.parquet" (19 columns) ~columns:["dep_delay"; "carrier"] ~filters:[dep_delay > 15.] ~limit:10.

A join's left input comes before its right, and append's q before rest. A step that the plan reaches more than once, as equal subplans are after optimize, is labelled #1, #2, … in the order the printer first meets them: the first time as #1 followed by the step and its inputs, and each later time as #1 alone:

query → k string, n int64, m int64
join ~on:(keys ["k"])
├ #1 aggregate ~by:["k"] ["n" := rows]
│ └ csv "x.csv" (3 columns) ~columns:["k"]
└ select [keep (names ["k"]); "m" := n]
  └ #1

A step too long for the margin continues on the next lines, indented under it. Lines fit the formatter's margin counted from column 0, since a formatter does not tell its current indentation: inside an indented box they overrun the margin by that indentation.