This is an automated email from the ASF dual-hosted git repository.
laskoviymishka 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 8a16908d4 fix(udf): replace state during JSON decoding (#1626)
8a16908d4 is described below
commit 8a16908d4582f4cf4c80c0a758804d88520f88ea
Author: Minh Vu <[email protected]>
AuthorDate: Tue Aug 4 12:19:22 2026 +0200
fix(udf): replace state during JSON decoding (#1626)
## Summary
Make UDF JSON decoding replace receiver state instead of merging into
it.
## Problem
Several UDF `UnmarshalJSON` implementations decode through aliases of
their existing receivers. Go leaves struct fields unchanged when they
are absent from a new JSON document, so reusing a receiver can retain
values from an earlier decode.
At the metadata level, a malformed document without `definitions` can
pass validation using retained definitions, and the lazy definition
index can remain stale. Exported nested values such as `Parameter`,
`DefinitionVersion`, `Definition`, and `DefinitionLogEntry` also retain
omitted optional fields when decoded directly.
## Fix
Decode each value into fresh state and assign it to the receiver only
after decoding succeeds. Top-level metadata is also validated before
assignment, then initializes its definition index from the replacement
state.
## Tests
Added receiver-reuse coverage for top-level metadata and exported nested
values. The tests verify omitted optional fields are cleared, cached
indexes are replaced, and failed metadata decoding does not partially
overwrite the last valid state.
```
go test ./udf -count=1
```
---
udf/metadata.go | 71 +++++++++++++++++++++++++--------------------
udf/metadata_test.go | 82 ++++++++++++++++++++++++++++++++++++++++++++++++++++
2 files changed, 121 insertions(+), 32 deletions(-)
diff --git a/udf/metadata.go b/udf/metadata.go
index 88e3841f7..69c914736 100644
--- a/udf/metadata.go
+++ b/udf/metadata.go
@@ -204,24 +204,26 @@ type Parameter struct {
func (p *Parameter) UnmarshalJSON(b []byte) error {
type Alias Parameter
+ next := Parameter{}
aux := struct {
Type json.RawMessage `json:"type"`
*Alias
- }{Alias: (*Alias)(p)}
+ }{Alias: (*Alias)(&next)}
if err := json.Unmarshal(b, &aux); err != nil {
return err
}
if len(aux.Type) == 0 {
- return fmt.Errorf("%w: parameter %q requires a type",
ErrInvalidUDFMetadata, p.Name)
+ return fmt.Errorf("%w: parameter %q requires a type",
ErrInvalidUDFMetadata, next.Name)
}
typ, err := unmarshalType(aux.Type)
if err != nil {
return err
}
- p.Type = typ
+ next.Type = typ
+ *p = next
return nil
}
@@ -252,26 +254,25 @@ type DefinitionVersion struct {
func (v *DefinitionVersion) UnmarshalJSON(b []byte) error {
type Alias DefinitionVersion
+ next := DefinitionVersion{VersionID: -1, TimestampMS: -1}
aux := struct {
Representations []json.RawMessage `json:"representations"`
*Alias
- }{Alias: (*Alias)(v)}
-
- v.VersionID = -1
- v.TimestampMS = -1
+ }{Alias: (*Alias)(&next)}
if err := json.Unmarshal(b, &aux); err != nil {
return err
}
- v.Representations = make([]Representation, len(aux.Representations))
+ next.Representations = make([]Representation, len(aux.Representations))
for i, raw := range aux.Representations {
repr, err := unmarshalRepresentation(raw)
if err != nil {
return err
}
- v.Representations[i] = repr
+ next.Representations[i] = repr
}
+ *v = next
return nil
}
@@ -404,6 +405,7 @@ type Definition struct {
func (d *Definition) UnmarshalJSON(b []byte) error {
type Alias Definition
+ next := Definition{CurrentVersionID: -1}
// definition-id and parameters are shadowed to distinguish a required
// field that is absent from one that is legitimately empty: a
// zero-parameter definition has "" as its definition-id and [] as its
@@ -413,9 +415,7 @@ func (d *Definition) UnmarshalJSON(b []byte) error {
Parameters json.RawMessage `json:"parameters"`
ReturnType json.RawMessage `json:"return-type"`
*Alias
- }{Alias: (*Alias)(d)}
-
- d.CurrentVersionID = -1
+ }{Alias: (*Alias)(&next)}
if err := json.Unmarshal(b, &aux); err != nil {
return err
@@ -424,27 +424,28 @@ func (d *Definition) UnmarshalJSON(b []byte) error {
if aux.DefinitionID == nil {
return fmt.Errorf("%w: definition is missing definition-id",
ErrInvalidUDFMetadata)
}
- d.DefinitionID = *aux.DefinitionID
+ next.DefinitionID = *aux.DefinitionID
if len(aux.Parameters) == 0 {
- return fmt.Errorf("%w: definition %q is missing parameters",
ErrInvalidUDFMetadata, d.DefinitionID)
+ return fmt.Errorf("%w: definition %q is missing parameters",
ErrInvalidUDFMetadata, next.DefinitionID)
}
- if err := json.Unmarshal(aux.Parameters, &d.Parameters); err != nil {
+ if err := json.Unmarshal(aux.Parameters, &next.Parameters); err != nil {
return err
}
- if d.Parameters == nil {
- return fmt.Errorf("%w: definition %q is missing parameters",
ErrInvalidUDFMetadata, d.DefinitionID)
+ if next.Parameters == nil {
+ return fmt.Errorf("%w: definition %q is missing parameters",
ErrInvalidUDFMetadata, next.DefinitionID)
}
if len(aux.ReturnType) == 0 {
- return fmt.Errorf("%w: definition %q requires a return-type",
ErrInvalidUDFMetadata, d.DefinitionID)
+ return fmt.Errorf("%w: definition %q requires a return-type",
ErrInvalidUDFMetadata, next.DefinitionID)
}
typ, err := unmarshalType(aux.ReturnType)
if err != nil {
return err
}
- d.ReturnType = typ
+ next.ReturnType = typ
+ *d = next
return nil
}
@@ -605,14 +606,13 @@ type DefinitionVersionRef struct {
func (r *DefinitionVersionRef) UnmarshalJSON(b []byte) error {
type Alias DefinitionVersionRef
+ next := DefinitionVersionRef{VersionID: -1}
// definition-id is shadowed to distinguish the required field being
// absent from the legitimately empty id of a zero-parameter definition.
aux := struct {
DefinitionID *string `json:"definition-id"`
*Alias
- }{Alias: (*Alias)(r)}
-
- r.VersionID = -1
+ }{Alias: (*Alias)(&next)}
if err := json.Unmarshal(b, &aux); err != nil {
return err
@@ -621,7 +621,8 @@ func (r *DefinitionVersionRef) UnmarshalJSON(b []byte)
error {
if aux.DefinitionID == nil {
return fmt.Errorf("%w: definition-log reference is missing
definition-id", ErrInvalidUDFMetadata)
}
- r.DefinitionID = *aux.DefinitionID
+ next.DefinitionID = *aux.DefinitionID
+ *r = next
return nil
}
@@ -639,11 +640,14 @@ type DefinitionLogEntry struct {
func (e *DefinitionLogEntry) UnmarshalJSON(b []byte) error {
type Alias DefinitionLogEntry
- aux := (*Alias)(e)
+ next := DefinitionLogEntry{TimestampMS: -1}
- e.TimestampMS = -1
+ if err := json.Unmarshal(b, (*Alias)(&next)); err != nil {
+ return err
+ }
+ *e = next
- return json.Unmarshal(b, aux)
+ return nil
}
func (e DefinitionLogEntry) validate() error {
@@ -870,14 +874,13 @@ func (m *metadata) init() {
func (m *metadata) UnmarshalJSON(b []byte) error {
type Alias metadata
+ next := metadata{FormatVersionValue: -1}
// definition-log is shadowed to reject the required field being absent
// or null, which the default decoding cannot distinguish from empty.
aux := struct {
DefinitionLog json.RawMessage `json:"definition-log"`
*Alias
- }{Alias: (*Alias)(m)}
-
- m.FormatVersionValue = -1
+ }{Alias: (*Alias)(&next)}
if err := json.Unmarshal(b, &aux); err != nil {
return err
@@ -886,16 +889,20 @@ func (m *metadata) UnmarshalJSON(b []byte) error {
if len(aux.DefinitionLog) == 0 {
return fmt.Errorf("%w: definition-log is required",
ErrInvalidUDFMetadata)
}
- if err := json.Unmarshal(aux.DefinitionLog, &m.DefinitionLogList); err
!= nil {
+ if err := json.Unmarshal(aux.DefinitionLog, &next.DefinitionLogList);
err != nil {
return err
}
- if m.DefinitionLogList == nil {
+ if next.DefinitionLogList == nil {
return fmt.Errorf("%w: definition-log is required",
ErrInvalidUDFMetadata)
}
+ if err := next.validate(); err != nil {
+ return err
+ }
+ *m = next
m.init()
- return m.validate()
+ return nil
}
// indexBy indexes a slice into a map, using a provided extractKey function.
diff --git a/udf/metadata_test.go b/udf/metadata_test.go
index d1de628b9..9dea80072 100644
--- a/udf/metadata_test.go
+++ b/udf/metadata_test.go
@@ -749,6 +749,88 @@ func TestParseMetadataWithEmptyDefinitionLog(t *testing.T)
{
assert.Empty(t, meta.DefinitionLog())
}
+func TestMetadataUnmarshalReplacesReceiverState(t *testing.T) {
+ firstDoc := minimalMetadata(t)
+ firstDoc["doc"] = "old documentation"
+ firstDoc["properties"] = map[string]any{"old": "value"}
+ first, err := json.Marshal(firstDoc)
+ require.NoError(t, err)
+
+ var meta metadata
+ require.NoError(t, json.Unmarshal(first, &meta))
+ _, ok := meta.DefinitionByID("int")
+ require.True(t, ok)
+
+ secondDoc := minimalMetadata(t)
+ secondDefinition := definition(t, secondDoc, 0)
+ secondDefinition["definition-id"] = "long"
+ secondDefinition["parameters"].([]any)[0].(map[string]any)["type"] =
"long"
+ refs :=
secondDoc["definition-log"].([]any)[0].(map[string]any)["definition-versions"].([]any)
+ refs[0].(map[string]any)["definition-id"] = "long"
+ second, err := json.Marshal(secondDoc)
+ require.NoError(t, err)
+ require.NoError(t, json.Unmarshal(second, &meta))
+
+ assert.Empty(t, meta.Doc())
+ assert.Empty(t, meta.Properties())
+ _, ok = meta.DefinitionByID("int")
+ assert.False(t, ok)
+ _, ok = meta.DefinitionByID("long")
+ assert.True(t, ok)
+
+ delete(secondDoc, "definitions")
+ missingDefinitions, err := json.Marshal(secondDoc)
+ require.NoError(t, err)
+ err = json.Unmarshal(missingDefinitions, &meta)
+ assert.ErrorContains(t, err, "at least one definition is required")
+ _, ok = meta.DefinitionByID("long")
+ assert.True(t, ok, "a failed decode must not partially replace valid
metadata")
+}
+
+func TestUDFValuesUnmarshalReplaceReceiverState(t *testing.T) {
+ t.Run("parameter", func(t *testing.T) {
+ var parameter Parameter
+ require.NoError(t,
json.Unmarshal([]byte(`{"name":"x","type":"int","doc":"old"}`), ¶meter))
+ require.NoError(t,
json.Unmarshal([]byte(`{"name":"y","type":"long"}`), ¶meter))
+ assert.Equal(t, "y", parameter.Name)
+ assert.Equal(t, "long", parameter.Type.String())
+ assert.Empty(t, parameter.Doc)
+ })
+
+ t.Run("definition version", func(t *testing.T) {
+ var version DefinitionVersion
+ first :=
`{"version-id":1,"representations":[{"type":"sql","dialect":"trino","sql":"x"}],"deterministic":true,"on-null-input":"return-null","timestamp-ms":1}`
+ second :=
`{"version-id":2,"representations":[{"type":"sql","dialect":"spark","sql":"y"}],"timestamp-ms":2}`
+ require.NoError(t, json.Unmarshal([]byte(first), &version))
+ require.NoError(t, json.Unmarshal([]byte(second), &version))
+ assert.Equal(t, 2, version.VersionID)
+ assert.False(t, version.Deterministic)
+ assert.Empty(t, version.OnNullInput)
+ })
+
+ t.Run("definition", func(t *testing.T) {
+ firstDoc := definition(t, minimalMetadata(t), 0)
+ firstDoc["specific-name"] = "old_name"
+ firstDoc["doc"] = "old documentation"
+ first, err := json.Marshal(firstDoc)
+ require.NoError(t, err)
+ second, err := json.Marshal(definition(t, minimalMetadata(t),
0))
+ require.NoError(t, err)
+
+ var def Definition
+ require.NoError(t, json.Unmarshal(first, &def))
+ require.NoError(t, json.Unmarshal(second, &def))
+ assert.Empty(t, def.SpecificName)
+ assert.Empty(t, def.Doc)
+ })
+
+ t.Run("definition log entry", func(t *testing.T) {
+ entry := DefinitionLogEntry{DefinitionVersions:
[]DefinitionVersionRef{{DefinitionID: "int", VersionID: 1}}}
+ require.NoError(t, json.Unmarshal([]byte(`{"timestamp-ms":2}`),
&entry))
+ assert.Empty(t, entry.DefinitionVersions)
+ })
+}
+
func TestOnNullInputValues(t *testing.T) {
for _, behavior := range []OnNullInput{OnNullInputCall,
OnNullInputReturnNull} {
m := minimalMetadata(t)
