Module Talon.Source

Sources: tables read in batches.

A source is data that a query reads when it runs, such as a file: its schema is known when it is built, and its rows arrive in batches. Formats build sources with v, and a program's own format is a peer of the shipped ones. Query.of_source makes a query of one.

A source is a contract between talon and its author:

Pushdown. Before reading, talon asks the source about each conjunct of the filters on it, translated into a Pred.t, and the source answers with no IO whether it can apply it (answer).
Request. Talon then asks for the source's parts with a request: the columns it reads, the conjuncts the source can use, and a limit. Each place in a plan that reads the source is a read of its own, with its own request.
Pull. Talon opens a part's reader, pulls batches with next, and calls close once it has no more use for the reader.

The rows of a source are those of its parts in part order, then batch order. They are fixed by the data and the request, never by the machine's core count.

Pushdown

#type answer =

#| Exact
(*
The source applies the conjunct exactly: it yields no row that fails it, so talon removes the conjunct from the plan.
*)
#| Inexact
(*
The source may use the conjunct to skip rows that fail it, as row group statistics do, and never skips a row that passes it. Talon applies the conjunct again.
*)
#| Unsupported
(*
The source ignores the conjunct. Talon applies it.
*)

The type for a source's answers about a conjunct.

#module Pred : sig ... end

Predicates on one row: the filters that sources apply.

Reading

#type request = {

#columns : string list;
(*
The columns talon reads, distinct, in schema order. It may be empty, as when a query only counts rows.
*)
#filters : Pred.t list;
(*
The conjuncts that the source answered Exact or Inexact, in plan order. The source applies each it answered Exact.
*)
#limit : int option;
(*
Some n when talon reads no more than the first n rows that the request yields, so the source may stop after them. It is None unless every conjunct of the filters on the source is Exact.
*)

}

The type for what talon asks of a source when it runs.

#type reader = {

#next : unit -> (table option, Error.t) Stdlib.result;
(*
next () is Ok (Some b) with the next batch b of the part, whose columns are the request's, in its order, with the source's types; Ok None at the end of the part; or Error e. A batch may have no rows. A batch with other columns or types raises Invalid_argument when talon reads it.
*)
#close : unit -> unit;
(*
close () releases the reader. Talon calls it once, when next has returned Ok None or Error _, when it needs no more of the part's rows, or when the run fails, and calls next no more afterwards.
*)

}

The type for readers of a part's batches.

#type part = {

#rows : int option;
(*
Some n promises that the part yields exactly n rows for the request, and talon trusts it to skip the part without reading it; None if unknown.
*)
#open_ : unit -> (reader, Error.t) Stdlib.result;
(*
open_ () opens a reader on the part. Talon opens a part at most once, and parts must not depend on one another.
*)

}

The type for parts: the units a source reads, such as files or row groups.

Sources

#type t

The type for sources. Two sources are the same iff they are one value.

val v : 
  name:string ->
  schema:Schema.t ->
  ?rows:int ->
  ?sorted:Order.t list ->
  ?pushdown:(Pred.t -> answer) ->
  (request -> (part list, Error.t) Stdlib.result) ->
  t

v ~name ~schema ?rows ?sorted ?pushdown parts is the source called name that yields rows of the columns schema, with:

rows, the number of rows the source yields for a request without filters, which plans print, and which turns a slice from the end into one from the start. Defaults to unknown.
sorted, the order the rows come in, by talon's total order. Talon checks it as rows arrive, and a row out of order fails the run. Defaults to no order.
pushdown, which answers for one conjunct whether the source applies it. Talon calls it when it plans a run, any number of times: it must be pure and do no IO. Defaults to Unsupported for every conjunct.

Talon calls parts once for each place in the optimized plan that reads the source, with that place's request, and parts gives the source's parts in order, or the Error that fails the run. A source that can be read only once, such as a stream, documents that it reads once, and its parts is an Error after its first call: a plan that needs it in several places reads a table that run makes of it first.

Plans print a source as name followed by its number of columns, and by rows when given, as in parquet "carriers.parquet" (2 columns), so name says what the source reads: csv "flights.csv".

Raises Invalid_argument if name is empty, is not valid UTF-8 or holds a control character, if rows is negative, or if a key of sorted names no column of schema, names a column twice, or names a column whose type is or contains an extension type, which has no order of its own.

Paths

A directory of a partitioned dataset names partition values in its path, as Hive writes them: year=2024/month=01/part-0.parquet. This layout has no specification, and what the partition values mean, like their types, is the caller's: see hive.

#val hive : string list -> (table, Error.t) Stdlib.result

hive paths is the table of the keys and values in paths, one row per path in its order: the column "path", then one string column per key of the first path, in the order they appear in it. A path is split at each /, and every part but the last is a directory; a directory holding = is a key and a value, split at its first = and percent-decoded by bytes, where only a pair of hex digits decodes. A value __HIVE_DEFAULT_PARTITION__ is null. Each file's keys are matched by name across paths.

Values are text. Types are the caller's: Expr.Str.parse them, so one directory cannot change a column's type for its siblings.

Error e on a path that is not valid UTF-8; an empty key or the key "path", which holds the file's path; a key or value not valid UTF-8 once decoded; a key repeated in one path; or paths whose keys differ. An invalid percent sequence is not an error and stays literal. It is pure and reads no file: paths are any strings.