File Formats

A file format lets Beacon read and write a new type of file. The template has one example: example_tsv in native/src/tsv.rs. It reads and writes tab-separated text files with the extension .tsv.

What Beacon adds for a format

For a format with the name example_tsv, Beacon adds these SQL objects:

-- Read files.
SELECT * FROM read_example_tsv(['stations.tsv'], {null_value: '-'});

-- Show the merged schema of files.
SELECT * FROM read_example_tsv_schema(['stations.tsv']);

-- Make a table on a folder.
CREATE EXTERNAL TABLE stations STORED AS example_tsv LOCATION 'tsv/' OPTIONS ('null_value' '-');

-- Write the result of a query.
COPY (SELECT * FROM stations) TO 'result.tsv' STORED AS example_tsv;

STORED AS accepts only the format name. An extension is not a STORED AS name.

What Beacon does and what the plugin does

The plugin works on one file in each call. Beacon does the work for many files.

Beacon	Plugin
Lists the files of a path, a glob or a folder.	Reads the schema of one file.
Merges the schemas of all files and widens the types.	Opens one file and gives record batches.
Casts each column to the table type. Fills a missing column with nulls.
Adds the partition columns of PARTITIONED BY.
Applies the filters and the limit again.
For COPY, splits the output into files.	Writes one output file.

So a format can be small.

The two traits

Implement two traits from beacon_plugin_sdk::file_format.

FormatFactory

FormatFactory describes the format. Beacon reads its values one time, when it loads the plugin.

#[derive(Debug)]
pub struct TsvFormat;

impl FormatFactory for TsvFormat {
    fn name(&self) -> &str {
        "example_tsv"
    }

    fn extensions(&self) -> Vec<String> {
        vec!["tsv".to_string()]
    }

    fn output_kind(&self) -> OutputKind {
        OutputKind::Flat
    }

    fn can_write(&self) -> bool {
        true
    }

    fn table_function_args(&self) -> Vec<ArgSpec> {
        vec![ArgSpec { option_key: "null_value".to_string(), kind: ArgKind::String }]
    }

    fn create(&self, options: &HashMap<String, String>) -> Result<Arc<dyn FormatReader>> {
        let null_value = options.get("null_value").cloned().unwrap_or_default();
        Ok(Arc::new(TsvReader { null_value }))
    }
}

Method	Required	Use
name	yes	The format name: STORED AS <name>, read_<name> and read_<name>_schema.
extensions	yes	The file extensions, without the dot. Beacon lists the files of a folder by the first extension.
output_kind	yes	Flat or Nd. See Output kinds.
create	yes	Makes a FormatReader from the table options.
table_function_args	no	The fields of the options struct of read_<name>.
can_write	no	true when the format has a writer. The default is false.
split_output	no	true when Beacon can spread one output over many files. The default is true.
content_type, writes_directory, wants_local_file	no	Beacon does not use them yet.

FormatReader

FormatReader is one format with its table options. Each call is about one file.

Method	Required	Use
infer_schema(store, object)	yes	The schema of one file. Beacon calls it for each file and merges the results.
open(store, object, request)	yes	A stream of record batches for one file.
write(store, request, input)	for a writer	Writes one output file. See Write a file.
write_ordering(schema)	no	The sort order that the writer needs. Beacon sorts the input to it.
infer_stats(store, object, table_schema)	no	Statistics of one file. Beacon does not call it yet.

The trait is an async_trait. Add async-trait and futures to Cargo.toml.

Export the factories

pub fn formats() -> Vec<Arc<dyn FormatFactory>> {
    vec![Arc::new(TsvFormat)]
}

beacon_plugin_sdk::export_plugin!(scalar = functions, formats = tsv::formats, tables = memory::tables);

Give only the keys that the plugin uses, in any order. A plugin with only formats uses export_plugin!(formats = tsv::formats).

Names

• The format name is the key for STORED AS. It must be free: not a built-in format, a format of another plugin or a table type.
• read_<name> and read_<name>_schema must be free too.
• A name and an extension have lowercase letters, digits and _ only.
• An extension is not a key. Two formats can share one. So example_tsv uses tsv, which the Beacon CSV format also reads.

