You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2026/08/18 21:33:13 UTC
(iceberg-go) branch main updated: perf(table): decode metadata updates in one pass (#1822)
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 8f6e79118 perf(table): decode metadata updates in one pass (#1822)
8f6e79118 is described below
commit 8f6e791182d4268c02bd0261b135f8134047f124
Author: Minh Vu <vu...@gmail.com>
AuthorDate: Tue Aug 18 23:33:08 2026 +0200
perf(table): decode metadata updates in one pass (#1822)
* perf(table): decode metadata updates in one pass
Signed-off-by: Minh Vu <vu...@gmail.com>
* perf(table): improve metadata update benchmark
Signed-off-by: Minh Vu <vu...@gmail.com>
* fix(table): align action decoding with JSON semantics
Signed-off-by: Minh Vu <vu...@gmail.com>
* fix(table): preserve duplicate action decode errors
Signed-off-by: Minh Vu <vu...@gmail.com>
* fix(table): preserve discriminator errors on early exits
Signed-off-by: Minh Vu <vu...@gmail.com>
* perf(table): avoid redundant action decoding
Signed-off-by: Minh Vu <vu...@gmail.com>
* style(table): satisfy nlreturn
Signed-off-by: Minh Vu <vu...@gmail.com>
---------
Signed-off-by: Minh Vu <vu...@gmail.com>
---
table/updates.go | 115 +++++++++++++++++++++++++++++++++----------
table/updates_bench_test.go | 117 ++++++++++++++++++++++++++++++++++++++++++++
table/updates_test.go | 82 +++++++++++++++++++++++++++++++
3 files changed, 287 insertions(+), 27 deletions(-)
diff --git a/table/updates.go b/table/updates.go
index d59c66a47..5c002623e 100644
--- a/table/updates.go
+++ b/table/updates.go
@@ -25,6 +25,7 @@ import (
"fmt"
"maps"
"slices"
+ "strings"
"github.com/apache/iceberg-go"
"github.com/apache/iceberg-go/io"
@@ -100,7 +101,7 @@ var requiredUpdateFields = map[string][]string{
UpdateRemoveEncryptionKey: {"key-id"},
}
-func normalizeLegacyPropertyFields(action string, raw json.RawMessage) (json.RawMessage, error) {
+func normalizeLegacyPropertyFields(action string, object map[string]json.RawMessage) bool {
var modern, legacy string
switch action {
case UpdateSetProperties:
@@ -108,35 +109,30 @@ func normalizeLegacyPropertyFields(action string, raw json.RawMessage) (json.Raw
case UpdateRemoveProperties:
modern, legacy = "removals", "removed"
default:
- return raw, nil
- }
-
- var object map[string]json.RawMessage
- if err := json.Unmarshal(raw, &object); err != nil {
- return nil, err
+ return false
}
+ normalized := false
if _, ok := object[modern]; !ok {
if value, ok := object[legacy]; ok {
object[modern] = value
+ normalized = true
}
}
- delete(object, legacy)
+ if _, ok := object[legacy]; ok {
+ delete(object, legacy)
+ normalized = true
+ }
- return json.Marshal(object)
+ return normalized
}
-func validateRequiredUpdateFields(action string, raw json.RawMessage) error {
+func validateRequiredUpdateFields(action string, object map[string]json.RawMessage) error {
fields := requiredUpdateFields[action]
if len(fields) == 0 {
return nil
}
- var object map[string]json.RawMessage
- if err := json.Unmarshal(raw, &object); err != nil {
- return err
- }
-
for _, field := range fields {
value, ok := object[field]
if !ok || bytes.Equal(bytes.TrimSpace(value), []byte("null")) {
@@ -147,6 +143,14 @@ func validateRequiredUpdateFields(action string, raw json.RawMessage) error {
return nil
}
+func validateUpdateActionDecode(raw json.RawMessage) error {
+ var actionWire struct {
+ Action *string `json:"action"`
+ }
+
+ return json.Unmarshal(raw, &actionWire)
+}
+
func (u *Updates) UnmarshalJSON(data []byte) error {
var rawUpdates []json.RawMessage
if err := json.Unmarshal(data, &rawUpdates); err != nil {
@@ -158,17 +162,49 @@ func (u *Updates) UnmarshalJSON(data []byte) error {
updates = make(Updates, 0, len(rawUpdates))
}
for _, raw := range rawUpdates {
- var baseWire struct {
- Action *string `json:"action"`
- }
- if err := json.Unmarshal(raw, &baseWire); err != nil {
+ var object map[string]json.RawMessage
+ if err := json.Unmarshal(raw, &object); err != nil {
return err
}
- if baseWire.Action == nil {
- return fmt.Errorf("%w: update requires field %q", iceberg.ErrInvalidArgument, "action")
+
+ actionValue, exactAction := object["action"]
+ actionFieldCount := 0
+ caseVariantAction := false
+ for field := range object {
+ if strings.EqualFold(field, "action") {
+ actionFieldCount++
+ if field != "action" {
+ caseVariantAction = true
+ }
+ }
}
- base := baseUpdate{ActionName: *baseWire.Action}
+ var action string
+ if exactAction && !caseVariantAction {
+ if bytes.Equal(bytes.TrimSpace(actionValue), []byte("null")) {
+ if err := validateUpdateActionDecode(raw); err != nil {
+ return err
+ }
+
+ return fmt.Errorf("%w: update requires field %q", iceberg.ErrInvalidArgument, "action")
+ }
+ if err := json.Unmarshal(actionValue, &action); err != nil {
+ return err
+ }
+ } else {
+ var baseWire struct {
+ Action *string `json:"action"`
+ }
+ if err := json.Unmarshal(raw, &baseWire); err != nil {
+ return err
+ }
+ if baseWire.Action == nil {
+ return fmt.Errorf("%w: update requires field %q", iceberg.ErrInvalidArgument, "action")
+ }
+ action = *baseWire.Action
+ }
+
+ base := baseUpdate{ActionName: action}
var upd Update
switch base.ActionName {
@@ -219,14 +255,39 @@ func (u *Updates) UnmarshalJSON(data []byte) error {
case UpdateRemoveEncryptionKey:
upd = &removeEncryptionKeyUpdate{}
default:
+ if err := validateUpdateActionDecode(raw); err != nil {
+ return err
+ }
+
return fmt.Errorf("%w: unknown update action: %s", iceberg.ErrInvalidArgument, base.ActionName)
}
- normalized, err := normalizeLegacyPropertyFields(base.ActionName, raw)
- if err != nil {
- return err
+ if normalizeLegacyPropertyFields(base.ActionName, object) {
+ if err := validateUpdateActionDecode(raw); err != nil {
+ return err
+ }
+ if actionFieldCount > 1 {
+ for field := range object {
+ if strings.EqualFold(field, "action") {
+ delete(object, field)
+ }
+ }
+ actionJSON, err := json.Marshal(base.ActionName)
+ if err != nil {
+ return err
+ }
+ object["action"] = actionJSON
+ }
+ normalized, err := json.Marshal(object)
+ if err != nil {
+ return err
+ }
+ raw = normalized
}
- raw = normalized
- if err := validateRequiredUpdateFields(base.ActionName, raw); err != nil {
+ if err := validateRequiredUpdateFields(base.ActionName, object); err != nil {
+ if actionErr := validateUpdateActionDecode(raw); actionErr != nil {
+ return actionErr
+ }
+
return err
}
diff --git a/table/updates_bench_test.go b/table/updates_bench_test.go
new file mode 100644
index 000000000..1605d075a
--- /dev/null
+++ b/table/updates_bench_test.go
@@ -0,0 +1,117 @@
+// 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 (
+ "encoding/json"
+ "fmt"
+ "testing"
+)
+
+var benchmarkUpdates Updates
+
+func BenchmarkUnmarshalUpdates(b *testing.B) {
+ for _, legacy := range []bool{false, true} {
+ label := "canonical"
+ if legacy {
+ label = "legacy-properties"
+ }
+ for _, count := range []int{1, 10, 100, 1000} {
+ b.Run(fmt.Sprintf("%s/%d", label, count), func(b *testing.B) {
+ payload, err := benchmarkUpdatesPayload(count, legacy)
+ if err != nil {
+ b.Fatal(err)
+ }
+
+ b.ReportAllocs()
+ b.SetBytes(int64(len(payload)))
+ b.ResetTimer()
+ for b.Loop() {
+ var updates Updates
+ if err := json.Unmarshal(payload, &updates); err != nil {
+ b.Fatal(err)
+ }
+ if len(updates) != count {
+ b.Fatalf("decoded %d updates, want %d", len(updates), count)
+ }
+ benchmarkUpdates = updates
+ }
+ })
+ }
+ }
+}
+
+func benchmarkUpdatesPayload(count int, legacyProperties bool) ([]byte, error) {
+ updates := make([]map[string]any, count)
+ for i := range updates {
+ switch i % 8 {
+ case 0:
+ updates[i] = map[string]any{
+ "action": "assign-uuid",
+ "uuid": "550e8400-e29b-41d4-a716-446655440000",
+ }
+ case 1:
+ updates[i] = map[string]any{
+ "action": "upgrade-format-version",
+ "format-version": 2,
+ }
+ case 2:
+ updates[i] = map[string]any{
+ "action": "set-location",
+ "location": "s3://bucket/table",
+ }
+ case 3:
+ field := "updates"
+ if legacyProperties {
+ field = "updated"
+ }
+ updates[i] = map[string]any{
+ "action": "set-properties",
+ field: map[string]string{"key": "value"},
+ }
+ case 4:
+ field := "removals"
+ if legacyProperties {
+ field = "removed"
+ }
+ updates[i] = map[string]any{
+ "action": "remove-properties",
+ field: []string{"key"},
+ }
+ case 5:
+ updates[i] = map[string]any{
+ "action": "remove-schemas",
+ "schema-ids": []int{1, 2, 3},
+ }
+ case 6:
+ updates[i] = map[string]any{
+ "action": "remove-partition-specs",
+ "spec-ids": []int{1, 2, 3},
+ }
+ default:
+ updates[i] = map[string]any{
+ "action": "set-snapshot-ref",
+ "ref-name": "main",
+ "type": "branch",
+ "snapshot-id": 1,
+ }
+ }
+ }
+
+ return json.Marshal(updates)
+}
diff --git a/table/updates_test.go b/table/updates_test.go
index abf81fe63..fd79aea90 100644
--- a/table/updates_test.go
+++ b/table/updates_test.go
@@ -768,6 +768,88 @@ func TestUnmarshalUpdatesRejectsMissingOrNullAction(t *testing.T) {
}
}
+func TestUnmarshalUpdatesUsesJSONActionFieldSemantics(t *testing.T) {
+ tests := []struct {
+ name string
+ data string
+ expectedAction string
+ assert func(*testing.T, Update)
+ }{
+ {
+ name: "accepts mixed-case action key",
+ data: `[{"Action":"set-location","location":"s3://bucket/table"}]`,
+ expectedAction: UpdateSetLocation,
+ assert: func(t *testing.T, update Update) {
+ t.Helper()
+ location, ok := update.(*setLocationUpdate)
+ require.True(t, ok)
+ assert.Equal(t, "s3://bucket/table", location.Location)
+ },
+ },
+ {
+ name: "uses the last duplicate case-variant action key",
+ data: `[{"action":"set-location","ACTION":"assign-uuid","location":"s3://bucket/table","uuid":"550e8400-e29b-41d4-a716-446655440000"}]`,
+ expectedAction: UpdateAssignUUID,
+ assert: func(t *testing.T, update Update) {
+ t.Helper()
+ assignUUID, ok := update.(*assignUUIDUpdate)
+ require.True(t, ok)
+ assert.Equal(t, uuid.MustParse("550e8400-e29b-41d4-a716-446655440000"), assignUUID.UUID)
+ },
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ var updates Updates
+ require.NoError(t, json.Unmarshal([]byte(tt.data), &updates))
+ require.Len(t, updates, 1)
+ assert.Equal(t, tt.expectedAction, updates[0].Action())
+ tt.assert(t, updates[0])
+ })
+ }
+
+ t.Run("rejects unknown action from duplicate case-variant key", func(t *testing.T) {
+ var updates Updates
+ err := json.Unmarshal([]byte(`[{"action":"set-location","ACTION":"unknown","location":"s3://bucket/table"}]`), &updates)
+ require.ErrorContains(t, err, "unknown update action: unknown")
+ })
+}
+
+func TestUnmarshalUpdatesPreservesDuplicateActionTypeError(t *testing.T) {
+ tests := []struct {
+ name string
+ data string
+ }{
+ {
+ name: "valid final action",
+ data: `[{"action":123,"action":"set-properties","updated":{"k":"v"}}]`,
+ },
+ {
+ name: "valid final action with missing required field",
+ data: `[{"action":123,"action":"set-location"}]`,
+ },
+ {
+ name: "unknown final action",
+ data: `[{"action":123,"action":"unknown"}]`,
+ },
+ {
+ name: "null final action",
+ data: `[{"action":123,"action":null}]`,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ var updates Updates
+ var typeErr *json.UnmarshalTypeError
+ err := json.Unmarshal([]byte(tt.data), &updates)
+
+ require.ErrorAs(t, err, &typeErr)
+ })
+ }
+}
+
func TestUnmarshalUpdatesAcceptsOptionalAndEmptyValues(t *testing.T) {
data := []byte(`[
{"action":"add-schema","schema":{"type":"struct","fields":[]}},