Local files and Parquet¶

Read and write data on the local filesystem: line/JSON text files (connector='file') and Parquet files or directories (connector='parquet'). Built into the engine core, no external dependency.

Overview¶

These are clink's built-in local-filesystem connectors, registered by clink::core itself rather than an optional impls/ module, so they are always available. Two connector families:

file (alias filesystem): newline-delimited text or JSON records. A single TEXT column is read/written as one line per record (the string channel); a multi-column table with format='json' is read/written as one JSON object per line (the row channel).
parquet: Apache Parquet files. The string/int64 channels read and write single-column files through the shared ArrowBatcher<T> seam; the row channel (format='json') reads and writes typed-columnar Parquet where each declared column is its own Arrow column. A parquet source reads one file (path) or a whole directory (prefix).

For Parquet on object stores or HDFS instead of the local disk, see S3, GCS, Azure and WebHDFS, which ride the same Parquet seams over a remote filesystem.

Dependency and version¶

Component	Provenance	Version
Engine core (`clink::core`)	Built in	n/a
Apache Arrow / Parquet	From-source toolchain (the `parquet` connector only)	`24.0.0`

No client library and no CLINK_WITH_* knob: the file connector needs nothing beyond the standard library, and the parquet connector is built whenever Arrow is (CLINK_BUILD_ARROW, on by default).

Factories¶

Factory name	Direction	Channel	Notes
`file_text_source`	source	string	one line per record
`file_json_source`	source	row	one JSON object per line
`file_text_sink`	sink	string	one line per record
`file_json_sink`	sink	row	one JSON object per line
`file_json_upsert_sink`	sink	row	nets a changelog by primary key
`partition_file_sink`	sink	row	one file per `partition_by` value (append-only)
`file_2pc_sink_row` / `file_2pc_sink_string`	sink	row / string	exactly-once (stage then atomic rename)
`parquet_string_source` / `parquet_int64_source`	source	string / int64	`path` (one file, or a directory of parts) or `prefix` (a directory tree)
`parquet_row_source`	source	row	typed-columnar (one Arrow column per declared column); `path` may be a directory of parts
`parquet_string_sink` / `parquet_int64_sink`	sink	string / int64	a directory of part files, one per subtask per checkpoint interval
`parquet_row_sink`	sink	row	typed-columnar, the same directory of part files
`parquet_row_2pc_sink`	sink	row	exactly-once typed-columnar

In clink --capabilities these appear under four connector identities: file, file_2pc, parquet and parquet_2pc (the 2PC records carry the exactly-once declaration; the plain ones are at-least-once).

Configuration¶

Options are read from BuildContext params in src/cluster/built_in_factories.cpp (the file/parquet source/sink factories) and src/sql/install.cpp (the row-channel factories). On the SQL path they are the WITH (...) properties.

Option	Applies to	Required	Default	Description
`path`	both	one of `path`/`prefix`	(none)	For `file`, the file path; a parallel sink appends `.<subtask_idx>` so each subtask writes a distinct file. For a `parquet` sink, the directory its part files are written to. For a `parquet` source, one file, or a directory whose `*.parquet` parts are all read (what a `parquet` sink wrote).
`prefix`	`parquet` source	one of `path`/`prefix`	(none)	A directory to read recursively; every matching Parquet file is read, sharded round-robin across subtasks (file `i` is read by subtask `i % parallelism`).
`recursive`	`parquet` source (`prefix`)	No	`true`	Descend into sub-directories when listing.
`suffix`	`parquet` source (`prefix`)	No	`.parquet`	Only files whose name ends with this are read.
`format`	`file`	No	line text	`format='json'` selects the multi-column row channel (one JSON object per line).
`partition_by`	`file` sink	No	unset	Route rows to one file per distinct value of the named column(s); append-only.
`mode`	`file` sink	No	append	`mode='upsert'` nets a changelog by the table's PRIMARY KEY.
`delivery_guarantee`	`file` / `parquet` sink	No	at-least-once	`delivery_guarantee='exactly_once'` selects the 2PC sink.
`json_non_finite`	`file` JSON sink	No	`null`	How a NaN or infinite double is written: `'null'`, or `'string'` for `"NaN"`, `"Infinity"` and `"-Infinity"`. Any other value is refused. Applies to every `format='json'` file sink (plain, upsert, exactly-once, `partition_by`); a `parquet` sink refuses it.