Start the format name with the plugin name.

Options

Options come from two places. create gets both as one map of strings.

Place	Example	Key in create
The options struct of read_<name>	read_example_tsv([...], {null_value: '-'})	null_value
OPTIONS of CREATE EXTERNAL TABLE or COPY	OPTIONS ('null_value' '-') or OPTIONS ('format.null_value' '-')	null_value

• table_function_args lists the fields that read_<name> accepts. Beacon refuses another field.
• An ArgSpec has a kind: String, Bool or StringList. Beacon joins a StringList with ,.
• A NULL field gives no option.
• In OPTIONS, Beacon removes the prefix format.. It drops other keys with a ., such as execution.*.
• The key single_file_output is for Beacon. The plugin does not get it. See Write a file.
• Give a default for each option. create must accept a missing key.

Read the schema

infer_schema gives the columns and types of one file. It can read all of the file, or only a header.

The example_tsv reader reads the file and infers the type of each column:

fn infer_type<'a>(values: impl Iterator<Item = &'a str> + Clone) -> DataType {
    if values.clone().next().is_none() {
        DataType::Utf8
    } else if values.clone().all(|v| v.parse::<i64>().is_ok()) {
        DataType::Int64
    } else if values.clone().all(|v| v.parse::<f64>().is_ok()) {
        DataType::Float64
    } else {
        DataType::Utf8
    }
}

Beacon calls infer_schema for many files at the same time. Then it merges the schemas:

• Two types of one family widen, such as Int64 and Float64 to Float64.
• Two types of different families, such as a number and a string, fail by default. BEACON_TYPE_WIDENING_ON_CONFLICT and BEACON_TYPE_WIDENING_STRATEGY change this rule.
• An error names the file: cannot read the schema of <path>.

Open a file

open gets a ScanRequest:

Field	Meaning
schema	The wanted columns, in order, with the merged table types. It has no partition columns.
filters	Filters on schema. They are hints. Beacon applies them again.
limit	A row limit hint. The reader can give more rows.
batch_size	The batch size of the session.

The batches hold some or all of the wanted columns. Beacon finds a column by its name. The example_tsv reader does these steps:

1. It reads the file through the store.
2. For each wanted column, it finds the column in the file. It skips a column that the file does not have.
3. It casts the text values to the table type. A value that does not fit becomes null.
4. It sets the row count of the batch. count(*) asks for no columns, and a batch without columns needs a row count.
5. It splits the rows into batches of batch_size.

let options = RecordBatchOptions::new().with_row_count(Some(table.rows.len()));
let batch = RecordBatch::try_new_with_options(Arc::clone(&schema), columns, &options)?;

To skip work, read only the wanted columns, and use the filters to skip parts of a file.

Write a file

COPY (query) TO 'path' STORED AS <name> writes through the format. Return true from can_write and implement write.

Beacon calls write one time for each output file. write gets a WriteRequest and the rows of that file as a stream:

Field	Meaning
path	The output path in the store.
schema	The schema of the input stream.
local_path	Always None in Beacon main. Write through the store.

write returns the row count. The example_tsv writer does these steps:

1. It writes the column names as the first line.
2. For each row, it writes the values with tabs between them. A null becomes the null_value option.
3. It refuses a value with a tab or a line end, because that value breaks the layout.
4. It puts the text in the store with one put call.

async fn write(
    &self,
    store: Arc<dyn ObjectStore>,
    request: WriteRequest,
    mut input: SendableRecordBatchStream,
) -> Result<u64> {
    let mut text = /* the header line */;
    let mut rows = 0;
    while let Some(batch) = input.next().await {
        let batch = batch?;
        // Add one line for each row of `batch`.
        rows += batch.num_rows() as u64;
    }
    store.put(&request.path, text.into_bytes().into()).await.map_err(external)?;
    Ok(rows)
}

Rules for a writer:

