abhirathod95 opened a new pull request, #900:
URL: https://github.com/apache/iceberg-go/pull/900
WriteRecords only rolls files when they hit targetFileSize. When the input
is a long-lived channel (e.g. streaming events into time-based partitions like
hour=2024010100), low-volume partitions may never reach the target size. This
means:
- Buffered data stays invisible to readers until the file is eventually
closed
- Open file writers and goroutines accumulate for stale partitions (e.g.
last hour's partition will never receive new data but its writer lives forever)
- Buffered data is lost on crash
**table/write_records.go**
New public API options:
- WithIdleTimeout(d) — flush the current data file when no records arrive
for d. After flush, the writer stays alive and opens a new file on the next
record.
- WithReaperTimeout(d) — tear down writer goroutines that have been idle
for d, freeing resources. Independent from idle flush. Defaults to 10x idle
timeout if omitted.
Both are threaded through writeRecordConfig → recordWritingArgs.
**table/arrow_utils.go**
Added idleTimeout and reaperTimeout fields to recordWritingArgs so they
flow from the public API down to newWriterFactory.
**table/rolling_data_writer.go**
writerFactory changes:
- New fields: idleTimeout, reaperTimeout, reaperCancel, reaperDone
- newWriterFactory starts a reapIdleWriters goroutine when idleTimeout > 0
&& reaperTimeout >= 0. A negative reaper timeout disables it.
- reapIdleWriters ticks at reaperTimeout, scans all writers, and tears
down any whose lastWriteTime exceeds the threshold.
- closeAll stops the reaper and waits for it before closing writers.
RollingDataWriter changes:
- New field: lastWriteTime atomic.Int64 — updated on every record write.
- stream() rewritten from for range recordCh to an explicit select loop
with three cases:
a. recordCh — write record, reset idle timer if file is open
b. idleTimerC — flush current file (identical to size-based roll)
c. ctx.Done() — drain buffered records, flush, exit
- close() calls cancel() only (not close(recordCh)) to avoid
send-on-closed-channel panics when the reaper races with Add.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
