magmalake iceberg

iceberg · API

iceberg.scan

Scan planning: which data files a query must read, and which deletes apply.

TableScan walks the metadata tree exactly once per plan:

metadata -> snapshot -> manifest list -> manifests -> manifest entries

pruning at every level with the strongest evidence available at that level:

  1. Manifest list. A delete manifest is collected separately; a data manifest is dropped when ManifestEvaluator proves that no partition it summarises can match. The manifest's own partition spec is used, not the table's current one, because that is what its partition tuples were written with.
  2. Manifest entry. DELETED entries are skipped outright. A live entry is dropped when its partition tuple fails the inclusive projection, or when InclusiveMetricsEvaluator proves its column bounds cannot match.
  3. Delete association. Every surviving data file is paired with the delete files that apply to it under the spec's scope rules — the sequence-number comparisons differ between position and equality deletes, and a deletion vector supersedes any position delete file for the same data file.

Reading the data files themselves is out of scope until parquet.mojo exists; plan_files stops at the task list, which is exactly what ib_scan_plan_files_json returns, so the two can be diffed directly.

Structs

FileScanTask

struct FileScanTask

One data file to read, with the deletes that apply and what is left of the filter after partitioning has been accounted for.

Implements Copyable, Movable, Writable

Fields
data_file: DataFile
delete_files: List[DataFile]
residual: String: The residual filter, in the DSL. ["true"] when nothing is left.
start: Int64
length: Int64
spec_id: Int
data_sequence_number: Int64: The data file's sequence number after inheritance — what the delete scope rules compare against, and what _last_updated_sequence_number reports for rows the file has not had rewritten.

write_to

def write_to(self, mut writer: T)

TableScan

struct TableScan

A configured, not-yet-executed scan. Every setter returns a new scan.

Implements Copyable, Movable

Fields
metadata: TableMetadata
io: FileIO
split_size: Int: Target bytes per task, or 0 for one task per data file.
snapshot_id: Int64
has_snapshot: Bool
filter_dsl: String
selected: List[String]
case_sensitive: Bool

__init__

def __init__(out self, var metadata: TableMetadata, var io: FileIO)

of

def of(var metadata: TableMetadata) -> Self

with_split_size

def with_split_size(self, bytes: Int) -> Self

Divide data files into tasks of about bytes each; 0 disables.

Uses the split_offsets a writer recorded, so no file is opened to decide where the boundaries are. Set this before plan_files() if you are handing the plan out — a worker reproducing one task must split the same way the planner did, and the split is a property of the scan rather than of the read.

use_snapshot

def use_snapshot(self, id: Int64) -> Self

use_ref

def use_ref(self, name: String) -> Self

as_of

def as_of(self, timestamp_ms: Int64) -> Self

filter

def filter(self, dsl: String) -> Self

select

def select(self, var columns: List[String]) -> Self

with_io

def with_io(self, var io: FileIO) -> Self

case_insensitive

def case_insensitive(self) -> Self

has_any_snapshot

def has_any_snapshot(self) -> Bool

False for a table that has been created but never written to.

Such a table is not an error to scan — it has a schema and no rows — so every path below falls back to the table's current schema rather than demanding a snapshot that does not exist.

snapshot

def snapshot(self) -> Snapshot

current_schema

def current_schema(self) -> Schema

The schema rows come back in.

A scan of the table as it is uses the table's current schema, so a column added since the last commit is there, null for every row. A scan of a chosen snapshot (use_snapshot, use_ref, as_of) uses the schema that snapshot was written with. Java and PyIceberg draw the same line.

schema

def schema(self) -> Schema

projected_field_ids

def projected_field_ids(self) -> List[Int]

plan_files

def plan_files(self) -> List[FileScanTask]

The scan's file tasks, planned with a cache of its own.

The cache lives exactly as long as the call: a snapshot's manifests were written together and share their schemas, so this is where the reuse is. A caller planning several snapshots — time travel, a diff between two refs — should hold one cache across them and use plan_files_with.

plan_files_with

def plan_files_with(self, mut cache: ManifestCache) -> List[FileScanTask]