• Write through the given store. Beacon gives local_path = None.
• Use a multipart upload for a large file. One put keeps the whole file in memory.
• A query with no rows also calls write one time. Write a valid empty file.
• Return an error for a value that the format cannot hold. The COPY then fails.
• Beacon accepts only COPY and append writes. INSERT OVERWRITE fails.

How Beacon splits the output:

Setting	Effect
split_output() == false	One output gives one file.
OPTIONS ('single_file_output' 'true')	One file at path.
OPTIONS ('single_file_output' 'false')	A folder at path, with one or more files.
PARTITIONED BY (col)	One folder for each value of col.

write_ordering asks Beacon to sort the input. Return SortKey values with a column name, the direction and the null order. A column that is not in the input gives a plan error.

Output kinds

Kind	infer_schema gives	open gives
OutputKind::Flat	The column types.	Normal Arrow columns, one row per record.
OutputKind::Nd	The element types.	One beacon.nd column for each variable. Beacon broadcasts the arrays to rows.

Use Nd for gridded data, such as a variable over time, depth and position. beacon_plugin_sdk::nd has the types. Each column is an NdArrowArray: a flat Arrow array and named dimensions in C order. A column can use a part of the dimensions. Beacon broadcasts it to the full grid.

This sketch gives a grid v(row, col) with the coordinates row and col:

use beacon_plugin_sdk::nd::encoding::encode_nd_record_batch;
use beacon_plugin_sdk::nd::{Dimension, Dimensions, NdArrowArray, NdRecordBatch};

let dims = |spec: &[(&str, usize)]| {
    Dimensions::try_new(spec.iter().map(|(name, size)| Dimension::new(*name, *size)).collect())
};
let row = NdArrowArray::try_new(Arc::new(Int64Array::from_iter_values(0..rows as i64)), dims(&[("row", rows)])?)?;
let col = NdArrowArray::try_new(Arc::new(Int64Array::from_iter_values(0..cols as i64)), dims(&[("col", cols)])?)?;
let v = NdArrowArray::try_new(Arc::new(Float64Array::from(values)), dims(&[("row", rows), ("col", cols)])?)?;

let schema = Arc::new(Schema::new(vec![
    Field::new("row", DataType::Int64, true),
    Field::new("col", DataType::Int64, true),
    Field::new("v", DataType::Float64, true),
]));
let nd = NdRecordBatch::try_new(schema, vec![row, col, v], dims(&[("row", rows), ("col", cols)])?)?;
let batch = encode_nd_record_batch(&nd)?;

Give only the columns of request.schema. A null in the Arrow array is a missing value. Change fill values to nulls when you build the array. The Beacon fixture plugin has a full Nd example: beacon-db/beacon-core/tests/fixtures/example-plugin/native/src/formats.rs.

Rules for format code

• Do all I/O through the ObjectStore that Beacon gives. Then the format works on local disk, on S3 and on the other Beacon stores.
• Use the object_store crate that the SDK re-exports: beacon_plugin_sdk::object_store::store.
• Do not call tokio::spawn, tokio::time or Handle::current(). The plugin copy of tokio has no runtime in a format call.
• Plugin code that uses tokio::fs or tokio::net must build its own runtime.
• Keep each poll short. CPU work in a poll blocks one Beacon worker thread.
• Return errors. The SDK changes a panic in a call, a future or a stream to a query error.
• Do not panic in name, extensions or the other getters of FormatFactory.

What Beacon does not support yet

Part	Effect
infer_stats	Beacon does not prune files of a plugin format with statistics.
wants_local_file, writes_directory, content_type	Beacon ignores them. A writer always gets local_path = None.
JSON query output	The JSON query API cannot return a result in a plugin format. Use SQL COPY.
The crawler	Dataset discovery does not find files of a plugin format. Use read_<name> or CREATE EXTERNAL TABLE.

The template implements infer_stats for a later Beacon version. It gives the row count.

Test a format

The unit tests in tsv.rs call the reader and the writer with an InMemory store. They need no Beacon server. See Testing.