magmalake iceberg

iceberg · API

iceberg.read

Reading rows: TableScan.to_table() and to_batches().

This is the half of a scan that plan_files() stops short of. For each FileScanTask it opens the data file through the FileIO, decodes it with parquet.mojo, resolves every projected column by field id, applies the deletes the planner associated, evaluates whatever is left of the filter, and hands back Arrow arrays.

Column projection follows the spec's order exactly. For each projected field id, in turn:

  1. the column with that id in the data file;
  2. otherwise, the value from the manifest's partition tuple, when an identity partition field is defined on that source column — the "metadata-only Hive migration" case;
  3. otherwise, the column the table property schema.name-mapping.default maps that id onto, for files written without field ids at all;
  4. otherwise, the field's initial-default;
  5. otherwise, null.

Schema evolution falls out of that: a renamed column is found by id under its new name, a column added after the file was written resolves to its default or null, and a promoted column (int->long, float->double, a decimal whose precision grew) is read at the file's physical width and cast to the table's current type by a columnar kernel.

Nothing is materialised per cell. The rows a scan returns are Arrow arrays throughout: iceberg.kernels casts them, builds constant arrays for the columns a file does not have, and applies deletes and the residual predicate as one List[Bool] per batch followed by a single filter pass per column. A Datum is built only when a caller asks for one — ScanResult.value, CSV and JSON output — or for the handful of predicate shapes no kernel covers, which _vector_leaf names.

Deletes are applied in the order the spec implies. A deletion vector, when one applies, replaces every position delete file for that data file; otherwise position delete files are read and their pos values collected for this file path. Equality deletes are matched last, on the columns named by equality_ids, through a hashed lookup over canonically encoded keys, with null equal to null as the spec requires. The planner has already scoped every delete by sequence number and partition.

Structs

ScanColumn

struct ScanColumn

One output column: its name, its Iceberg type, and its Arrow array.

A primitive column is a single array. A struct, list or map column is a whole tree, so the array lives in the column's own arena with root naming it and every child referring to a sibling index — the layout parquet.mojo decodes into and iceberg.nested operates on. ctype is the column's Iceberg type, which is what tells a Datum, a JSON cell or a CSV cell how to read the bytes; it is empty for a metadata column, which is always one of a handful of primitives.

Implements Copyable, Movable

Fields
name: String
field_id: Int
kind: UInt8
precision: Int
scale: Int
length: Int
ctype: ColumnType
arena: ArrayArena
root: Int

__init__

def __init__(out self, var name: String, field_id: Int, kind: UInt8, precision: Int, scale: Int, length: Int, var array: ArrayData)
def __init__(out self, var name: String, field_id: Int, kind: UInt8, precision: Int, scale: Int, length: Int, var ctype: ColumnType, var tree: ColumnTree)

empty

def empty(var name: String, field_id: Int, kind: UInt8, precision: Int, scale: Int, length: Int) -> Self

array

def array(ref self) -> ref[self.arena.nodes["element"]] ArrayData

The column's Arrow array; its children index into self.arena.

is_nested

def is_nested(self) -> Bool

num_rows

def num_rows(self) -> Int

take_tree

def take_tree(deinit self) -> ColumnTree

ScanResult

struct ScanResult

The rows a scan returned, column-major, as Arrow arrays.

Implements Copyable, Defaultable, Movable

Fields
columns: List[ScanColumn]

__init__

def __init__(out self)
def __init__(out self, var columns: List[ScanColumn])

num_columns

def num_columns(self) -> Int

num_rows

def num_rows(self) -> Int

name

def name(self, i: Int) -> String

value

def value(self, row: Int, col: Int) -> Datum

One cell, typed as the table's current schema says it is.

This is the only place a Datum is built on the read path, and it is built on demand. A Datum is a tagged scalar, so a struct, list or map cell arrives as its canonical JSON text — cell gives the same thing without the Datum around it.

cell

def cell(self, row: Int, col: Int) -> String

One cell as JSON — nested for a struct, list or map.

column

def column(ref self, i: Int) -> ref[self.columns["element"].arena.nodes["element"]] ArrayData

append

def append(mut self, other: Self)

Concatenate another result with the same columns.

to_csv

def to_csv(self, header: Bool = True) -> String

to_json

def to_json(self) -> String

One JSON object per row, in an array — the Appendix-D encoding.

to_batch

def to_batch(self) -> RecordBatch

The same rows as an Arrow RecordBatch, ready for export_c.

A nested column brings its whole subtree across, renumbered into the batch's arena, so the C Data Interface export of a nested scan is the same call as the export of a flat one.

take_batch

def take_batch(deinit self) -> RecordBatch

to_batch, consuming the result so no buffer is copied.

NameMapping

struct NameMapping

schema.name-mapping.default: fallback ids for files without them.

Implements Copyable, Defaultable, Movable

Fields
names: List[String]
ids: List[Int]

__init__

def __init__(out self)

parse

def parse(text: String) -> Self

id_for

def id_for(self, name: String) -> Int

name_for

def name_for(self, id: Int) -> String

ScanOptions

struct ScanOptions

How to read, as opposed to what.

Implements Copyable, Defaultable, Movable

Fields
batch_size: Int: Rows per output batch. The default is one batch per Parquet row group.
limit: Int: Stop after this many rows; -1 for all of them.
verify_crc: Bool
prune: Bool: Use the residual to drop row groups and pages before decoding.
lazy: Bool: Retained, and no longer read.
split_size: Int: Target bytes per scan task, or 0 to leave one task per data file.
num_workers: Int: How many OS threads a scan may use, in total.
files_at_once: Int: Data files a sequential scan reads together, 1 for one at a time.
overlap_io: Bool: With one decoding thread, fetch the next files_at_once files on a second thread while the current ones decode. On by default: the waits for object storage then overlap the decoding instead of adding to it, as pyarrow's I/O pool does beside its CPU thread.