The file JSON row source decodes each line against the table's declared columns, as the Kafka JSON bridge does: a DECIMAL value is carried exactly at the column's scale, a REAL value is rounded to float precision, an integer token in a DOUBLE column becomes a double, so x / 2 over {"x":9} is 4.5, and a whole numeral such as 7.0 in a BIGINT or INTEGER column becomes an integer, so n / 2 over {"n":7.0} is 3, as the column's type says. A directory path decodes the same way.

JSON has no token for NaN or an infinity, so the file JSON sinks write a non-finite double as null by default, or as "NaN", "Infinity" and "-Infinity" with json_non_finite='string'. A partition_by sink names the file of a NaN or infinite partition value NaN, Infinity or -Infinity in either case. Read back by the file source, null is SQL NULL, and the strings stay strings: they are not turned back into doubles. See NaN and infinity in JSON output. A parquet sink stores the value itself.

The parquet row source derives its Arrow schema from the table's declared columns (schema_columns), so no separate schema option is needed. It resolves those columns by name, with the same type, in any file: one clink wrote, or one another tool wrote (pyarrow, DuckDB, Spark, ClickHouse). A file from another tool carries none of clink's own columns, and its rows read with no event time; a declared column the file lacks, or holds as another type, is refused at open, naming it.

SQL usage¶

connector='file' (or filesystem) and connector='parquet' are both first-class SQL source and sink connectors. The channel follows the table shape: a single TEXT column uses the string channel, format='json' uses the row channel.

-- Read a directory of Parquet files, aggregate, write JSON lines.
CREATE TABLE events (user_id BIGINT, amount BIGINT) WITH (
  connector = 'parquet',
  prefix    = '/data/events'          -- every *.parquet under here, sharded across subtasks
);

CREATE TABLE totals (user_id BIGINT, total BIGINT) WITH (
  connector = 'file',
  format    = 'json',
  path      = '/out/totals.ndjson'
);

INSERT INTO totals SELECT user_id, SUM(amount) AS total FROM events GROUP BY user_id;

Writing Parquet produces a directory of part files, which a parquet table reads back whole:

CREATE TABLE pq_events (user_id BIGINT, amount BIGINT) WITH (
  connector = 'parquet', path = '/out/events');  -- /out/events/sub0-...-000000.parquet, ...

INSERT INTO pq_events SELECT user_id, amount FROM events;

Exactly-once to local Parquet, partitioned text output, and upsert:

-- exactly-once typed-columnar Parquet (staging/ then atomic rename on checkpoint).
CREATE TABLE pq_out (id BIGINT, px DOUBLE) WITH (
  connector = 'parquet', path = '/out/pq', delivery_guarantee = 'exactly_once');

-- one file per 'region' value (append-only).
CREATE TABLE by_region (region STRING, n BIGINT) WITH (
  connector = 'file', format = 'json', path = '/out/regions', partition_by = 'region');

Example¶

Programmatic local Parquet round-trip with the single-file ParquetSink<T>, plus a multi-object directory read. The parquet connector's factories use ParquetRollingSink<T> (clink/connectors/parquet_rolling_sink.hpp) instead, so that a job's output survives a restore.

#include "clink/connectors/parquet_sink.hpp"
#include "clink/connectors/parquet_source.hpp"
#include "clink/connectors/multi_object_parquet_source.hpp"
#include "clink/core/arrow_batcher.hpp"
#include <arrow/filesystem/localfs.h>

using namespace clink;

// Write one file.
ParquetSink<std::int64_t> sink("/data/part-0.parquet", int64_arrow_batcher());
sink.open();
Batch<std::int64_t> b;
b.emplace(1);
b.emplace(2);
sink.on_data(b);
sink.close();

// Read a whole directory, sharded for subtask 0 of 4.
MultiObjectParquetSource<std::int64_t>::Options o;
o.prefix = "/data";       // every *.parquet under /data
o.subtask_idx = 0;
o.parallelism = 4;
MultiObjectParquetSource<std::int64_t> src(
    []() -> std::shared_ptr<arrow::fs::FileSystem> {
        return std::make_shared<arrow::fs::LocalFileSystem>();
    },
    o, int64_arrow_batcher());
src.open();
// drive src.produce(emitter) until it returns false; src.close();

Delivery semantics¶

