# iceberg.scan > A module of iceberg's API. 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` ```mojo 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` ```mojo def write_to(self, mut writer: T) ``` ### `TableScan` ```mojo 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__` ```mojo def __init__(out self, var metadata: TableMetadata, var io: FileIO) ``` #### `of` ```mojo def of(var metadata: TableMetadata) -> Self ``` #### `with_split_size` ```mojo 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` ```mojo def use_snapshot(self, id: Int64) -> Self ``` #### `use_ref` ```mojo def use_ref(self, name: String) -> Self ``` #### `as_of` ```mojo def as_of(self, timestamp_ms: Int64) -> Self ``` #### `filter` ```mojo def filter(self, dsl: String) -> Self ``` #### `select` ```mojo def select(self, var columns: List[String]) -> Self ``` #### `with_io` ```mojo def with_io(self, var io: FileIO) -> Self ``` #### `case_insensitive` ```mojo def case_insensitive(self) -> Self ``` #### `has_any_snapshot` ```mojo 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` ```mojo def snapshot(self) -> Snapshot ``` #### `current_schema` ```mojo 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` ```mojo def schema(self) -> Schema ``` #### `projected_field_ids` ```mojo def projected_field_ids(self) -> List[Int] ``` #### `plan_files` ```mojo 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` ```mojo def plan_files_with(self, mut cache: ManifestCache) -> List[FileScanTask] ``` `plan_files`, reusing (and adding to) a cache the caller owns. #### `name_mapping` ```mojo def name_mapping(self) -> NameMapping ``` `schema.name-mapping.default`, or an empty mapping. #### `to_table` ```mojo def to_table(self, options: ScanOptions = ScanOptions()) -> ScanResult ``` Read every planned file and return the rows, projected and filtered. #### `to_batches` ```mojo def to_batches(self, options: ScanOptions = ScanOptions()) -> List[RecordBatch] ``` The same rows, as Arrow `RecordBatch`es 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` ```mojo 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` ```mojo 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` ```mojo 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 `FileScanTask`s 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` ```mojo 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` ```mojo 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` ```mojo def plan_files_json(self) -> String ``` The plan in the same shape `ib_scan_plan_files_json` emits. ### `BatchReader` ```mojo 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: ```mojo 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__` ```mojo def __init__(out self, scan: TableScan, options: ScanOptions = ScanOptions()) ``` #### `num_tasks` ```mojo def num_tasks(self) -> Int ``` How many data files the plan has — the scan's whole work list. #### `next_batch` ```mojo 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` ```mojo 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` ```mojo def failed(self) -> Bool ``` Whether a read failed inside `batches()` and ended the loop early. #### `raise_if_failed` ```mojo def raise_if_failed(self) ``` Re-raise what `batches()` could not. A no-op after a clean scan. ## Aliases ### `TRUE_RESIDUAL` ```mojo 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.