Formats

Sources

Query.of_source s makes a query of a source. Running the query asks the source for the columns it needs, offers it the filters it may apply, and reads its rows batch by batch. Each place in a plan that reads a source is a read of its own, and a read of a file opens it again.

open Talon

let ( let* ) = Result.bind

let late flights =
  Query.(of_source flights |> filter Expr.(Col.int "dep_delay" > int 15))

let late_flights () =
  let* flights = Talon_csv.file ~nulls:[ "NA" ] "flights.csv" in
  Query.run (late flights)

CSV

Talon_csv.file path returns the source of a CSV file. Without a format, it sniffs one from the file's first 16,384 records:

• the separator: ,, a tab, ; or |, whichever splits every record into the same number of fields, and into the most;
• the header: the first record names the columns, unless, in every column whose values are not text, its field reads as that column's type;
• the types: bool, int64, float64, date, datetime[us] (or datetime[us, UTC] with offsets) when every value reads as one, and string otherwise.

A number with a leading zero, such as 007, is text, since it writes an identifier. Sniffing never infers a categorical.

An unquoted empty field is null, and so is one equal to a null token given with ~nulls. A quoted field is never null: "" is the empty string. Fields follow RFC 4180, strictly: a quoted field may hold separators, line breaks and doubled quotes.

To see or change a format, sniff it from a reader:

let csv =
  String.concat "\n"
    [
      "station,day,temp,note";
      "oslo,2024-03-01,-3.5,";
      {|oslo,2024-03-02,NA,"sensor ""B"" offline"|};
      {|lima,2024-03-01,22,"humid, calm"|};
    ]

let reader () = Bytesrw.Bytes.Reader.of_string csv
let format = Error.get_ok (Talon_csv.sniff ~nulls:[ "NA" ] (reader ()))
let () = Format.printf "%a@." Talon_csv.pp_format format

csv (4 columns, separator ',', quote '"', header, nulls ["NA"]), types sniffed from 3 rows
  station string
  day date
  temp float64
  note string

Talon_csv.with_type declares another type for a column, and Talon_csv.format states a whole format. Talon_csv.decode reads a reader into a table, and Talon_csv.file ~format reads a file as the format says:

let format = Talon_csv.with_type "temp" (Type.Any Type.float32) format
let readings = Error.get_ok (Talon_csv.decode format (reader ()))

table 3 rows × 4 columns
 station  day         temp      note
 string   date        float32   string
 oslo     2024-03-01  -3.50000  ∅
 oslo     2024-03-02         ∅  sensor "B" offline
 lima     2024-03-01  22.00000  humid, calm

A field that does not read as its column's type fails the read. The error gives the line, the column, the text and the fix, and says when the type was sniffed:

line 3, column 6: "warm": column "temp": cannot read as float64: not a number. Declare the null token (~nulls) or another type (with_type).

Talon_csv.encode runs a query and writes its rows in a format, which decode reads back. A field is quoted when it holds the separator, the quote or a line break, or when it is empty text:

let buf = Buffer.create 256

let () =
  Error.get_ok
    (Talon_csv.encode format (Query.of_table readings)
       (Bytesrw.Bytes.Writer.of_buffer buf))

station,day,temp,note
oslo,2024-03-01,-3.5,
oslo,2024-03-02,,"sensor ""B"" offline"
lima,2024-03-01,22,"humid, calm"

JSON

Talon_json reads objects, one per row, from a Bytesrw reader, in one of two framings: Lines, one object per line, which NDJSON and JSON Lines write, and Array, a single JSON array of objects. Talon_json.sniff infers the framing from the first value and each column's type from the file's first 16,384 rows. It reads an object as a record, an array as a list, a number as int64 when every value is an integer int64 holds and as float64 otherwise, and a column of mixed values as string, holding each value's JSON text:

let events =
  {|{"id":1,"user":{"name":"ana"},"tags":["a","b"],"score":1.5}
{"id":2,"user":null,"tags":[],"score":null}
|}

