///|
pub(all) struct BatchProgress {
rows_accepted : Int
batches_accepted : Int
completed : Bool
}
///|
/// Visit copied batches without constructing an array of all batches. Returning
/// false declines the current batch and stops. Already accepted batches are not
/// rolled back. This bounds export buffering, not the retained dataset itself.
pub fn Table::visit_batches(
self : Table,
size : Int,
consume : (Array[Row]) -> Bool,
offset? : Int = 0,
) -> Result[BatchProgress, Issue] {
if size <= 0 {
return Err(
issue("invalid_batch_size", self.name, "Batch size must be positive"),
)
}
if offset < 0 || offset > self.rows.length() {
return Err(
issue(
"invalid_batch_offset",
self.name,
"Offset must identify a row boundary",
),
)
}
let mut position = offset
let mut batches_accepted = 0
while position < self.rows.length() {
let count = size.min(self.rows.length() - position)
let batch : Array[Row] = []
for ri in position..<(position + count) {
batch.push({ cells: self.rows[ri].cells.copy(), })
}
if !consume(batch) {
return Ok({
rows_accepted: position - offset,
batches_accepted,
completed: false,
})
}
position += count
batches_accepted += 1
}
Ok({ rows_accepted: position - offset, batches_accepted, completed: true, })
}
///|
/// Encode one NDJSON chunk per callback, supporting backpressure and resumption
/// by accepted row count. Callers are responsible for durable write semantics.
pub fn Table::visit_ndjson(
self : Table,
size : Int,
consume : (String) -> Bool,
offset? : Int = 0,
) -> Result[BatchProgress, Issue] {
self.visit_batches(
size,
rows => {
let out = StringBuilder()
for row in rows {
out.write_string(row.to_json_text())
out.write_char('\n')
}
consume(out.to_string())
},
offset~,
)
}
///|
/// A CSV batch contains a header only at offset zero. Later chunks can be
/// concatenated verbatim. Use the same projection and size when resuming.
pub fn Table::visit_csv(
self : Table,
columns : Array[String],
size : Int,
consume : (String) -> Bool,
offset? : Int = 0,
) -> Result[BatchProgress, Issue] {
let mut first = offset == 0
if self.rows.is_empty() && offset == 0 && size > 0 {
let accepted = consume(columns.map(csv_cell).join(",") + "\r\n")
return Ok({
rows_accepted: 0,
batches_accepted: if accepted {
1
} else {
0
},
completed: accepted,
})
}
self.visit_batches(
size,
rows => {
let out = StringBuilder()
if first {
out.write_string(columns.map(csv_cell).join(","))
out.write_string("\r\n")
}
for row in rows {
let cells = columns.map(name => {
match row.get(name) {
Some(value) => csv_cell(value.display())
None => ""
}
})
out.write_string(cells.join(","))
out.write_string("\r\n")
}
let accepted = consume(out.to_string())
if accepted {
first = false
}
accepted
},
offset~,
)
}