laskoviymishka commented on code in PR #735:
URL: https://github.com/apache/iceberg-go/pull/735#discussion_r2884996733
##########
table/scanner_internal_test.go:
##########
@@ -231,3 +235,59 @@ func TestBuildPartitionEvaluatorWithInvalidSpecID(t
*testing.T) {
assert.ErrorIs(t, err, ErrPartitionSpecNotFound)
assert.ErrorContains(t, err, "id 999")
}
+
+// TestSynthesizeRowLineageColumns verifies that _row_id and
_last_updated_sequence_number
+// are filled from task constants when those columns are present and null.
+func TestSynthesizeRowLineageColumns(t *testing.T) {
+ ctx := context.Background()
+ firstRowID := int64(1000)
+ dataSeqNum := int64(5)
+ task := FileScanTask{FirstRowID: &firstRowID, DataSequenceNumber:
&dataSeqNum}
+ rowOffset := int64(0)
+
+ // Build a batch with a data column plus _row_id and
_last_updated_sequence_number (all nulls).
+ schema := arrow.NewSchema(
+ []arrow.Field{
+ {Name: "x", Type: arrow.PrimitiveTypes.Int64, Nullable:
true},
+ {Name: iceberg.RowIDColumnName, Type:
arrow.PrimitiveTypes.Int64, Nullable: true},
+ {Name: iceberg.LastUpdatedSequenceNumberColumnName,
Type: arrow.PrimitiveTypes.Int64, Nullable: true},
+ },
+ nil,
+ )
+ const nrows = 3
+ xBldr := array.NewInt64Builder(compute.GetAllocator(ctx))
+ defer xBldr.Release()
+ xBldr.AppendValues([]int64{1, 2, 3}, nil)
+ rowIDBldr := array.NewInt64Builder(compute.GetAllocator(ctx))
+ defer rowIDBldr.Release()
+ rowIDBldr.AppendNulls(nrows)
+ seqBldr := array.NewInt64Builder(compute.GetAllocator(ctx))
+ defer seqBldr.Release()
+ seqBldr.AppendNulls(nrows)
+
+ batch := array.NewRecordBatch(schema, []arrow.Array{
Review Comment:
Inline NewArray() calls go directly into NewRecordBatch with no way to
release them afterward — same leak pattern. Fix by assigning to locals first
and releasing after batch construction. Then add memory.NewCheckedAllocator +
defer mem.AssertSize(t, 0) to make the class of leak self-enforcing going
forward — same pattern used in #762 to catch exactly this.
##########
table/arrow_scanner.go:
##########
@@ -390,6 +390,86 @@ func (as *arrowScan) getRecordFilter(ctx context.Context,
fileSchema *iceberg.Sc
return nil, false, nil
}
+// synthesizeRowLineageColumns fills _row_id and _last_updated_sequence_number
from task constants
+// when those columns are present in the batch (e.g. from ToRequestedSchema).
Per the Iceberg v3
+// row lineage spec: if the value is null in the file, it is inherited
(synthesized) from the file's
+// first_row_id and data_sequence_number; otherwise the value from the file is
kept.
+// rowOffset is the 0-based row index within the current file and is updated
so _row_id stays
+// correct across multiple batches from the same file (first_row_id +
row_position).
+func synthesizeRowLineageColumns(
Review Comment:
Same root cause as #762 — bldr.NewArray() starts at refcount=1,
array.NewRecordBatch retains to refcount=2, but the local refs in newCols are
never released. Fix needs a release loop after the batch is created:
```
rec := array.NewRecordBatch(schema, newCols, nrows)
for _, c := range newCols {
c.Release()
}
return rec, nil
```
--
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]