let reader () = Bytesrw.Bytes.Reader.of_string events
let format = Error.get_ok (Talon_json.sniff (reader ()))
let () = Format.printf "%a@." Talon_json.pp_format format

json (4 columns, lines), types sniffed from 2 rows
  id int64
  user record[name string]
  tags list[string]
  score float64

A key the format does not name is skipped, and a column whose key a row lacks is null. A duplicate key fails the read. Talon_json.with_type declares a date or a datetime, which sniffing never infers, and Talon_json.source streams the rows in batches, so a large top-level array yields rows before its closing bracket is read:

let table = Error.get_ok (Talon_json.decode format (reader ()))
let () = Format.printf "%d rows@." (Talon.rows table)

2 rows

Talon_json.encode runs a query and writes its rows, one object per line or one array, which decode reads back.

To read a gzip or zstd file, apply a filter of Lz_bytesrw to the reader. Talon_json.file opens a path, as Talon_csv.file does.

Parquet

Talon_parquet.file path maps a Parquet file in memory and reads its footer. The source reads only the column chunks a query needs, reads uncompressed, Snappy, gzip, Zstandard and LZ4_RAW chunks, and states its number of rows.

Each Parquet column reads as the talon type that holds its values: boolean as bool, int32 and int64 as their annotated integer type, float and double as float32 and float64, DATE as date, TIMESTAMP as datetime of its unit (in UTC when adjusted to UTC), strings as string, and other byte arrays as binary. Talon_parquet.sniff reads the format from a file's bytes, which Rig_disk.of_file opens:

let carriers () =
  let path = "carriers.parquet" in
  let* bytes = Rig_disk.of_file path |> Result.map_error (Error.v ~file:path) in
  let* format = Talon_parquet.sniff bytes in
  Format.printf "%a@." Talon_parquet.pp_format format;
  Ok (Talon_parquet.source format bytes)

parquet (2 columns)
  carrier string ← optional binary (STRING)
  name string    ← optional binary (STRING)

A DECIMAL column has no talon type of its own: declare it with Talon_parquet.with_type as float64, or, up to 18 digits, as int64 for its unscaled integer. Talon refuses encrypted files and the codecs it does not read when it sniffs the footer.

A Parquet source uses filters to skip row groups. A comparison of a column with a literal goes to the source, which skips each row group whose minimum and maximum show that no row passes; the filter stays in the plan for the rows of the groups it reads. A filter on one text column leaves the plan: the source applies it as it reads, once per value of a dictionary-encoded column chunk. Query.optimize shows it:

query → carrier string, name string
parquet (2 columns) ~filters:[carrier >= "F"]

Writing

Talon_parquet.encode runs a query and writes its rows as a Parquet file. Talon_parquet.format states the format from the query's schema, with the codec and the row-group size talon uses, and defaults for both:

let write_snapshot q =
  Out_channel.with_open_bin "snapshot.parquet" @@ fun oc ->
  let w = Bytesrw.Bytes.Writer.of_out_channel oc in
  Talon_parquet.encode (Talon_parquet.format (Query.schema q)) q w

Every column is written optional, so a column with nulls and one without have one type in the file. Each type writes as the Parquet type that reads back as it, and types with no Parquet type of their own are described in the module's Writing section. A duration has no unit a Parquet file can record, so Talon_parquet.format refuses it: select its ticks first with Expr.cast Type.int64 (Col.span "d"), and record the unit in the column's name.

Many files as one table

A dataset is often a directory of Parquet files, such as year=2024/month=01/part-0.parquet, one per partition. A query over them is a fold of one query per file. This section builds one, step by step, over a small dataset:

