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 d3c4accf8 perf(rest): reuse adjacent scan task residuals (#2224)
d3c4accf8 is described below
commit d3c4accf83cd58de91dc5e42499da1309c456186
Author: Minh Vu <[email protected]>
AuthorDate: Tue Oct 20 19:16:03 2026 +0200
perf(rest): reuse adjacent scan task residuals (#2224)
* perf(rest): reuse consecutive scan task residual expressions
* perf(rest): fast-path absent scan task residuals
* chore: keep benchmark candidate free of temporary workflow
* fix(rest): satisfy nlreturn lint in residual decoder
---
catalog/rest/scan_task_decoder.go | 35 ++++++-
.../rest/scan_task_residual_reuse_bench_test.go | 110 +++++++++++++++++++++
catalog/rest/scan_task_residual_reuse_test.go | 84 ++++++++++++++++
3 files changed, 228 insertions(+), 1 deletion(-)
diff --git a/catalog/rest/scan_task_decoder.go
b/catalog/rest/scan_task_decoder.go
index 1ed3900a5..337eab18a 100644
--- a/catalog/rest/scan_task_decoder.go
+++ b/catalog/rest/scan_task_decoder.go
@@ -161,6 +161,7 @@ func decodeScanTasks(
out := make([]table.FileScanTask, 0, len(wire.FileScanTasks))
dvOwners := make(map[int]string)
deleteRefTaskIDs := make([]int, len(deletes))
+ var residuals taskResidualDecoder
for i := range wire.FileScanTasks {
wireTask := wire.FileScanTasks[i]
if wireTask.DataFile == nil {
@@ -172,7 +173,7 @@ func decodeScanTasks(
return nil, fmt.Errorf("%w: decoding scan tasks:
file-scan-tasks[%d].data-file: %w", ErrRESTError, i, err)
}
- residual, err := decodeTaskResidual(wireTask.ResidualFilter,
schema, fallbackResidual)
+ residual, err := residuals.decode(wireTask.ResidualFilter,
schema, fallbackResidual)
if err != nil {
return nil, fmt.Errorf("%w: decoding scan tasks:
file-scan-tasks[%d].residual-filter: %w", ErrRESTError, i, err)
}
@@ -809,6 +810,38 @@ func decodeJSONFloat(raw json.RawMessage, bitSize int)
(float64, error) {
return parsed, nil
}
+// taskResidualDecoder avoids reparsing an identical residual for adjacent scan
+// tasks. It is local to one envelope: no global cache or retained task JSON.
+type taskResidualDecoder struct {
+ lastRaw json.RawMessage
+ lastValue iceberg.BooleanExpression
+}
+
+func (d *taskResidualDecoder) decode(
+ raw json.RawMessage,
+ schema *iceberg.Schema,
+ fallback iceberg.BooleanExpression,
+) (iceberg.BooleanExpression, error) {
+ // Most scans omit per-task residuals. Preserve that zero-allocation
+ // fallback and do not retain a cached expression across missing
filters.
+ if len(raw) == 0 {
+ d.lastRaw, d.lastValue = nil, nil
+
+ return fallback, nil
+ }
+ if len(d.lastRaw) > 0 && bytes.Equal(raw, d.lastRaw) {
+ return d.lastValue, nil
+ }
+
+ value, err := decodeTaskResidual(raw, schema, fallback)
+ if err != nil {
+ return nil, err
+ }
+ d.lastRaw, d.lastValue = raw, value
+
+ return value, nil
+}
+
func decodeTaskResidual(
raw json.RawMessage,
schema *iceberg.Schema,
diff --git a/catalog/rest/scan_task_residual_reuse_bench_test.go
b/catalog/rest/scan_task_residual_reuse_bench_test.go
new file mode 100644
index 000000000..d636a6316
--- /dev/null
+++ b/catalog/rest/scan_task_residual_reuse_bench_test.go
@@ -0,0 +1,110 @@
+// 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 rest
+
+import (
+ "encoding/json"
+ "fmt"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+)
+
+var residualReuseBenchmarkSink iceberg.BooleanExpression
+
+func BenchmarkScanTaskResidualReuse(b *testing.B) {
+ metadata := newScanTaskDecoderMetadata()
+ for _, taskCount := range []int{64, 1024} {
+ for _, distribution := range []string{"identical",
"runs-of-16", "unique", "missing"} {
+ residuals := make([]json.RawMessage, taskCount)
+ for i := range residuals {
+ if distribution == "missing" {
+ continue
+ }
+ value := 34
+ switch distribution {
+ case "runs-of-16":
+ value += i / 16
+ case "unique":
+ value += i
+ }
+ // Give each task its own JSON buffer, as in
real decoded wire payloads.
+ residuals[i] =
json.RawMessage(fmt.Sprintf(`{"type":"eq","term":"id","value":%d}`, value))
+ }
+ for _, variant := range []struct {
+ name string
+ cached bool
+ }{{"uncached", false}, {"cached", true}} {
+ b.Run(fmt.Sprintf("tasks=%d/%s/%s", taskCount,
distribution, variant.name), func(b *testing.B) {
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ var cache taskResidualDecoder
+ for _, raw := range residuals {
+ var (
+ value
iceberg.BooleanExpression
+ err error
+ )
+ if variant.cached {
+ value, err =
cache.decode(raw, metadata.schema, nil)
+ } else {
+ value, err =
decodeTaskResidual(raw, metadata.schema, nil)
+ }
+ if err != nil {
+ b.Fatal(err)
+ }
+
residualReuseBenchmarkSink = value
+ }
+ }
+ })
+ }
+ }
+ }
+}
+
+func BenchmarkDecodeScanTasksWithResiduals(b *testing.B) {
+ metadata := newScanTaskDecoderMetadata()
+ base := validScanTasksWire().FileScanTasks[0].DataFile
+ for _, distribution := range []string{"identical", "unique", "missing"}
{
+ const taskCount = 1024
+ wire := ScanTasks{FileScanTasks: make([]RESTFileScanTask,
taskCount)}
+ for i := range wire.FileScanTasks {
+ data := *base
+ var raw json.RawMessage
+ if distribution != "missing" {
+ value := 34
+ if distribution == "unique" {
+ value += i
+ }
+ raw =
json.RawMessage(fmt.Sprintf(`{"type":"eq","term":"id","value":%d}`, value))
+ }
+ wire.FileScanTasks[i] = RESTFileScanTask{DataFile:
&data, ResidualFilter: raw}
+ }
+ b.Run(distribution, 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_residual_reuse_test.go
b/catalog/rest/scan_task_residual_reuse_test.go
new file mode 100644
index 000000000..1d60db741
--- /dev/null
+++ b/catalog/rest/scan_task_residual_reuse_test.go
@@ -0,0 +1,84 @@
+// 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 rest
+
+import (
+ "encoding/json"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestDecodeScanTasksReusesAdjacentResidualFilters(t *testing.T) {
+ t.Parallel()
+
+ metadata := newScanTaskDecoderMetadata()
+ base := validScanTasksWire().FileScanTasks[0].DataFile
+ fallback := iceberg.AlwaysTrue{}
+ filters := []string{
+ `{"type":"eq","term":"id","value":34}`,
+ `{"type":"eq","term":"id","value":34}`,
+ `{"type":"eq","term":"id","value":35}`,
+ `{"type":"eq","term":"id","value":34}`,
+ "",
+ `{"type":"eq","term":"id","value":34}`,
+ }
+ wire := ScanTasks{FileScanTasks: make([]RESTFileScanTask, len(filters))}
+ for i, filter := range filters {
+ data := *base
+ wire.FileScanTasks[i] = RESTFileScanTask{
+ DataFile: &data,
+ ResidualFilter: json.RawMessage(filter),
+ }
+ }
+
+ tasks, err := DecodeScanTasks(wire, metadata, metadata.schema, fallback)
+ require.NoError(t, err)
+ require.Len(t, tasks, len(filters))
+
+ assert.True(t,
tasks[0].Residual.Equals(iceberg.EqualTo(iceberg.Reference("id"), int64(34))))
+ assert.Same(t, tasks[0].Residual, tasks[1].Residual, "consecutive equal
filters should reuse the parsed expression")
+ assert.True(t,
tasks[2].Residual.Equals(iceberg.EqualTo(iceberg.Reference("id"), int64(35))))
+ assert.True(t,
tasks[3].Residual.Equals(iceberg.EqualTo(iceberg.Reference("id"), int64(34))))
+ assert.NotSame(t, tasks[0].Residual, tasks[3].Residual, "different
filters should invalidate the consecutive cache")
+ assert.Equal(t, fallback, tasks[4].Residual)
+ assert.True(t,
tasks[5].Residual.Equals(iceberg.EqualTo(iceberg.Reference("id"), int64(34))))
+ assert.NotSame(t, tasks[3].Residual, tasks[5].Residual, "missing
filters should reset the consecutive cache")
+}
+
+func TestDecodeScanTasksResidualReuseDoesNotHideErrors(t *testing.T) {
+ t.Parallel()
+
+ metadata := newScanTaskDecoderMetadata()
+ base := validScanTasksWire().FileScanTasks[0].DataFile
+ wire := ScanTasks{FileScanTasks: make([]RESTFileScanTask, 3)}
+ for i := range wire.FileScanTasks {
+ data := *base
+ wire.FileScanTasks[i] = RESTFileScanTask{
+ DataFile: &data,
+ ResidualFilter:
json.RawMessage(`{"type":"eq","term":"id","value":34}`),
+ }
+ }
+ wire.FileScanTasks[2].ResidualFilter = json.RawMessage(`null`)
+
+ _, err := DecodeScanTasks(wire, metadata, metadata.schema, nil)
+ require.ErrorContains(t, err, "file-scan-tasks[2].residual-filter")
+ require.ErrorContains(t, err, "explicit null residual-filter is
invalid")
+}
