laskoviymishka commented on code in PR #900:
URL: https://github.com/apache/iceberg-go/pull/900#discussion_r3089634179
##########
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:
Avoids the send-on-closed panic, but the drain loop's default case means any
record that enters the buffer after the drain started is silently dropped and
never Released.
##########
table/rolling_data_writer_test.go:
##########
@@ -333,3 +344,189 @@ func (s *RollingDataWriterTestSuite)
TestBytesWrittenReflectsCompressedSize() {
s.Require().NoError(err)
s.Equal(int64(100), df.Count())
}
+
+// newTestWriterFactory creates a writerFactory with the given idle and reaper
+// timeouts for use in standalone (non-suite) tests.
+func newTestWriterFactory(t *testing.T, loc string, arrSchema *arrow.Schema,
targetFileSize int64, idleTimeout, reaperTimeout time.Duration) *writerFactory {
+ t.Helper()
+
+ icebergSchema, err := ArrowSchemaToIcebergWithFreshIDs(arrSchema, false)
+ require.NoError(t, err)
+
+ spec := iceberg.NewPartitionSpec()
+ meta, err := NewMetadata(icebergSchema, &spec, UnsortedSortOrder, loc,
iceberg.Properties{})
+ require.NoError(t, err)
+
+ metaBuilder, err := MetadataBuilderFromBase(meta, "")
+ require.NoError(t, err)
+
+ writeUUID := uuid.New()
+ args := recordWritingArgs{
+ sc: arrSchema,
+ fs: iceio.LocalFS{},
+ writeUUID: &writeUUID,
+ idleTimeout: idleTimeout,
+ reaperTimeout: reaperTimeout,
+ counter: func(yield func(int) bool) {
+ for i := 0; ; i++ {
+ if !yield(i) {
+ break
+ }
+ }
+ },
+ }
+
+ factory, err := newWriterFactory(loc, args, metaBuilder, icebergSchema,
targetFileSize)
+ require.NoError(t, err)
+
+ return factory
+}
+
+func buildTestRecord(t *testing.T, mem memory.Allocator, arrSchema
*arrow.Schema, numRows int) arrow.RecordBatch {
+ t.Helper()
+
+ bldr := array.NewRecordBuilder(mem, arrSchema)
+ defer bldr.Release()
+
+ for i := range numRows {
+ bldr.Field(0).(*array.Int32Builder).Append(int32(i))
+ bldr.Field(1).(*array.StringBuilder).Append("value")
+ }
+
+ return bldr.NewRecordBatch()
+}
+
+var testSchema = arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type: arrow.PrimitiveTypes.Int32, Nullable: true},
+ {Name: "name", Type: arrow.BinaryTypes.String, Nullable: true},
+}, nil)
+
+func TestIdleTimeoutFlushesFile(t *testing.T) {
+ mem := memory.NewGoAllocator()
+ loc := filepath.ToSlash(t.TempDir())
+ // reaperTimeout=-1 disables the reaper for this test.
+ factory := newTestWriterFactory(t, loc, testSchema, 1024*1024,
200*time.Millisecond, -1)
+ defer factory.closeAll()
+
+ outputCh := make(chan iceberg.DataFile, 10)
+ writer := factory.newRollingDataWriter(context.Background(), nil, "",
nil, outputCh)
+
+ record := buildTestRecord(t, mem, testSchema, 5)
+ require.NoError(t, writer.Add(record))
+ record.Release()
+
+ select {
+ case df := <-outputCh:
+ assert.Equal(t, int64(5), df.Count())
+ case <-time.After(5 * time.Second):
+ t.Fatal("timed out waiting for idle flush")
+ }
+}
+
+func TestIdleTimerResetsOnActivity(t *testing.T) {
+ mem := memory.NewGoAllocator()
+ loc := filepath.ToSlash(t.TempDir())
+ factory := newTestWriterFactory(t, loc, testSchema, 1024*1024,
300*time.Millisecond, -1)
+ defer factory.closeAll()
+
+ outputCh := make(chan iceberg.DataFile, 10)
+ writer := factory.newRollingDataWriter(context.Background(), nil, "",
nil, outputCh)
+
+ // Send records at intervals shorter than the idle timeout so
+ // the timer keeps resetting and does not flush prematurely.
+ for range 5 {
+ record := buildTestRecord(t, mem, testSchema, 3)
+ require.NoError(t, writer.Add(record))
+ record.Release()
+ time.Sleep(100 * time.Millisecond)
+ }
+
+ // No file should have been flushed yet (all 15 rows in one file).
+ select {
+ case <-outputCh:
+ t.Fatal("idle flush should not have fired while records were
being added")
+ default:
+ }
+
+ // Now wait for the idle timeout to fire.
+ select {
+ case df := <-outputCh:
+ assert.Equal(t, int64(15), df.Count())
+ case <-time.After(5 * time.Second):
+ t.Fatal("timed out waiting for idle flush")
+ }
+}
+
+func TestWriterReopensAfterIdleFlush(t *testing.T) {
+ mem := memory.NewGoAllocator()
+ loc := filepath.ToSlash(t.TempDir())
+ factory := newTestWriterFactory(t, loc, testSchema, 1024*1024,
200*time.Millisecond, -1)
+ defer factory.closeAll()
+
+ outputCh := make(chan iceberg.DataFile, 10)
+ writer := factory.newRollingDataWriter(context.Background(), nil, "",
nil, outputCh)
+
+ // First batch: write and let idle flush close the file.
+ record1 := buildTestRecord(t, mem, testSchema, 5)
+ require.NoError(t, writer.Add(record1))
+ record1.Release()
+
+ select {
+ case df := <-outputCh:
+ assert.Equal(t, int64(5), df.Count())
+ case <-time.After(5 * time.Second):
+ t.Fatal("timed out waiting for first idle flush")
+ }
+
+ // Second batch: same writer reopens a new file automatically.
+ record2 := buildTestRecord(t, mem, testSchema, 7)
+ require.NoError(t, writer.Add(record2))
+ record2.Release()
+
+ require.NoError(t, writer.closeAndWait())
+ close(outputCh)
+
+ var dataFiles []iceberg.DataFile
+ for df := range outputCh {
+ dataFiles = append(dataFiles, df)
+ }
+
+ require.Len(t, dataFiles, 1)
+ assert.Equal(t, int64(7), dataFiles[0].Count())
+}
+
+func TestReaperCleansUpIdleWriters(t *testing.T) {
+ mem := memory.NewGoAllocator()
+ loc := filepath.ToSlash(t.TempDir())
+ factory := newTestWriterFactory(t, loc, testSchema, 1024*1024,
200*time.Millisecond, 500*time.Millisecond)
+ defer factory.closeAll()
+
+ ctx := context.Background()
+ outputCh := make(chan iceberg.DataFile, 10)
+
+ writer, err := factory.getOrCreateRollingDataWriter(ctx, nil, "part=a",
nil, outputCh)
Review Comment:
reapIdleWriters calls closeAndWait() without holding w.mu. Meanwhile,
getOrCreateRollingDataWriter holds w.mu, loads the writer from the map, and
returns it.
Between cancel() and writers.Delete() inside closeAndWait(), the main
pipeline can get the cancelled writer:
```
Reaper: cancel() → writers.Delete() → wg.Wait()
↑ window
Main: mu.Lock() → writers.Load() → returns cancelled writer → Add() →
ctx.Done() → error
````
Add() returns context.Canceled. The main write pipeline fails with an
unexpected error. Data for a "just-reaped" partition breaks the whole pipeline
instead of creating a new writer.
##########
table/rolling_data_writer.go:
##########
@@ -404,8 +491,51 @@ func (r *RollingDataWriter) closeAndWait() error {
}
}
+// reapIdleWriters periodically scans all writers and tears down any whose
+// goroutine has been idle (no record writes) for longer than reaperTimeout.
+// This prevents long-lived partitions (e.g. hour=2024010100) from
+// accumulating goroutines that will never receive data again.
+func (w *writerFactory) reapIdleWriters(ctx context.Context) {
+ defer close(w.reaperDone)
+ ticker := time.NewTicker(w.reaperTimeout)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-ticker.C:
+ now := time.Now().UnixNano()
+ threshold := w.reaperTimeout.Nanoseconds()
+
+ w.writers.Range(func(key, value any) bool {
+ writer, ok := value.(*RollingDataWriter)
+ if !ok {
+ return true
+ }
+
+ lastWrite := writer.lastWriteTime.Load()
+ if lastWrite == 0 || (now-lastWrite) <
threshold {
+ return true
+ }
+
+ // closeAndWait cancels the goroutine, waits
for it to
+ // drain, and removes the writer from the map.
+ _ = writer.closeAndWait()
Review Comment:
Error silently discarded. Final flush failure means data loss with no
signal. At minimum log it; ideally send to an error channel the caller can
inspect.
--
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]