let ( // ) a b = a ^ "/" ^ b

let dataset = Filename.temp_dir "talon-guide" ""

let write_part path rows =
  Sys.mkdir (Filename.dirname path) 0o755;
  Out_channel.with_open_bin path @@ fun oc ->
  let w = Bytesrw.Bytes.Writer.of_out_channel oc in
  let t =
    Talon.v [ ("x", Column.v Type.int64 (Array.of_list rows)) ]
  in
  Result.get_ok
    (Talon_parquet.encode
       (Talon_parquet.format (Schema.v [ ("x", Type.Any Type.int64) ]))
       (Query.of_table t) w)

let () =
  write_part (dataset // "year=2024" // "part-0.parquet") [ 1; 2; 3 ];
  write_part (dataset // "year=2025" // "part-0.parquet") [ 4; 5 ];
  write_part (dataset // "year=2026" // "part-0.parquet") [ 6; 7; 8 ]

Listing

Talon has no directory walk of its own: the listing is a few lines that depend on the dataset, such as which files count. Sort the listing by a key built with /, so year=1 sorts before year=10 on every platform, and skip the entries a writer leaves behind:

(* A name starting with [.] or [_] is a writer's bookkeeping, such as
   [_SUCCESS]. A directory [_key=value] holds a partition, not bookkeeping. *)
let hidden ~directory name =
  name.[0] = '.'
  || (name.[0] = '_' && not (directory && String.contains name '='))

let rec walk root relative =
  let dir = if relative = "" then root else root // relative in
  Array.to_list (Sys.readdir dir)
  |> List.concat_map @@ fun name ->
     let below = if relative = "" then name else relative // name in
     let directory = Sys.is_directory (root // below) in
     if hidden ~directory name then []
     else if directory then walk root below
     else [ below ]

let paths root =
  List.map (( // ) root) (List.sort String.compare (walk root ""))

# paths dataset |> List.map (fun p -> String.sub p (String.length dataset) (String.length p - String.length dataset));;

- : string list =
["/year=2024/part-0.parquet"; "/year=2025/part-0.parquet";
 "/year=2026/part-0.parquet"]

Sys.is_directory follows symbolic links, so a link to a directory lists it again; a listing that needs otherwise uses Unix.lstat.

Partition values

Source.hive decodes the partition values in the paths into a table, one row per path. Its values are text, percent-decoded, with __HIVE_DEFAULT_PARTITION__ as null. Types are the caller's, through Str.parse, so a new directory cannot change a column's type:

let manifest =
  let* t = Source.hive (paths dataset) in
  Query.of_table t
  |> Query.derive
       Expr.[ "year" := Str.parse Type.int64 (Col.string "year") ]
  |> Query.run

let manifest = Error.get_ok manifest

Filter the manifest first

Talon_parquet.file reads a file's footer when the source is built, and a source holds its file's mapping for as long as it lives. Filter the manifest before building the sources, so only the files a query can read cost one:

let recent =
  Query.of_table manifest
  |> Query.filter Expr.(Col.int "year" >= int 2025)
  |> Query.run
  |> Error.get_ok

One query over the files

Each surviving file becomes a query with its partition columns derived as literals, and Kit.union_all appends the branches, casting each column to Type.common over all of them. A filter on a partition column is folded to true or false on each branch, so an excluded file's rows are never read, and a filter mixing partition and file columns reaches each file with the partition part folded away:

let branch path year =
  let value =
    match year with Some y -> Expr.int y | None -> Expr.null
  in
  Query.of_source (Error.get_ok (Talon_parquet.file path))
  |> Query.derive Expr.[ "year" := store Type.int64 value ]

let everything =
  let paths = Column.options Kind.string (column recent "path")
  and years = Column.options Kind.int (column recent "year") in
  Kit.union_all
    (Array.to_list (Array.map2
       (fun p y -> branch (Option.get p) y)
       paths years))

# everything
  |> Query.filter Expr.(Col.int "year" = int 2025 || Col.int "x" > int 7)
  |> Query.run
  |> Error.get_ok
  |> Talon.rows;;

- : int = 3

The sample data is on disk until it is removed:

let () =
  let rec remove path =
    if Sys.is_directory path then begin
      Array.iter (fun name -> remove (path // name)) (Sys.readdir path);
      Sys.rmdir path
    end
    else Sys.remove path
  in
  remove dataset

Arrow IPC

Talon_arrow.file path maps an Arrow IPC file or stream in memory. A column's buffers are shared with the file where they can be, and the batches of the file become the batches of the source:

let events_arrow () =
  let* events = Talon_arrow.file "events.arrow" in
  Query.run (Query.of_source events)

Each Arrow type reads as the talon type that holds its values: dictionary columns as categoricals when one dictionary of distinct strings without nulls stands for the whole file and otherwise as their value type, fixed-size lists of numbers as tensors, and extension fields as extensions over their storage. A decimal field is undeclared until Talon_arrow.with_type gives it a type, as float64 or, up to 18 digits, as int64 for its unscaled integer, as in a Parquet file; a query that leaves one undeclared is refused when it is built. A field talon cannot hold is refused when the file is opened, naming the column. Talon_arrow.with_type also reads a fixed-size list as a list when its elements can be null.

Talon_arrow.encode writes a query's rows. Talon_arrow.format states the format from the query's schema, with the framing (a file by default, or a stream), the buffer compression and the number of rows of each record batch:

let write_events q =
  Out_channel.with_open_bin "events.arrow" @@ fun oc ->
  let w = Bytesrw.Bytes.Writer.of_out_channel oc in
  Talon_arrow.encode (Talon_arrow.format (Query.schema q)) q w

A file's bytes depend only on its rows and its format. Custom metadata of the schema and of each field travels with the format, read by Talon_arrow.metadata and set by Talon_arrow.with_metadata.

Errors

Failures in files are Talon.Error.t values. An error holds a message and the places it was found: the file, the line and column of its text, the Parquet row group, the byte range, and the raw text that failed. Error.pp prints them coarsest first, as in flights.csv:48213:12: "NA": …, and Error.get_ok raises Failure with that text.

Your own source

Source.v makes a source of anything that yields tables: a format talon does not ship, a database cursor, a simulation. It takes a name, a schema, and a function from a request to the source's parts, each of which opens a reader of batches. This source counts from 0 in parts of 1,000 rows:

let counter n =
  let schema = Schema.v [ ("i", Type.Any Type.int64) ] in
  let batch (req : Source.request) lo hi =
    if req.columns = [] then Talon.v ~rows:(hi - lo) []
    else Talon.v [ ("i", Column.v Type.int64 (Array.init (hi - lo) (( + ) lo))) ]
  in
  let part req lo hi =
    let open_ () =
      let read = ref false in
      let next () =
        if !read then Ok None
        else (
          read := true;
          Ok (Some (batch req lo hi)))
      in
      Ok { Source.next; close = ignore }
    in
    { Source.rows = Some (hi - lo); open_ }
  in
  let parts req =
    Ok (List.init ((n + 999) / 1000) (fun k -> part req (k * 1000) (min n ((k + 1) * 1000))))
  in
  Source.v ~name:"counter" ~schema ~rows:n ~sorted:[ Order.asc "i" ] parts

let total =
  Query.(of_source (counter 2500) |> aggregate ~by:[] Expr.[ "sum" := sum (Col.int "i") ])

• A batch has the request's columns, in schema order. A query that only counts rows asks for none.
• ~sorted claims an order, which talon checks as rows arrive.
• ~pushdown answers, for each filter, whether the source applies it (Exact), uses it to skip rows (Inexact), or ignores it (Unsupported). It defaults to Unsupported.

A reader's next returns an Error for a failure in the data, and talon calls close once it needs no more of the part.

The example 05-formats reads and writes this page's CSV and reads a Parquet file, and 10-custom-source writes a source that applies filters itself.

this page is tested with mdxedit this page ↗