file text/JSON sink (file_text_sink / file_json_sink): at-least-once. A run that starts from empty state truncates the file; a run restored from a checkpoint appends to it, so the rows before the restore point stay and the rows after it are written again as the sources replay them. Every checkpoint barrier hands the lines before it to the operating system before the checkpoint is acknowledged, which a process failure does not undo (a host failure that loses the page cache is not covered), and a restored run first cuts a partial last line that a kill left behind, so its first line is not glued to the fragment. A write the stream refuses, such as one to a full disk, fails the sink at once. Before these, lines waited in the stream's buffer until it filled or the input ended, so a checkpoint could complete over lines a killed process then lost; a write error went unnoticed; and a restored run appended straight after a torn line. The built-in file_line_sink and file_int64_sink behave the same way; they used to truncate their file on every open, a restored one included, which erased every line before the restore point.
file 2PC sink (delivery_guarantee='exactly_once', file_2pc_sink_*): exactly-once. Each checkpoint interval is staged then atomically renamed into committed/ on the global commit, with write_fsync_rename durability (toggle with CLINK_STATE_FSYNC). This is the local reference implementation the object-store 2PC sinks mirror. Each part is named sub<N>-<run>-<ckpt>.dat, where <run> is unique to each run of the sink, so a rerun into the same directory adds its files beside the earlier run's and never replaces one. Parts used to be named after the checkpoint alone, and a run that reused an earlier run's checkpoint id renamed its file over that run's: every run that commits only at the end of the input does, since that commit always takes the same id, as can another job writing to the directory.
parquet sink (parquet_*_sink, parquet_row_sink): at-least-once, a directory of complete part files (ParquetRollingSink). Each subtask closes a part at every checkpoint barrier and at the end of input, so its output is readable while the job runs and everything up to the last barrier survives a kill. A part is written to <name>.inprogress, fsynced with its directory (unless CLINK_STATE_FSYNC=0) and renamed when complete, so a reader never sees a torn .parquet file and the part is on stable storage before the checkpoint that relies on it completes. A run restored from a checkpoint keeps every part already written, including those past the restore point, whose rows the sources then replay; a run that starts from empty state replaces the previous run's parts (only files named sub<N>-*.parquet). Read the directory with a parquet source (path or prefix), which resolves columns by name as for one file, pyarrow.dataset, or DuckDB's read_parquet('<dir>/*.parquet'). A path that still holds the single file an earlier release wrote is refused at open, naming the file.
parquet 2PC row sink (delivery_guarantee='exactly_once', parquet_row_2pc_sink): exactly-once, the same staging then atomic-rename protocol with complete, externally-readable Parquet files, named sub<N>-<run>-<ckpt>.parquet so that, as for the file 2PC sink, a rerun into the same directory adds its files beside the earlier run's.
Sources (file_*_source, parquet_*_source): bounded, and replay from a checkpoint by re-opening and skipping the record/batch count already emitted (deterministic for a fixed file or a fixed directory listing).

Limitations¶

The file text sink, the parquet rolling sink and both 2PC sinks must each be the only sink on their chain: at every checkpoint barrier the first two write out what they received before the barrier and the 2PC sinks prepare the interval, which holds only while it owns its chain's checkpoint, so Dag::add_sink refuses another sink beside it, whichever is added first. SQL and job graphs plan every sink as a subtask of its own; with the Dag API, put the other sink in a separate Dag (see checkpointing).
The parquet source's multi-object directory read (prefix) shards by object across same-parallelism subtasks; cross-parallelism rescale of the assignment is not coordinated.
No Hive-partition pruning or column projection on the Parquet read path; the whole file is read.
The file text sink truncates on open unless the run restores from a checkpoint, and the partitioning sink is append-only.
The programmatic single-file ParquetSink<T> (see the example) keeps no checkpoint state and writes the file's footer only at close. A process killed before then leaves a file without one, which readers reject; a cancelled task closes the file with the rows it held, so it reads as complete; and a run restored from a checkpoint truncates the file and writes only the rows after the restore point, so those before it are lost. Use it where a failure re-reads the input from the start; the parquet factory sink keeps its output across a restore.
The typed-columnar parquet_row_* factories carry one Arrow column per declared column; the parquet_string_source/_sink factories carry a single column, so a multi-object prefix read on the string channel expects single-column files.