plan_files, reusing (and adding to) a cache the caller owns.

name_mapping

def name_mapping(self) -> NameMapping

schema.name-mapping.default, or an empty mapping.

to_table

def to_table(self, options: ScanOptions = ScanOptions()) -> ScanResult

Read every planned file and return the rows, projected and filtered.

to_batches

def to_batches(self, options: ScanOptions = ScanOptions()) -> List[RecordBatch]

The same rows, as Arrow RecordBatches straight off the kernels.

One batch per Parquet batch read, not one per data file: nothing is concatenated on the way out, which is what makes this the fast path.

to_batches_for_paths

def to_batches_for_paths(self, paths: List[String], options: ScanOptions = ScanOptions()) -> List[RecordBatch]

The rows of just these data files, as this scan would return them.

to_batches reads the whole plan. This reads the part of it named by paths, which is what lets the plan be split across processes: one planner hands out plan_files(), and each worker asks only for the files it was given.

The filter is on the planned tasks, not on the table, so everything the plan decided still applies to the files that survive it — partition pruning, the residual, and the delete files attached to each task. A worker therefore returns exactly the rows the whole-table scan would have returned for those files, which is what makes the union of workers equal the whole.

Paths that the plan does not contain are ignored rather than raising: a snapshot can be replaced between planning and reading, and a worker holding a stale ticket should come back empty rather than fail the query. A caller that needs to know can compare lengths.

to_batches_for_splits

def to_batches_for_splits(self, paths: List[String], starts: List[Int64], options: ScanOptions = ScanOptions()) -> List[RecordBatch]

The rows of specific splits, named by (path, start) pairs.

to_batches_for_paths reads whole files, which is the right unit when a file is a task. Once with_split_size divides files, a worker holds something finer: one byte range of one file. Matching on the pair is what distinguishes two tasks over the same path.

The scan must be configured the way the planner was — same snapshot, same filter, same split_size — because this re-plans and then selects. A plan that divided differently has no task at the requested offset and the split comes back empty, which is the same rule to_batches_for_paths follows for an unknown path: a worker holding a stale ticket returns nothing rather than failing the query.

to_batches_for_tasks

def to_batches_for_tasks(self, var tasks: List[FileScanTask], options: ScanOptions = ScanOptions()) -> List[RecordBatch]

The rows of tasks that were already planned, without planning again.

to_batches_for_paths and to_batches_for_splits re-plan and then select, which reads the manifest list and every manifest the filter keeps. That's cheap next to a local warehouse and expensive next to object storage, where a worker handed one file would make a request per manifest before reading it. A planner that ships the FileScanTasks themselves lets the worker go straight to the data.

The scan supplies what the tasks don't carry: the schema, the projection, the name mapping, the partition specs and case sensitivity. It should be configured as the planner's was, with the same filter and selection; the residual comes from each task.

to_batch_reader

def to_batch_reader(self, options: ScanOptions = ScanOptions()) -> BatchReader

The same rows again, one batch at a time instead of all at once.

to_batches holds the whole result before the caller sees the first batch, so a scan that returns most of a large table costs the whole table in memory. A BatchReader reads the plan in waves and hands the batches out as it goes, so a caller that folds — counts, sums, writes onward — holds one wave rather than one result. The rows and their order are the same as to_batches; see BatchReader for what a wave is and why the order survives num_workers.

Planning happens here, not on the first batch, so a bad snapshot or an unreadable manifest raises from this call rather than from the loop.

count

def count(self, options: ScanOptions = ScanOptions()) -> Int64

How many rows this scan would return, without reading the ones the manifests have already counted.

A DataFile carries record_count, so a task whose every row survives to the caller contributes its count with the data file never opened — which is the whole point: an unfiltered COUNT(*) becomes a walk of the manifests instead of a decode of the table. _countable_from_metadata decides which tasks those are, and the rest are read and their surviving rows counted, so a scan that cannot be answered from metadata is still answered, only slowly.

Adding the two halves is exact because the tasks partition the scan's rows: a row belongs to exactly one data file, and to_batches produces each task's rows independently of every other task's. The answer is therefore len(to_table()) by construction, and the tests assert exactly that on every fixture that takes the reading path.

