# iceberg.read

> A module of iceberg's API.

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](https://github.com/magmalake/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`

```mojo
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__`

```mojo
def __init__(out self, var name: String, field_id: Int, kind: UInt8, precision: Int, scale: Int, length: Int, var array: ArrayData)
```

```mojo
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`

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

#### `array`

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

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

#### `is_nested`

```mojo
def is_nested(self) -> Bool
```

#### `num_rows`

```mojo
def num_rows(self) -> Int
```

#### `take_tree`

```mojo
def take_tree(deinit self) -> ColumnTree
```

### `ScanResult`

```mojo
struct ScanResult
```

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

Implements: `Copyable`, `Defaultable`, `Movable`

Fields:

- `columns: List[ScanColumn]`

#### `__init__`

```mojo
def __init__(out self)
```

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

#### `num_columns`

```mojo
def num_columns(self) -> Int
```

#### `num_rows`

```mojo
def num_rows(self) -> Int
```

#### `name`

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

#### `value`

```mojo
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`

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

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

#### `column`

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

#### `append`

```mojo
def append(mut self, other: Self)
```

Concatenate another result with the same columns.

#### `to_csv`

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

#### `to_json`

```mojo
def to_json(self) -> String
```

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

#### `to_batch`

```mojo
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`

```mojo
def take_batch(deinit self) -> RecordBatch
```

`to_batch`, consuming the result so no buffer is copied.

### `NameMapping`

```mojo
struct NameMapping
```

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

Implements: `Copyable`, `Defaultable`, `Movable`

Fields:

- `names: List[String]`
- `ids: List[Int]`

#### `__init__`

```mojo
def __init__(out self)
```

#### `parse`

```mojo
def parse(text: String) -> Self
```

#### `id_for`

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

#### `name_for`

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

### `ScanOptions`

```mojo
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__`

```mojo
def __init__(out self)
```

### `OpenDataFile`

```mojo
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__`

```mojo
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`

```mojo
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`

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

Writes the fetched `needed()` ranges into the reader.

#### `decode`

```mojo
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`

```mojo
def is_metadata_column(name: String) -> Bool
```

### `arrow_type_of`

```mojo
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`

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

Every column the expression touches, by id.

### `eval_leaf`

```mojo
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`

```mojo
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`

```mojo
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`

```mojo
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`

```mojo
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`

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

A result with the right columns and no rows.

## Aliases

### `FOOTER_PREFETCH`

```mojo
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.

### `POS_DELETE_FILE_PATH_ID`

```mojo
comptime POS_DELETE_FILE_PATH_ID = 2147483546
```

### `POS_DELETE_POS_ID`

```mojo
comptime POS_DELETE_POS_ID = 2147483545
```

### `META_FILE`

```mojo
comptime META_FILE = String("_file")
```

### `META_POS`

```mojo
comptime META_POS = String("_pos")
```

### `META_SPEC_ID`

```mojo
comptime META_SPEC_ID = String("_spec_id")
```

### `META_PARTITION`

```mojo
comptime META_PARTITION = String("_partition")
```

### `META_ROW_ID`

```mojo
comptime META_ROW_ID = String("_row_id")
```

### `META_LAST_UPDATED`

```mojo
comptime META_LAST_UPDATED = String("_last_updated_sequence_number")
```

### `NAME_MAPPING_PROPERTY`

```mojo
comptime NAME_MAPPING_PROPERTY = String("schema.name-mapping.default")
```

### `SRC_FILE`

```mojo
comptime SRC_FILE = 0
```

### `SRC_CONSTANT`

```mojo
comptime SRC_CONSTANT = 1
```

### `SRC_POS`

```mojo
comptime SRC_POS = 2
```

### `SRC_ROW_ID`

```mojo
comptime SRC_ROW_ID = 3
```