zeroshade commented on code in PR #1993: URL: https://github.com/apache/iceberg-go/pull/1993#discussion_r3960997783
########## table/write_read_tuning_test.go: ########## @@ -0,0 +1,179 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table + +import ( + "fmt" + "strings" + "testing" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/arrow-go/v18/parquet/file" + "github.com/apache/iceberg-go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func manyRowsJSON(n int) string { + var sb strings.Builder + sb.WriteString("[") + for i := range n { + if i > 0 { + sb.WriteString(",") + } + fmt.Fprintf(&sb, `{"id":%d,"data":"row-%d"}`, i, i) + } + sb.WriteString("]") + + return sb.String() +} + +func TestWithArrowBatchSizeCapsDecodedBatches(t *testing.T) { + const numRows = 100 + const batchSize = 7 + tbl := buildV3TableWithRows(t, manyRowsJSON(numRows)) + + _, records, err := tbl.Scan(WithArrowBatchSize(batchSize)).ToArrowRecords(t.Context()) + require.NoError(t, err) + + var totalRows, batches int64 + for rec, err := range records { + require.NoError(t, err) + assert.LessOrEqual(t, rec.NumRows(), int64(batchSize)) + totalRows += rec.NumRows() + batches++ + rec.Release() + } + assert.Equal(t, int64(numRows), totalRows) + assert.Greater(t, batches, int64(numRows/batchSize)) + + // Control: the default batch size returns all rows in one batch. + _, records, err = tbl.Scan().ToArrowRecords(t.Context()) + require.NoError(t, err) + batches = 0 + for rec, err := range records { + require.NoError(t, err) + batches++ + rec.Release() + } + assert.Equal(t, int64(1), batches) +} + +func TestWithArrowBatchSizeIgnoresNonPositive(t *testing.T) { + tbl := buildV3TableWithRows(t, manyRowsJSON(3)) + + scan := tbl.Scan(WithArrowBatchSize(0), WithArrowBatchSize(-5)) + assert.Empty(t, scan.options.Get(ParquetBatchSizeKey, "")) +} + +func TestWithArrowBatchSizeDoesNotMutateCallerOptions(t *testing.T) { + tbl := buildV3TableWithRows(t, manyRowsJSON(3)) + + callerOpts := iceberg.Properties{"include_empty_files": "true"} + scan := tbl.Scan(WithOptions(callerOpts), WithArrowBatchSize(9)) + assert.Equal(t, "9", scan.options.Get(ParquetBatchSizeKey, "")) + assert.Empty(t, callerOpts[ParquetBatchSizeKey]) + assert.Equal(t, "true", scan.options.Get("include_empty_files", "")) +} + +func TestRecordQueueCapacityDefaultAndOverride(t *testing.T) { + f := &writerFactory{} Review Comment: **major** — WithRecordBatchBufferSize is unpinned end-to-end; its only test is a getter tautology TestRecordQueueCapacityDefaultAndOverride constructs a bare &writerFactory{}, assigns .recordBufferSize directly, and asserts recordQueueCapacity() echoes it back. It never calls WithRecordBatchBufferSize, never populates recordWritingArgs, and never observes cap(recordCh) -- so the entire chain writeRecordConfig -> recordWritingArgs -> writerFactory -> channel capacity is untested. I verified the production wiring is in fact correct via a probe (cap 64 with default, cap 3 with override), so this is a test gap rather than a code defect, but the PR's headline knob currently has no regression protection. Suggest an end-to-end assertion on cap(w.recordCh) after newWriterFactory(recordWritingArgs{recordBatchBufferSize: N}, ...), mirroring the harness at table/partitioned_fanout_writer_test.go:175. <details><summary>Evidence</summary> ```text Mutation 1 - replaced `recordBufferSize: args.recordBatchBufferSize,` with `recordBufferSize: 0,` at table/rolling_data_writer.go:223; `go test ./table/ -count=1 -timeout=900s` => `ok github.com/apache/iceberg-go/table 8.109s`. Mutation 2 (independent) - replaced `make(chan arrow.RecordBatch, w.recordQueueCapacity())` with `make(chan arrow.RecordBatch, rollingDataWriterQueueCapacity)` at table/rolling_data_writer.go:380; `go test ./table/ -count=1 -timeout=900s` => `ok github.com/apache/iceberg-go/table 5.922s`. Probe confirming the wiring itself works: `default (0) -> cap(recordCh)=64` / `override (3) -> cap(recordCh)=3`. ``` </details> ########## table/rewrite_data_files.go: ########## @@ -378,6 +420,9 @@ func ExecuteCompactionGroup(ctx context.Context, tbl *Table, group CompactionTas if cfg.scanConcurrency > 0 { scanOpts = append(scanOpts, WithMaxConcurrency(cfg.scanConcurrency)) } + if cfg.readBatchSize > 0 { + scanOpts = append(scanOpts, WithArrowBatchSize(cfg.readBatchSize)) + } Review Comment: **major** — All three new ExecuteCompactionGroup forwardings can be deleted with the package still green, unlike the existing targetFileSize forwarding The three new `if cfg.X > 0 { ...append... }` blocks that forward readBatchSize/recordBatchBufferSize/parquetRowGroupLimit into the scan and write options are covered only by TestCompactionGroupTuningOptions (write_read_tuning_test.go:120), which applies the CompactionGroupOption closures to a bare compactionGroupConfig struct and asserts the fields were set. It never calls ExecuteCompactionGroup, so nothing detects the forwarding being dropped. This is a deviation from the project's own convention: the pre-existing WithTargetFileSize forwarding IS pinned end-to-end. Suggest one ExecuteCompactionGroup test asserting output row groups are bounded by WithCompactionParquetRowGroupLimit, which is the cheapest observable of the three. <details><summary>Evidence</summary> ```text Mutation - deleted the readBatchSize, recordBatchBufferSize and parquetRowGroupLimit append blocks from ExecuteCompactionGroup; `go test ./table/ -count=1 -timeout=900s` => `ok github.com/apache/iceberg-go/table 5.429s`. Control mutation - deleted only the PRE-EXISTING `if cfg.targetFileSize > 0 { writeOpts = append(writeOpts, WithTargetFileSize(cfg.targetFileSize)) }` block; same command => `FAIL github.com/apache/iceberg-go/table 5.355s`. Existing forwarding is pinned, new forwarding is not. ``` </details> ########## table/rewrite_data_files.go: ########## @@ -252,6 +255,45 @@ func WithCompactionScanConcurrency(n int) CompactionGroupOption { } } +// WithCompactionReadBatchSize caps the number of rows decoded per Arrow +// record batch while reading the group's tasks, forwarded to the scan +// as [WithArrowBatchSize]. Together with +// [WithCompactionRecordBatchBufferSize] it bounds the memory held by +// the compaction's read+write pipeline: buffered batches times rows per +// batch. A non-positive value keeps the table's +// read.parquet.batch-size property. +func WithCompactionReadBatchSize(n int64) CompactionGroupOption { + return func(c *compactionGroupConfig) { Review Comment: **minor** — WithCompactionReadBatchSize doc overstates the bound it provides The comment claims the two options together bound "the memory held by the compaction's read+write pipeline: buffered batches times rows per batch". Neither knob bounds the delete-side allocations: GetRecords calls readAllDeleteFiles(ctx, as.fs, tasks, as.concurrency) and readAllDeletionVectors(ctx, as.fs, tasks, as.concurrency) (table/arrow_scanner.go:2179 and :2193), which materialise positional deletes and DV bitmaps for every task in the group up front, sized by delete volume rather than by batch size or buffer depth. On a delete-heavy compaction that term can dominate. Suggest narrowing the wording to the record pipeline specifically and noting the delete-side memory is not covered. ########## table/table.go: ########## @@ -1304,6 +1305,27 @@ func WithRowLineage() ScanOption { } } +// WithArrowBatchSize caps the number of rows decoded per Arrow record +// batch when reading data files, overriding the table's +// read.parquet.batch-size property for this scan. Smaller batches bound +// the memory a scan holds per decoded batch, which matters when the +// consumer buffers batches (e.g. a compaction's read+write pipeline). +// A non-positive value is ignored. +func WithArrowBatchSize(n int64) ScanOption { + if n <= 0 { Review Comment: **minor** — WithArrowBatchSize is silently discarded when WithOptions is applied after it WithArrowBatchSize stores into scan.options, but WithOptions (table/table.go:1280) does `scan.options = maps.Clone(opts)`, replacing the map wholesale. WithArrowBatchSize is the first ScanOption to write into scan.options, so this ordering hazard is newly introduced by this PR. A caller who writes tbl.Scan(WithArrowBatchSize(n), WithOptions(userProps)) gets no cap and no error -- the memory bound silently does not apply. TestWithArrowBatchSizeDoesNotMutateCallerOptions only covers the working order (WithOptions first). Suggest either documenting the ordering requirement on WithArrowBatchSize, having WithOptions merge rather than replace, or storing the batch size in a dedicated Scan field like concurrency/limit rather than in the generic options map. ########## table/rewrite_data_files.go: ########## @@ -252,6 +255,45 @@ func WithCompactionScanConcurrency(n int) CompactionGroupOption { } } +// WithCompactionReadBatchSize caps the number of rows decoded per Arrow +// record batch while reading the group's tasks, forwarded to the scan +// as [WithArrowBatchSize]. Together with +// [WithCompactionRecordBatchBufferSize] it bounds the memory held by +// the compaction's read+write pipeline: buffered batches times rows per +// batch. A non-positive value keeps the table's +// read.parquet.batch-size property. +func WithCompactionReadBatchSize(n int64) CompactionGroupOption { + return func(c *compactionGroupConfig) { Review Comment: **nit** — Inconsistent parameter types and names across the new option family WithArrowBatchSize and WithCompactionReadBatchSize take int64 while WithParquetRowGroupLimit, WithRecordBatchBufferSize, WithCompactionRecordBatchBufferSize and WithCompactionParquetRowGroupLimit take int, with no evident reason for the split (the batch size is ultimately consumed via props.GetInt, which returns an int). Separately, the same underlying knob is named WithArrowBatchSize at the scan layer but WithCompactionReadBatchSize at the compaction layer, whereas the other two compaction options mirror their write-layer names exactly (WithCompaction + the write option name). Suggest aligning on int and on WithCompactionArrowBatchSize for symmetry. -- 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]
