This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 1fdbecb79 perf(rest): decode derived deletion vectors once (#2271)
1fdbecb79 is described below
commit 1fdbecb79954d3f53b73ce62662b59bc67d8bae1
Author: Minh Vu <[email protected]>
AuthorDate: Fri Oct 23 18:33:01 2026 +0200
perf(rest): decode derived deletion vectors once (#2271)
perf(rest): decode shared and derived deletion vectors once
Signed-off-by: Minh Vu <[email protected]>
---
catalog/rest/scan_task_decoder.go | 62 +++++++++++++++++++++++++++-
catalog/rest/scan_task_decoder_bench_test.go | 24 +++++++++++
catalog/rest/scan_task_decoder_test.go | 23 +++++++++++
3 files changed, 107 insertions(+), 2 deletions(-)
diff --git a/catalog/rest/scan_task_decoder.go
b/catalog/rest/scan_task_decoder.go
index 337eab18a..f61c667d3 100644
--- a/catalog/rest/scan_task_decoder.go
+++ b/catalog/rest/scan_task_decoder.go
@@ -27,6 +27,7 @@ import (
"reflect"
"slices"
"strconv"
+ "strings"
"github.com/apache/iceberg-go"
"github.com/apache/iceberg-go/table"
@@ -123,6 +124,52 @@ func DecodeScanTasks(
return decodeScanTasks(wire, metadata, schema, fallbackResidual,
newPartitionDecodePlanCache())
}
+// deriveDeletionVectorOwners identifies a unique data-file path for unowned
+// Puffin deletion vectors. A conflicting owner stays unresolved so the normal
+// task validation can report the conflict in its original order.
+func deriveDeletionVectorOwners(wire ScanTasks) []string {
+ hasDerivedDV := false
+ for _, del := range wire.DeleteFiles {
+ if del.ReferencedDataFile == nil &&
+ strings.EqualFold(del.FileFormat,
string(iceberg.PuffinFile)) {
+ hasDerivedDV = true
+
+ break
+ }
+ }
+ if !hasDerivedDV {
+ return nil
+ }
+
+ owners := make([]string, len(wire.DeleteFiles))
+ conflicts := make([]bool, len(wire.DeleteFiles))
+ for _, task := range wire.FileScanTasks {
+ if task.DataFile == nil || task.DataFile.FilePath == "" {
+ continue
+ }
+ path := task.DataFile.FilePath
+ for _, ref := range task.DeleteFileReferences {
+ // Leave malformed references for the regular
validation pass.
+ if ref < 0 || ref >= len(wire.DeleteFiles) ||
conflicts[ref] {
+ continue
+ }
+ del := &wire.DeleteFiles[ref]
+ if del.ReferencedDataFile != nil ||
+ !strings.EqualFold(del.FileFormat,
string(iceberg.PuffinFile)) {
+ continue
+ }
+ if owners[ref] == "" {
+ owners[ref] = path
+ } else if owners[ref] != path {
+ owners[ref] = ""
+ conflicts[ref] = true
+ }
+ }
+ }
+
+ return owners
+}
+
func decodeScanTasks(
wire ScanTasks,
metadata table.ScanPlanningMetadata,
@@ -147,11 +194,19 @@ func decodeScanTasks(
return nil, nil
}
+ // Resolve unique owners first so unowned deletion vectors are built
once,
+ // even when multiple scan tasks reference the same vector.
+ derivedDVOwners := deriveDeletionVectorOwners(wire)
+
// Decode every delete before resolving references. This validates all
wire
// entries, including an unreferenced entry that will be rejected below.
deletes := make([]iceberg.DataFile, len(wire.DeleteFiles))
for i := range wire.DeleteFiles {
- decoded, err := decodeRESTDeleteFile(wire.DeleteFiles[i],
metadata, "", plans)
+ owner := ""
+ if derivedDVOwners != nil {
+ owner = derivedDVOwners[i]
+ }
+ decoded, err := decodeRESTDeleteFile(wire.DeleteFiles[i],
metadata, owner, plans)
if err != nil {
return nil, fmt.Errorf("%w: decoding scan tasks:
delete-files[%d]: %w", ErrRESTError, i, err)
}
@@ -220,7 +275,10 @@ func decodeScanTasks(
dvOwners[ref] = dataFile.FilePath()
// Derive referenced-data-file from the
FileScanTask association
// when the server omits Java's optional
extension field.
- if wireDelete.ReferencedDataFile == nil {
+ if wireDelete.ReferencedDataFile == nil &&
+ (derivedDVOwners == nil ||
derivedDVOwners[ref] != dataFile.FilePath()) {
+ // Conflicting or missing owners fall
back to the original
+ // per-reference decoder, preserving
the validation path.
deleteFile, err =
decodeRESTDeleteFile(wireDelete, metadata, dataFile.FilePath(), plans)
if err != nil {
return nil, fmt.Errorf("%w:
decoding scan tasks: delete-files[%d]: %w", ErrRESTError, ref, err)
diff --git a/catalog/rest/scan_task_decoder_bench_test.go
b/catalog/rest/scan_task_decoder_bench_test.go
index 6ac8466db..1b368f2e5 100644
--- a/catalog/rest/scan_task_decoder_bench_test.go
+++ b/catalog/rest/scan_task_decoder_bench_test.go
@@ -178,3 +178,27 @@ func BenchmarkDecodeJSONNumberForms(b *testing.B) {
})
}
}
+
+func BenchmarkDecodeScanTasksSharedDeletionVector(b *testing.B) {
+ metadata := newScanTaskDecoderMetadata()
+ for _, taskCount := range []int{100, 1000, 10000} {
+ wire := deletionVectorScanTasksWire(1, false)
+ task := wire.FileScanTasks[0]
+ wire.FileScanTasks = make([]RESTFileScanTask, taskCount)
+ for i := range wire.FileScanTasks {
+ wire.FileScanTasks[i] = task
+ }
+
+ b.Run(fmt.Sprintf("derived-owner/tasks=%d", taskCount), func(b
*testing.B) {
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ tasks, err := DecodeScanTasks(wire, metadata,
metadata.schema, nil)
+ if err != nil {
+ b.Fatal(err)
+ }
+ decodeScanTasksBenchmarkSink = len(tasks)
+ }
+ })
+ }
+}
diff --git a/catalog/rest/scan_task_decoder_test.go
b/catalog/rest/scan_task_decoder_test.go
index 7d0022505..1ad68e576 100644
--- a/catalog/rest/scan_task_decoder_test.go
+++ b/catalog/rest/scan_task_decoder_test.go
@@ -310,6 +310,29 @@ func
TestDecodeScanTasksDerivesDeletionVectorTargetWhenOmitted(t *testing.T) {
assert.Equal(t, new("s3://bucket/table/data.parquet"),
tasks[0].DeletionVectorFiles[0].ReferencedDataFile())
}
+func TestDecodeScanTasksSharesDerivedDeletionVector(t *testing.T) {
+ t.Parallel()
+
+ metadata := newScanTaskDecoderMetadata()
+ wire := validScanTasksWire()
+ wire.DeleteFiles[0].FileFormat = "PUFFIN"
+ wire.DeleteFiles[0].ContentOffset = new(int64(10))
+ wire.DeleteFiles[0].ContentSizeInBytes = new(int64(20))
+ wire.FileScanTasks[0].DeleteFileReferences = []int{0}
+ // Two file-scan tasks for the same data file share a DV without an
+ // explicit referenced-data-file on the REST wire.
+ wire.FileScanTasks = append(wire.FileScanTasks, wire.FileScanTasks[0])
+
+ tasks, err := DecodeScanTasks(wire, metadata, metadata.schema, nil)
+ require.NoError(t, err)
+ require.Len(t, tasks, 2)
+ require.Len(t, tasks[0].DeletionVectorFiles, 1)
+ require.Len(t, tasks[1].DeletionVectorFiles, 1)
+ assert.Equal(t, new("s3://bucket/table/data.parquet"),
+ tasks[0].DeletionVectorFiles[0].ReferencedDataFile())
+ assert.Same(t, tasks[0].DeletionVectorFiles[0],
tasks[1].DeletionVectorFiles[0])
+}
+
func TestRESTValueMapMatchesJavaContentFileParser(t *testing.T) {
t.Parallel()