This is not select count(x): select narrows the columns and never the rows, so a projection is ignored here. Snapshot selection is not ignored — plan_files has already resolved use_snapshot, use_ref and as_of, so time travel, a branch and a tag each count their own snapshot, and a table that has never been written to plans no tasks and counts zero.

plan_files_json

def plan_files_json(self) -> String

The plan in the same shape ib_scan_plan_files_json emits.

BatchReader

struct BatchReader

A scan being read a batch at a time, in the order to_batches returns.

The plan is consumed in waves: the next wave file scan tasks go through TableScan._read_files together, their batches are handed out in task order, and only when the last of them has been yielded is the next wave started. Nothing beyond the wave in flight is held, so a fold over a scan costs one wave rather than one result — with the default num_workers = 1 a wave is one data file.

Ordering

Rows arrive in exactly the order to_batches returns them, for every num_workers. That is the whole reason for the wave: within a wave the merge is _read_files' existing merge by task index, and waves are consumed in plan order, so thread scheduling cannot reach the output. The alternative — yielding each file as its worker finishes — would have made a scan's row order depend on which core won a race, and therefore differ between two runs of the same query on the same data. The num_workers docstring promises the opposite, and to_table/to_batches both depend on it; a streaming path that quietly broke it would be a worse API than one that buffers.

The wave is TableScan._worker_split's file-axis width, which is min(num_workers, len(tasks)), so every worker still gets a file and the parallel path is as wide as it was. What the wave costs instead is a barrier at its end: a wave is only done when its slowest file is, and a worker that finishes early idles until then, where to_batches would have moved it onto the next file. That is the price of the ordering guarantee, and it is paid in time, not in memory.

A scan with a limit streams one file at a time — the file axis is not used at all, exactly as in to_batches, because stopping early is only meaningful in order.

Errors

next_batch is the primitive and raises what a read raises. batches() exists for for batch in reader.batches():, and a for loop cannot carry an error out: Iterator.__next__ may raise only StopIteration. A read that fails inside the loop therefore ends the loop and parks the message on the reader, and the caller must call raise_if_failed afterwards or a truncated scan reads as a complete one:

var reader = table.scan().to_batch_reader(options)
var rows = 0
for batch in reader.batches():
    rows += batch.num_rows
reader.raise_if_failed()

Implements Movable

Fields
scan: TableScan
tasks: List[FileScanTask]
schema: Schema
ids: List[Int]
meta_columns: List[String]
mapping: NameMapping
options: ScanOptions
wave: Int: How many data files a wave reads before any of them is handed out.
next_task: Int: The first task of the wave that has not been read yet.
pending: List[RecordBatch]: The current wave's undelivered batches, reversed: pop() takes the next one in task order and moves it, where indexing would copy an arena.
rows: Int: Rows handed out, which is what a limit counts down.
done: Bool
failure: String: What a read raised inside batches(), where it could not escape.

__init__

def __init__(out self, scan: TableScan, options: ScanOptions = ScanOptions())

num_tasks

def num_tasks(self) -> Int

How many data files the plan has — the scan's whole work list.

next_batch

def next_batch(mut self) -> Optional[RecordBatch]

The next batch, or nothing once the scan is finished.

This is the API that reports a read failure honestly; batches() is the same walk with for-loop sugar and a deferred error.

batches

def batches(mut self) -> _BatchIter[origin_of(self)]

A for-loop view of this reader. See the note on errors above: the loop ends on a read failure and raise_if_failed is what reports it.

failed

def failed(self) -> Bool

Whether a read failed inside batches() and ended the loop early.

raise_if_failed

def raise_if_failed(self)

Re-raise what batches() could not. A no-op after a clean scan.

Aliases

comptime TRUE_RESIDUAL = String("[\22true\22]")

What Expr.text prints for a residual that has nothing left to check.

ResidualEvaluator.residual_for builds that residual as a single OP_TRUE node, and Expr.text prints exactly this for it, so the comparison is against a canonical string and not against something a user typed.