zeroshade commented on code in PR #900:
URL: https://github.com/apache/iceberg-go/pull/900#discussion_r3095067743
##########
table/rolling_data_writer.go:
##########
@@ -327,50 +345,120 @@ func (r *RollingDataWriter) stream(outputDataFilesCh
chan<- iceberg.DataFile) {
return nil
}
- for record := range r.recordCh {
+ writeRecord := func(record arrow.RecordBatch) error {
if currentWriter == nil {
fileCount := int(r.fileCount.Add(1))
var err error
currentWriter, err = r.factory.openFileWriter(
r.ctx, r.partitionKey, r.partitionValues,
r.partitionID, fileCount)
if err != nil {
- record.Release()
- r.sendError(err)
-
- return
+ return err
}
}
converted, err := ToRequestedSchema(r.ctx, r.factory.fileSchema,
r.factory.taskSchema, record,
SchemaOptions{IncludeFieldIDs: true, UseWriteDefault: true})
if err != nil {
- record.Release()
- r.sendError(err)
-
- return
+ return err
}
err = currentWriter.Write(converted)
converted.Release()
- record.Release()
if err != nil {
- r.sendError(err)
-
- return
+ return err
}
+ r.lastWriteTime.Store(time.Now().UnixNano())
+
if currentWriter.BytesWritten() >= r.factory.targetFileSize {
+ return closeWriter()
+ }
+
+ return nil
+ }
+
+ // Both the idle-timeout and non-idle-timeout paths use a select loop
+ // so the goroutine can be stopped via context cancellation (used by
+ // the reaper and closeAndWait) without closing recordCh.
+ idleTimeout := r.factory.idleTimeout
+
+ var idleTimerC <-chan time.Time
+ var idleTimer *time.Timer
+ if idleTimeout > 0 {
+ idleTimer = time.NewTimer(idleTimeout)
+ idleTimer.Stop()
+ defer idleTimer.Stop()
+ idleTimerC = idleTimer.C
+ }
+
+ for {
+ select {
+ case record, ok := <-r.recordCh:
+ if !ok {
+ // Channel closed; finalize any open file.
+ if err := closeWriter(); err != nil {
+ r.sendError(err)
+ }
+
+ return
+ }
+
+ err := writeRecord(record)
+ record.Release()
+ if err != nil {
+ r.sendError(err)
+
+ return
+ }
+
+ // Reset idle timer only when a file is open.
+ if idleTimer != nil && currentWriter != nil {
+ if !idleTimer.Stop() {
+ select {
+ case <-idleTimerC:
+ default:
+ }
Review Comment:
add a comment here about how Stop returns false if the timer already expired
etc. and how this is consuming the message on the channel. Hence why we don't
need to drain records on the idle timeout
##########
table/rolling_data_writer.go:
##########
@@ -383,7 +471,6 @@ func (r *RollingDataWriter) sendError(err error) {
func (r *RollingDataWriter) close() {
r.cancel()
- close(r.recordCh)
Review Comment:
Where does the `recordCh` get closed now? We don't want to leak the resource
and leave the channel open.
--
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]