__init__

def __init__(out self)

OpenDataFile

struct OpenDataFile

A data file between open_data_file and decode.

Implements Movable

Fields
data_file: DataFile
reader: ParquetReader[AllCodecs]
sparse: Bool: The reader holds the footer alone: needed() has to be filled.
residual: Expr
trivial: Bool
equality: List[_EqualityDeletes]
read_ids: List[Int]
n_projected: Int
plans: List[_ColumnPlan]
file_fields: List[Int]
leaves: List[_LeafPlan]
task_groups: List[Int]
deleted: List[Bool]

__init__

def __init__(out self, var data_file: DataFile, var reader: ParquetReader[AllCodecs], sparse: Bool, var residual: Expr, trivial: Bool, var equality: List[_EqualityDeletes], var read_ids: List[Int], n_projected: Int, var plans: List[_ColumnPlan], var file_fields: List[Int], var leaves: List[_LeafPlan], var task_groups: List[Int], var deleted: List[Bool])

needed

def needed(self) -> List[ByteRange]

The byte ranges to fetch and fill before decode: exactly what the selection decodes, or nothing when the whole file is here.

fill

def fill(mut self, ranges: List[ByteRange], bufs: List[List[UInt8]])

Writes the fetched needed() ranges into the reader.

decode

def decode(mut self, schema: Schema, meta_columns: List[String], options: ScanOptions) -> List[ScanResult]

The file's surviving rows, as Arrow batches.

Functions

is_metadata_column

def is_metadata_column(name: String) -> Bool

arrow_type_of

def arrow_type_of(kind: UInt8, precision: Int, scale: Int, length: Int) -> ArrowType

The Arrow type an Iceberg primitive is produced at.

collect_field_ids

def collect_field_ids(e: Expr, i: Int, mut out: List[Int])

Every column the expression touches, by id.

eval_leaf

def eval_leaf(e: Expr, i: Int, v: Datum) -> Bool

One bound leaf predicate against one value.

A predicate on a null value is false for everything except is-null and the negations that the planner's rewrite_not has already pushed down, which is Iceberg's three-valued logic collapsed the way a filter needs it.

eval_row

def eval_row(e: Expr, i: Int, row: List[Datum], ids: List[Int]) -> Bool

Evaluate a bound expression against one row.

row[k] is the value of field ids[k].

open_data_file

def open_data_file(io: FileIO, data_file: DataFile, delete_files: List[DataFile], data_sequence_number: Int64, spec: PartitionSpec, schema: Schema, projected: List[Int], meta_columns: List[String], mapping: NameMapping, residual_dsl: String, case_sensitive: Bool, options: ScanOptions, start: Int64 = Int64(0), length: Int64 = Int64(0), var tail: List[UInt8] = List()) -> OpenDataFile

One data file opened for reading: its footer parsed, the projection resolved against it, its deletes read, and the bytes the read needs named but, when sparse, not fetched yet.

read_data_file is open_data_file, then fetching needed(), then decode. A scan reading several files opens them all first, so their footers (passed in as tail, the file's last FOOTER_PREFETCH bytes) and then their chunks can each be fetched in one round of requests. The arguments are read_data_file's.

read_data_file

def read_data_file(io: FileIO, data_file: DataFile, delete_files: List[DataFile], data_sequence_number: Int64, spec: PartitionSpec, schema: Schema, projected: List[Int], meta_columns: List[String], mapping: NameMapping, residual_dsl: String, case_sensitive: Bool, options: ScanOptions, start: Int64 = Int64(0), length: Int64 = Int64(0)) -> List[ScanResult]

One data file's surviving rows, as Arrow batches.

start and length name a byte range of the file, as a split scan task carries. length == 0 means the whole file, which is what an unsplit task and every direct caller get.

projected is a list of field ids, which may name whole columns or fields nested inside them. Ids that share a top-level column come back as one column whose type has been pruned to exactly what was asked for, and the Parquet read is pruned with it: ["a.b", "c"] decodes the leaves under a.b and c, and nothing else.

read_data_file_table

def read_data_file_table(io: FileIO, data_file: DataFile, delete_files: List[DataFile], data_sequence_number: Int64, spec: PartitionSpec, schema: Schema, projected: List[Int], meta_columns: List[String], mapping: NameMapping, residual_dsl: String, case_sensitive: Bool, options: ScanOptions) -> ScanResult

read_data_file, with the batches concatenated into one result.

empty_scan_result

def empty_scan_result(schema: Schema, ids: List[Int], meta_columns: List[String]) -> ScanResult

A result with the right columns and no rows.

Aliases

comptime FOOTER_PREFETCH = 524288

Bytes read from a data file's end to get its footer in one request; the Rust and Java readers prefetch the same.

comptime POS_DELETE_FILE_PATH_ID = 2147483546
comptime POS_DELETE_POS_ID = 2147483545
comptime META_FILE = String("_file")
comptime META_POS = String("_pos")
comptime META_SPEC_ID = String("_spec_id")
comptime META_PARTITION = String("_partition")
comptime META_ROW_ID = String("_row_id")
comptime META_LAST_UPDATED = String("_last_updated_sequence_number")
comptime NAME_MAPPING_PROPERTY = String("schema.name-mapping.default")
comptime SRC_FILE = 0
comptime SRC_CONSTANT = 1
comptime SRC_POS = 2
comptime SRC_ROW_ID = 3