laskoviymishka commented on code in PR #1889:
URL: https://github.com/apache/iceberg-go/pull/1889#discussion_r3874741425
##########
table/dv/dv_writer.go:
##########
@@ -124,6 +124,33 @@ func (w *DVWriter) Add(dataFilePath string, positions
[]int64, specID int32, par
return nil
}
+// AddPosition accumulates one position to delete for a given data file.
+// It has the same first-write partition metadata and validation semantics as
Review Comment:
the "same validation semantics as Add" line is true per call, but there's a
subtle cross-call difference I'd call out here. A multi-position `Add`
validates every position before it mutates anything, so a single negative entry
rejects the whole batch and leaves no state behind. A loop of `AddPosition`
calls isn't transactional: positions written by earlier successful calls for
the same path stay written when a later call returns an error.
Our own caller only passes one position at a time so nothing changes today,
but someone migrating a multi-element `Add` into an `AddPosition` loop would
get different error-recovery behavior. A sentence in the godoc noting that
`AddPosition` calls aren't rolled back as a group would save that surprise.
wdyt?
##########
table/dv/dv_writer.go:
##########
@@ -124,6 +124,33 @@ func (w *DVWriter) Add(dataFilePath string, positions
[]int64, specID int32, par
return nil
}
+// AddPosition accumulates one position to delete for a given data file.
+// It has the same first-write partition metadata and validation semantics as
+// Add, without requiring callers that process one deleted row at a time to
+// allocate a one-element positions slice.
+func (w *DVWriter) AddPosition(dataFilePath string, position int64, specID
int32, partitionData map[int]any) error {
Review Comment:
I'd revisit the justification before we commit to a new public method here.
The doc and the PR description frame this as avoiding the one-element `[]int64`
allocation, but the benchmark reports 0 allocs/op for both paths. `Add` never
stores the slice, so escape analysis already stack-allocates the literal at the
call site. There's no heap allocation to eliminate.
What's actually saved is the loop preamble (the len==0 guard plus the two
range passes), which the numbers put at around 1.3 ns/op. That's real but tiny
next to everything else on the DV write path (Puffin serialization, the roaring
OR, the filesystem write), so I doubt it shows up end to end.
So I'd go one of two ways. Either implement this as `return
w.Add(dataFilePath, []int64{position}, ...)` and keep it as a pure
single-position convenience, since there's no perf delta to lose. Or keep the
hand-rolled body but rewrite the doc around the real mechanism (loop overhead,
not allocation) and lean on Java parity: `BaseDVFileWriter.delete` is
per-position natively, so `AddPosition` is a more faithful peer to Java's unit
of work than the batch `Add` ever was. That's a better reason to add it than an
allocation that doesn't happen. wdyt?
##########
table/dv/dv_writer_test.go:
##########
@@ -219,6 +243,19 @@ func TestDVWriterAddRejectsNegativePositions(t *testing.T)
{
assert.Equal(t, dataFiles[0].Count(), bm.Cardinality())
}
+func TestDVWriterAddPositionRejectsNegativePosition(t *testing.T) {
+ fs := newTestFS()
+ w := NewDVWriter(fs, unpartitionedResolver())
+
+ dataPath := "s3://bucket/data/file-001.parquet"
+ err := w.AddPosition(dataPath, -1, 0, nil)
Review Comment:
this covers the negative-on-first-call case, but `Add` has a companion test
(`TestDVWriterAddRejectsNegativePositions`) that pins the more interesting
invariant: a negative position after some valid ones fails without corrupting
the state already written.
The guard does run before Set so it's correct today, but nothing pins it,
and a refactor that moved Set ahead of the guard would slip through. I'd add
the parallel: `AddPosition(path, 1)` succeeds, `AddPosition(path, -1)` errors,
then Flush and assert exactly {1}.
##########
table/dv/dv_writer_test.go:
##########
@@ -193,6 +193,30 @@ func TestDVWriterDeduplicatesPositions(t *testing.T) {
verifyDVReadBack(t, fs, dataFiles[0])
}
+func TestDVWriterAddPosition(t *testing.T) {
+ fs := newTestFS()
+ spec := iceberg.NewPartitionSpecID(0, iceberg.PartitionField{
+ SourceIDs: []int{2},
+ FieldID: 1000,
+ Name: "region",
+ Transform: iceberg.IdentityTransform{},
+ })
+ w := NewDVWriter(fs, specMapResolver(spec))
+ dataPath := "s3://bucket/region=EU/file.parquet"
+ partition := map[int]any{1000: "EU"}
+
+ require.NoError(t, w.AddPosition(dataPath, 1, 0, partition))
+ require.NoError(t, w.AddPosition(dataPath, 3, 0, partition))
+ require.NoError(t, w.AddPosition(dataPath, 1, 0, map[int]any{1000:
"US"}))
+
+ dataFiles, err := w.Flush(context.Background(),
"mem://test/add-position.puffin")
+ require.NoError(t, err)
+ require.Len(t, dataFiles, 1)
+ assert.Equal(t, int64(2), dataFiles[0].Count())
+ assert.Equal(t, partition, dataFiles[0].Partition())
Review Comment:
`AddPosition` goes through the same maps.Clone path as `Add`, but the
defensive-copy invariant isn't exercised for it. This test reuses the partition
map across calls but never mutates it after capture, and there's no
`AddPosition` equivalent of `TestDVWriterAddDefensiveCopies`.
I'd add one that mutates the caller's map after the `AddPosition` call and
asserts `Partition()` still returns the original, so the clone stays pinned for
both methods.
##########
table/dv/dv_writer_bench_test.go:
##########
@@ -0,0 +1,46 @@
+// 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 dv
+
+import "testing"
+
+func BenchmarkDVWriterAddSinglePosition(b *testing.B) {
+ w := NewDVWriter(nil, nil)
+ const dataFilePath = "s3://bucket/data/file.parquet"
+
+ b.ReportAllocs()
+ b.ResetTimer()
+ for i := range b.N {
Review Comment:
both benchmarks accumulate all b.N positions into a single writer without
ever flushing, so the bitmap grows from 0 to roughly b.N across the run. The
relative comparison between the two still holds since they share that shape,
but the absolute ns/op is measuring Set into an ever-larger bitmap rather than
the typical per-commit cost.
I'd flush and re-create the writer each iteration (or seed a fixed bitmap
and benchmark the single add on top of it) if we want the absolute numbers to
mean anything.
##########
table/dv/dv_writer_bench_test.go:
##########
@@ -0,0 +1,46 @@
+// 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 dv
+
+import "testing"
+
+func BenchmarkDVWriterAddSinglePosition(b *testing.B) {
+ w := NewDVWriter(nil, nil)
+ const dataFilePath = "s3://bucket/data/file.parquet"
+
+ b.ReportAllocs()
Review Comment:
two small consistency nits while we're here. Every other benchmark in the
repo calls `b.ResetTimer()` before `b.ReportAllocs()` (see
partitions_bench_test.go and equality_delete_reader_bench_test.go), and uses
`for i := 0; i < b.N; i++` rather than the `range b.N` form. Both are harmless,
but worth matching the house style in the two new benchmarks.
##########
table/dv/dv_writer.go:
##########
@@ -124,6 +124,33 @@ func (w *DVWriter) Add(dataFilePath string, positions
[]int64, specID int32, par
return nil
}
+// AddPosition accumulates one position to delete for a given data file.
+// It has the same first-write partition metadata and validation semantics as
+// Add, without requiring callers that process one deleted row at a time to
+// allocate a one-element positions slice.
+func (w *DVWriter) AddPosition(dataFilePath string, position int64, specID
int32, partitionData map[int]any) error {
+ if position < 0 {
+ return fmt.Errorf("%w: invalid deletion position %d for %q:
positions must be >= 0",
+ iceberg.ErrInvalidArgument, position, dataFilePath)
+ }
+
+ entry, ok := w.entries[dataFilePath]
+ if !ok {
+ // Keep the same defensive-copy and first-write-wins semantics
as Add.
+ entry = &dvEntry{
Review Comment:
this dvEntry creation block (NewRoaringPositionBitmap, copy specID,
maps.Clone, insert into entries, append to order) is now verbatim in `Add`,
`Load`, and here. Two was already a bit of a smell; three means any new dvEntry
field has to be updated in three spots.
Could we pull it into a small helper and let each caller do its own mutation?
```go
func (w *DVWriter) getOrCreateEntry(dataFilePath string, specID int32,
partitionData map[int]any) *dvEntry {
entry, ok := w.entries[dataFilePath]
if !ok {
entry = &dvEntry{
bitmap: NewRoaringPositionBitmap(),
specID: specID,
partitionData: maps.Clone(partitionData),
}
w.entries[dataFilePath] = entry
w.order = append(w.order, dataFilePath)
}
return entry
}
```
`Add` and `Load` keep their bitmap.Or, `AddPosition` does its bitmap.Set.
Thoughts?
--
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]
