tanmayrauth commented on code in PR #1418: URL: https://github.com/apache/iceberg-go/pull/1418#discussion_r3530600040
########## udf/types.go: ########## @@ -0,0 +1,360 @@ +// 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 udf + +import ( + "encoding/json" + "errors" + "fmt" + "strings" + "unicode" + + "github.com/apache/iceberg-go" +) + +// ErrInvalidUDFType is returned when a UDF parameter or return type +// does not conform to the UDF spec's Types section. +var ErrInvalidUDFType = errors.New("invalid UDF type") + +// Type represents a UDF parameter or return type as defined by the UDF +// spec. UDF types are Iceberg types without field IDs: primitives are +// encoded as plain strings (e.g. "int", "decimal(9,2)") and nested types +// (list, map, struct) as JSON objects carrying only the fields the UDF +// spec requires. +// +// String returns the canonical string representation used to build +// definition IDs (see CanonicalDefinitionID). +type Type interface { + fmt.Stringer + + Equals(Type) bool + + // canonicalTo writes the canonical string form, sealing the + // interface to the implementations in this package. + canonicalTo(sb *strings.Builder) +} + +// PrimitiveType is a primitive or semi-structured UDF type, stored as the +// verbatim type string (e.g. "int", "decimal(9,2)", "variant"). Values are +// always validated: construct them with PrimitiveTypeOf. +type PrimitiveType struct { + name string +} + +// PrimitiveTypeOf validates s against the Iceberg type system and the UDF +// spec's rules for type strings (no spaces or quote characters) and returns +// the corresponding PrimitiveType. +func PrimitiveTypeOf(s string) (PrimitiveType, error) { + if s == "" { + return PrimitiveType{}, fmt.Errorf("%w: type string must not be empty", ErrInvalidUDFType) + } + + if strings.ContainsFunc(s, unicode.IsSpace) || strings.ContainsAny(s, `"'`) { Review Comment: This rejects any type string with a space, but the canonical Iceberg string for decimal has one — `decimal(P, S)`. That's what the table spec, Java's DecimalType.toString(), PyIceberg, and iceberg-go's own DecimalType.String()/parser all emit and normalize to. Two consequences I confirmed locally: 1. Reading UDF metadata whose param/return type is `decimal(9, 2)` (i.e. anything a conformant engine writes) fails outright — PrimitiveTypeOf rejects the spaced form, and it propagates to nested types like `list<decimal(9, 2)>` too. 2. Because the definition-id is derived from the verbatim stored string, an iceberg-go author writing `decimal(9,2)` produces a different definition-id than Java/Py (`decimal(9, 2)`) for the same overload, so they can't be correlated. Since PrimitiveTypeOf already validates by round-tripping through iceberg's NestedField parser, could it derive the stored canonical string from the *parsed* type (one normalization) instead of storing the raw input and rejecting whitespace? That way `decimal(9, 2)` and `decimal(9,2)` collapse to one canonical form and reads succeed. Worth pinning down what the spec declares the canonical definition-id form to be for parameterized types like decimal — but hard-rejecting the spaced form on the read path breaks compat either way. ########## udf/types.go: ########## @@ -0,0 +1,360 @@ +// 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 udf + +import ( + "encoding/json" + "errors" + "fmt" + "strings" + "unicode" + + "github.com/apache/iceberg-go" +) + +// ErrInvalidUDFType is returned when a UDF parameter or return type +// does not conform to the UDF spec's Types section. +var ErrInvalidUDFType = errors.New("invalid UDF type") + +// Type represents a UDF parameter or return type as defined by the UDF +// spec. UDF types are Iceberg types without field IDs: primitives are +// encoded as plain strings (e.g. "int", "decimal(9,2)") and nested types +// (list, map, struct) as JSON objects carrying only the fields the UDF +// spec requires. +// +// String returns the canonical string representation used to build +// definition IDs (see CanonicalDefinitionID). +type Type interface { + fmt.Stringer + + Equals(Type) bool + + // canonicalTo writes the canonical string form, sealing the + // interface to the implementations in this package. + canonicalTo(sb *strings.Builder) +} + +// PrimitiveType is a primitive or semi-structured UDF type, stored as the +// verbatim type string (e.g. "int", "decimal(9,2)", "variant"). Values are +// always validated: construct them with PrimitiveTypeOf. +type PrimitiveType struct { + name string +} + +// PrimitiveTypeOf validates s against the Iceberg type system and the UDF +// spec's rules for type strings (no spaces or quote characters) and returns +// the corresponding PrimitiveType. +func PrimitiveTypeOf(s string) (PrimitiveType, error) { + if s == "" { + return PrimitiveType{}, fmt.Errorf("%w: type string must not be empty", ErrInvalidUDFType) + } + + if strings.ContainsFunc(s, unicode.IsSpace) || strings.ContainsAny(s, `"'`) { + return PrimitiveType{}, fmt.Errorf("%w: type string %q must not contain spaces or quote characters", + ErrInvalidUDFType, s) + } + + // Validate through the iceberg type parser so the accepted set of + // primitive type strings always matches the rest of the module. + // NestedField is the exported entry point to that parser. + payload, err := json.Marshal(map[string]any{"id": 1, "name": "f", "required": false, "type": s}) + if err != nil { + return PrimitiveType{}, err + } + var field iceberg.NestedField + if err := json.Unmarshal(payload, &field); err != nil { + return PrimitiveType{}, fmt.Errorf("%w: %q is not a valid Iceberg type string: %w", ErrInvalidUDFType, s, err) + } + + return PrimitiveType{name: s}, nil +} + +func (p PrimitiveType) String() string { return p.name } + +func (p PrimitiveType) Equals(other Type) bool { + o, ok := other.(PrimitiveType) + + return ok && p.name == o.name +} + +func (p PrimitiveType) canonicalTo(sb *strings.Builder) { sb.WriteString(p.name) } + +func (p PrimitiveType) MarshalJSON() ([]byte, error) { return json.Marshal(p.name) } + +// ListType is a UDF list type carrying only its element type. +type ListType struct { + Element Type +} + +func (l ListType) String() string { return canonicalString(l) } + +func (l ListType) Equals(other Type) bool { + o, ok := other.(ListType) + + return ok && l.Element.Equals(o.Element) +} + +func (l ListType) canonicalTo(sb *strings.Builder) { + sb.WriteString("list<") + l.Element.canonicalTo(sb) + sb.WriteString(">") +} + +func (l ListType) MarshalJSON() ([]byte, error) { + return json.Marshal(struct { + Type string `json:"type"` + Element Type `json:"element"` + }{Type: "list", Element: l.Element}) +} + +// MapType is a UDF map type carrying only its key and value types. +type MapType struct { + Key Type + Value Type +} + +func (m MapType) String() string { return canonicalString(m) } + +func (m MapType) Equals(other Type) bool { + o, ok := other.(MapType) + + return ok && m.Key.Equals(o.Key) && m.Value.Equals(o.Value) +} + +func (m MapType) canonicalTo(sb *strings.Builder) { + sb.WriteString("map<") + m.Key.canonicalTo(sb) + sb.WriteString(",") + m.Value.canonicalTo(sb) + sb.WriteString(">") +} + +func (m MapType) MarshalJSON() ([]byte, error) { + return json.Marshal(struct { + Type string `json:"type"` + Key Type `json:"key"` + Value Type `json:"value"` + }{Type: "map", Key: m.Key, Value: m.Value}) +} + +// StructField is a single field of a UDF struct type, carrying only a name +// and a type. +type StructField struct { + Name string + Type Type +} + +// StructType is a UDF struct type carrying only its fields. +type StructType struct { + Fields []StructField +} + +func (s StructType) String() string { return canonicalString(s) } + +func (s StructType) Equals(other Type) bool { + o, ok := other.(StructType) + if !ok || len(s.Fields) != len(o.Fields) { + return false + } + for i, f := range s.Fields { + if f.Name != o.Fields[i].Name || !f.Type.Equals(o.Fields[i].Type) { + return false + } + } + + return true +} + +func (s StructType) canonicalTo(sb *strings.Builder) { + sb.WriteString("struct<") + for i, f := range s.Fields { + if i > 0 { + sb.WriteString(",") + } + sb.WriteString(f.Name) + sb.WriteString(":") + f.Type.canonicalTo(sb) + } + sb.WriteString(">") +} + +func (s StructType) MarshalJSON() ([]byte, error) { + fields := make([]struct { + Name string `json:"name"` + Type Type `json:"type"` + }, len(s.Fields)) + for i, f := range s.Fields { + fields[i].Name, fields[i].Type = f.Name, f.Type + } + + return json.Marshal(struct { + Type string `json:"type"` + Fields any `json:"fields"` + }{Type: "struct", Fields: fields}) +} + +func canonicalString(t Type) string { + var sb strings.Builder + t.canonicalTo(&sb) + + return sb.String() +} + +// validateType checks that t and all nested types are fully specified. +// PrimitiveType values are valid by construction; composite types built as +// literals may carry nil children, which this rejects. +func validateType(t Type) error { + switch v := t.(type) { + case nil: + return fmt.Errorf("%w: type must not be nil", ErrInvalidUDFType) + case PrimitiveType: + if v.name == "" { + return fmt.Errorf("%w: uninitialized primitive type; use PrimitiveTypeOf", ErrInvalidUDFType) + } + + return nil + case ListType: + return validateType(v.Element) + case MapType: + if err := validateType(v.Key); err != nil { + return err + } + + return validateType(v.Value) + case StructType: Review Comment: Struct field names are checked for non-emptiness but not uniqueness — `struct<a:int,a:string>` validates and round-trips here. Field names should be unique among siblings; worth rejecting duplicates, mirroring the duplicate-parameter-name check you already have at the definition level. ########## udf/metadata_builder.go: ########## @@ -0,0 +1,561 @@ +// 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 udf + +import ( + "errors" + "fmt" + "maps" + "slices" + "strings" + "time" + + "github.com/apache/iceberg-go" + + "github.com/google/uuid" +) + +// DefinitionOpt configures optional fields of a new Definition. +type DefinitionOpt func(*Definition) + +// WithSpecificName sets the definition's optional specific name, a stable +// user-assignable handle that must be unique among all definitions. +func WithSpecificName(name string) DefinitionOpt { + return func(d *Definition) { + d.SpecificName = name + } +} + +// WithDefinitionDoc sets the definition's documentation string. +func WithDefinitionDoc(doc string) DefinitionOpt { + return func(d *Definition) { + d.Doc = doc + } +} + +// WithReturnNullable sets the definition's return-nullable hint. +func WithReturnNullable(nullable bool) DefinitionOpt { + return func(d *Definition) { + d.ReturnNullable = &nullable + } +} + +// NewDefinition creates a definition for the given signature, deriving its +// definition-id from the parameter types. The definition starts with no +// versions; add them with MetadataBuilder.AddDefinitionVersion. +func NewDefinition(functionType FunctionType, params []Parameter, returnType Type, opts ...DefinitionOpt) (*Definition, error) { + definitionID, err := CanonicalDefinitionID(params) + if err != nil { + return nil, err + } + + // Keep parameters non-nil so a zero-parameter definition serializes as + // the required empty list rather than null. + parameters := slices.Clone(params) + if parameters == nil { + parameters = []Parameter{} + } + + def := &Definition{ + DefinitionID: definitionID, + Parameters: parameters, + ReturnType: returnType, + CurrentVersionID: -1, + FunctionType: functionType, + } + + for _, opt := range opts { + opt(def) + } + + return def, nil +} + +// DefinitionVersionOpt configures optional fields of a new DefinitionVersion. +type DefinitionVersionOpt func(*DefinitionVersion) + +// WithTimestampMS overrides the version's creation timestamp. +func WithTimestampMS(timestampMS int64) DefinitionVersionOpt { + return func(v *DefinitionVersion) { + v.TimestampMS = timestampMS + } +} + +// WithDeterministic marks whether the function version is deterministic. +func WithDeterministic(deterministic bool) DefinitionVersionOpt { + return func(v *DefinitionVersion) { + v.Deterministic = deterministic + } +} + +// WithOnNullInput sets the version's behavior for NULL input parameters. +func WithOnNullInput(behavior OnNullInput) DefinitionVersionOpt { + return func(v *DefinitionVersion) { + v.OnNullInput = behavior + } +} + +// NewDefinitionVersion creates a definition version from the given +// representations. The version id is assigned when the version is added to +// a definition via MetadataBuilder.AddDefinitionVersion. +// +// NewDefinitionVersion seeds TimestampMS to time.Now().UnixMilli(); use +// WithTimestampMS to override. +func NewDefinitionVersion(representations []Representation, opts ...DefinitionVersionOpt) (*DefinitionVersion, error) { + if len(representations) == 0 { + return nil, fmt.Errorf("%w: a definition version must have at least one representation", ErrInvalidUDFMetadata) + } + for _, repr := range representations { + if repr == nil { + return nil, fmt.Errorf("%w: representations must not contain nil entries", ErrInvalidUDFMetadata) + } + } + + version := &DefinitionVersion{ + Representations: slices.Clone(representations), + TimestampMS: time.Now().UnixMilli(), + } + + for _, opt := range opts { + opt(version) + } + + return version, nil +} + +// MetadataBuilder builds UDF metadata, either from scratch or on top of an +// existing metadata instance. Because metadata files are immutable, every +// change produces a complete new metadata object; when any definition's +// selected version changes, Build appends a definition-log entry capturing +// the selected version of every definition. +type MetadataBuilder struct { + base Metadata + + uuid uuid.UUID + loc string + props iceberg.Properties + secure bool + doc string + definitions []*Definition + definitionLog []DefinitionLogEntry + + definitionsByID map[string]*Definition + + logTimestampMS int64 + selectionChanged bool + hasChanges bool + + // error tracking for build chaining + // if set, subsequent operations become a noop + err error +} + +// NewMetadataBuilder creates a builder for a new UDF's metadata. +func NewMetadataBuilder() (*MetadataBuilder, error) { + return &MetadataBuilder{ + props: iceberg.Properties{}, + definitions: make([]*Definition, 0), + definitionLog: make([]DefinitionLogEntry, 0), + definitionsByID: make(map[string]*Definition), + }, nil +} + +// MetadataBuilderFromBase creates a builder seeded with the state of an +// existing metadata instance. +func MetadataBuilderFromBase(base Metadata) (*MetadataBuilder, error) { + if base == nil { + return nil, fmt.Errorf("%w: cannot create metadata builder from nil base", ErrInvalidUDFMetadata) + } + + b := &MetadataBuilder{ + base: base, + uuid: base.FunctionUUID(), + loc: base.Location(), + props: maps.Clone(base.Properties()), + secure: base.Secure(), + doc: base.Doc(), + definitions: cloneSlice(base.Definitions()), + definitionLog: cloneDefinitionLog(base.DefinitionLog()), + } + if b.props == nil { + b.props = iceberg.Properties{} + } + b.definitionsByID = indexBy(b.definitions, func(d *Definition) string { return d.DefinitionID }) + + return b, nil +} + +// HasChanges reports whether any builder operation changed the state +// relative to the base metadata. +func (b *MetadataBuilder) HasChanges() bool { return b.hasChanges } + +// Err returns the first error encountered by builder operations, if any. +func (b *MetadataBuilder) Err() error { return b.err } + +// SetUUID assigns the function's UUID. The UUID is generated once at +// creation and cannot be reassigned. +func (b *MetadataBuilder) SetUUID(newUUID uuid.UUID) *MetadataBuilder { + if b.err != nil { + return b + } + + if newUUID == uuid.Nil { + b.err = errors.New("cannot set uuid to null") + + return b + } + + if b.uuid == newUUID { + return b + } + + if b.uuid != uuid.Nil { + b.err = errors.New("cannot reassign uuid") + + return b + } + + b.uuid = newUUID + b.hasChanges = true + + return b +} + +// SetLoc sets the function's base location, used to create metadata +// file locations. +func (b *MetadataBuilder) SetLoc(loc string) *MetadataBuilder { + if b.err != nil || b.loc == loc { + return b + } + + b.loc = loc + b.hasChanges = true + + return b +} + +// SetProperties adds or updates the given properties. +func (b *MetadataBuilder) SetProperties(props iceberg.Properties) *MetadataBuilder { + if b.err != nil || len(props) == 0 { + return b + } + + maps.Copy(b.props, props) + b.hasChanges = true + + return b +} + +// RemoveProperties removes the given property keys. +func (b *MetadataBuilder) RemoveProperties(keys []string) *MetadataBuilder { + if b.err != nil || len(keys) == 0 { + return b + } + + for _, key := range keys { + delete(b.props, key) + } + b.hasChanges = true + + return b +} + +// SetSecure marks whether this is a secure function. +func (b *MetadataBuilder) SetSecure(secure bool) *MetadataBuilder { + if b.err != nil || b.secure == secure { + return b + } + + b.secure = secure + b.hasChanges = true + + return b +} + +// SetDoc sets the function's documentation string. +func (b *MetadataBuilder) SetDoc(doc string) *MetadataBuilder { + if b.err != nil || b.doc == doc { + return b + } + + b.doc = doc + b.hasChanges = true + + return b +} + +// SetLogTimestampMS overrides the timestamp used for the definition-log +// entry appended by Build. Writers that need deterministic metadata (or a +// timestamp consistent with an external commit) should set this; it +// defaults to time.Now().UnixMilli(). +func (b *MetadataBuilder) SetLogTimestampMS(timestampMS int64) *MetadataBuilder { + if b.err != nil { + return b + } + + b.logTimestampMS = timestampMS + + return b +} + +// AddDefinition adds a definition for a new signature. The definition's +// definition-id is derived from its parameters; if the definition already +// carries an id it must match the canonical form. Adding a definition +// whose signature or specific name already exists fails: there can be +// only one definition for a given signature. +func (b *MetadataBuilder) AddDefinition(def *Definition) *MetadataBuilder { + if b.err != nil { + return b + } + + if def == nil { + b.err = fmt.Errorf("%w: cannot add nil definition", ErrInvalidUDFMetadata) + + return b + } + + canonicalID, err := CanonicalDefinitionID(def.Parameters) + if b.setErr(err) { + return b + } + + if def.DefinitionID != "" && def.DefinitionID != canonicalID { + b.err = fmt.Errorf("%w: definition-id %q does not match canonical form %q of its parameters", + ErrInvalidUDFMetadata, def.DefinitionID, canonicalID) + + return b + } + + if _, ok := b.definitionsByID[canonicalID]; ok { + b.err = fmt.Errorf("%w: a definition for signature %q already exists", ErrInvalidUDFMetadata, canonicalID) + + return b + } + + if def.SpecificName != "" { + for _, existing := range b.definitions { + if existing.SpecificName == def.SpecificName { + b.err = fmt.Errorf("%w: a definition with specific-name %q already exists", + ErrInvalidUDFMetadata, def.SpecificName) + + return b + } + } + } + + added := def.Clone() + added.DefinitionID = canonicalID + // Keep parameters non-nil so a zero-parameter definition serializes as + // the required empty list rather than null. + if added.Parameters == nil { + added.Parameters = []Parameter{} + } + b.definitions = append(b.definitions, added) + b.definitionsByID[canonicalID] = added + b.hasChanges = true + b.selectionChanged = true + + return b +} + +// RemoveDefinition removes the definition with the given definition-id. +func (b *MetadataBuilder) RemoveDefinition(definitionID string) *MetadataBuilder { + if b.err != nil { + return b + } + + if _, ok := b.definitionsByID[definitionID]; !ok { + b.err = fmt.Errorf("%w: no definition with definition-id %q", ErrInvalidUDFMetadata, definitionID) + + return b + } + + delete(b.definitionsByID, definitionID) + b.definitions = slices.DeleteFunc(b.definitions, func(d *Definition) bool { + return d.DefinitionID == definitionID + }) + b.hasChanges = true + b.selectionChanged = true + + return b +} + +// AddDefinitionVersion adds a version to the definition with the given +// definition-id and selects it as the definition's current version. A +// version id of 0 is assigned the next id for the definition; an explicit +// version id must not already exist. Use SetCurrentVersion to roll back to +// an earlier version. +func (b *MetadataBuilder) AddDefinitionVersion(definitionID string, version *DefinitionVersion) *MetadataBuilder { + if b.err != nil { + return b + } + + if version == nil { + b.err = fmt.Errorf("%w: cannot add nil definition version", ErrInvalidUDFMetadata) + + return b + } + + def, ok := b.definitionsByID[definitionID] + if !ok { + b.err = fmt.Errorf("%w: no definition with definition-id %q", ErrInvalidUDFMetadata, definitionID) + + return b + } + + added := version.Clone() + if added.VersionID < 0 { + b.err = fmt.Errorf("%w: invalid negative version-id %d", ErrInvalidUDFMetadata, added.VersionID) + + return b + } + if added.VersionID == 0 { + added.VersionID = nextVersionID(def) + } else if def.Version(added.VersionID) != nil { + b.err = fmt.Errorf("%w: definition %q already has version-id %d", + ErrInvalidUDFMetadata, definitionID, added.VersionID) + + return b + } + + def.Versions = append(def.Versions, added) + def.CurrentVersionID = added.VersionID + b.hasChanges = true + b.selectionChanged = true + + return b +} + +// SetCurrentVersion sets the definition's current version to an existing +// version id, e.g. to roll back to an earlier implementation. +func (b *MetadataBuilder) SetCurrentVersion(definitionID string, versionID int) *MetadataBuilder { + if b.err != nil { + return b + } + + def, ok := b.definitionsByID[definitionID] + if !ok { + b.err = fmt.Errorf("%w: no definition with definition-id %q", ErrInvalidUDFMetadata, definitionID) + + return b + } + + if def.CurrentVersionID == versionID { + return b + } + + if def.Version(versionID) == nil { + b.err = fmt.Errorf("%w: definition %q has no version-id %d", ErrInvalidUDFMetadata, definitionID, versionID) + + return b + } + + def.CurrentVersionID = versionID + b.hasChanges = true + b.selectionChanged = true + + return b +} + +// Build validates and returns the metadata. When any definition's selected +// version changed, a definition-log entry capturing the current selection +// of every definition is appended, with entries ordered by definition-id. +func (b *MetadataBuilder) Build() (Metadata, error) { + if b.err != nil { + return nil, b.err + } + + uuid_ := b.uuid + if uuid_ == uuid.Nil { + uuid_ = uuid.New() + } + + definitionLog := cloneDefinitionLog(b.definitionLog) + if b.selectionChanged { + timestampMS := b.logTimestampMS + if timestampMS == 0 { Review Comment: `Build` treats a log timestamp of 0 as "unset" and substitutes time.Now(), so a writer can't set a deterministic epoch-0 log timestamp. DefinitionVersion.TimestampMS uses -1 as its sentinel and accepts 0 — could the log timestamp use the same -1 convention so the two agree? Minor. ########## udf/metadata.go: ########## @@ -0,0 +1,887 @@ +// 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 udf provides the metadata model for Iceberg SQL UDFs as +// specified by the Iceberg UDF spec. +// https://iceberg.apache.org/udf-spec/ +package udf + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "io" + "maps" + "slices" + "strings" + "sync" + + "github.com/apache/iceberg-go" + + "github.com/google/uuid" +) + +var ( + ErrInvalidUDFMetadata = errors.New("invalid UDF metadata") + ErrInvalidUDFMetadataFormatVersion = errors.New("invalid or missing format-version in UDF metadata") +) + +const ( + SupportedUDFFormatVersion = 1 + + initialVersionID = 1 +) + +// FunctionType indicates whether a definition is a scalar function (udf) +// or a table function (udtf). +type FunctionType string + +const ( + FunctionTypeUDF FunctionType = "udf" + FunctionTypeUDTF FunctionType = "udtf" +) + +// OnNullInput defines how a UDF behaves when any input parameter is NULL. +type OnNullInput string + +const ( + // OnNullInputCall means the function may handle NULLs internally, so + // engines must execute it even if some inputs are NULL. This is the + // default behavior when on-null-input is absent. + OnNullInputCall OnNullInput = "call" + // OnNullInputReturnNull means the function always returns NULL if any + // input argument is NULL, allowing engines to skip evaluation. + OnNullInputReturnNull OnNullInput = "return-null" +) + +// Representation is a single implementation of a definition version. The +// only representation type defined by the UDF spec is SQL; unrecognized +// types round-trip through UnknownRepresentation. +type Representation interface { + // RepresentationType returns the representation's type discriminator, + // e.g. "sql". + RepresentationType() string + + // isRepresentation seals the interface to the implementations in + // this package. + isRepresentation() +} + +// SQLRepresentation stores the function body as a SQL expression in a +// specific dialect. The SQL must reference parameters using the names +// declared in the definition's parameters. +type SQLRepresentation struct { + Dialect string `json:"dialect"` + SQL string `json:"sql"` +} + +// NewSQLRepresentation creates a SQL representation from a dialect +// identifier (e.g. "spark", "trino") and a SQL expression. +func NewSQLRepresentation(dialect, sql string) (SQLRepresentation, error) { + if dialect == "" { + return SQLRepresentation{}, fmt.Errorf("%w: sql representation requires a dialect", ErrInvalidUDFMetadata) + } + if sql == "" { + return SQLRepresentation{}, fmt.Errorf("%w: sql representation requires a sql expression", ErrInvalidUDFMetadata) + } + + return SQLRepresentation{Dialect: dialect, SQL: sql}, nil +} + +func (SQLRepresentation) RepresentationType() string { return "sql" } +func (SQLRepresentation) isRepresentation() {} + +func (r SQLRepresentation) MarshalJSON() ([]byte, error) { + return json.Marshal(struct { + Type string `json:"type"` + Dialect string `json:"dialect"` + SQL string `json:"sql"` + }{Type: "sql", Dialect: r.Dialect, SQL: r.SQL}) +} + +// UnknownRepresentation preserves a representation of an unrecognized type +// so that metadata written by newer or extended writers round-trips intact. +type UnknownRepresentation struct { + TypeName string + + raw json.RawMessage +} + +// Raw returns the representation's original JSON. +func (r UnknownRepresentation) Raw() json.RawMessage { return r.raw } + +func (r UnknownRepresentation) RepresentationType() string { return r.TypeName } +func (UnknownRepresentation) isRepresentation() {} + +func (r UnknownRepresentation) MarshalJSON() ([]byte, error) { + return slices.Clone(r.raw), nil +} + +func representationsEqual(a, b Representation) bool { + switch av := a.(type) { + case SQLRepresentation: + bv, ok := b.(SQLRepresentation) + + return ok && av == bv + case UnknownRepresentation: + bv, ok := b.(UnknownRepresentation) + + return ok && av.TypeName == bv.TypeName && bytes.Equal(av.raw, bv.raw) + default: + return false + } +} + +func unmarshalRepresentation(b json.RawMessage) (Representation, error) { + aux := struct { + Type string `json:"type"` + }{} + if err := json.Unmarshal(b, &aux); err != nil { + return nil, err + } + if aux.Type == "" { + return nil, fmt.Errorf("%w: representation requires a type", ErrInvalidUDFMetadata) + } + + if aux.Type == "sql" { + var r SQLRepresentation + if err := json.Unmarshal(b, &r); err != nil { + return nil, err + } + + return r, nil + } + + // Store the raw JSON compacted so that round-tripped representations + // compare equal: encoding/json compacts MarshalJSON output. + var compacted bytes.Buffer + if err := json.Compact(&compacted, b); err != nil { + return nil, err + } + + return UnknownRepresentation{TypeName: aux.Type, raw: compacted.Bytes()}, nil +} + +// Parameter is a single function parameter. +type Parameter struct { + Name string `json:"name"` + Type Type `json:"type"` + Doc string `json:"doc,omitempty"` +} + +func (p *Parameter) UnmarshalJSON(b []byte) error { + type Alias Parameter + aux := struct { + Type json.RawMessage `json:"type"` + *Alias + }{Alias: (*Alias)(p)} + + 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) + } + + typ, err := unmarshalType(aux.Type) + if err != nil { + return err + } + p.Type = typ + + return nil +} + +func (p Parameter) equals(other Parameter) bool { + return p.Name == other.Name && p.Doc == other.Doc && + p.Type != nil && other.Type != nil && p.Type.Equals(other.Type) +} + +// DefinitionVersion is a specific implementation of a definition at a +// given point in time. Versions are immutable: changes to a definition +// introduce a new version. +type DefinitionVersion struct { + // VersionID is a monotonically increasing identifier of the + // definition version. + VersionID int `json:"version-id"` + // Representations lists the UDF implementations of this version. + Representations []Representation `json:"representations"` + // Deterministic indicates whether the function is deterministic. + // Defaults to false. + Deterministic bool `json:"deterministic,omitempty"` + // OnNullInput defines how the UDF behaves when any input parameter + // is NULL. Defaults to OnNullInputCall when empty. + OnNullInput OnNullInput `json:"on-null-input,omitempty"` + // TimestampMS is the creation timestamp of this version (ms from epoch). + TimestampMS int64 `json:"timestamp-ms"` +} + +func (v *DefinitionVersion) UnmarshalJSON(b []byte) error { + type Alias DefinitionVersion + aux := struct { + Representations []json.RawMessage `json:"representations"` + *Alias + }{Alias: (*Alias)(v)} + + v.VersionID = -1 + v.TimestampMS = -1 + + if err := json.Unmarshal(b, &aux); err != nil { + return err + } + + v.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 + } + + return nil +} + +// NullInputBehavior returns the effective on-null-input behavior, +// applying the spec default (OnNullInputCall) when the field is absent. +func (v *DefinitionVersion) NullInputBehavior() OnNullInput { + if v.OnNullInput == "" { + return OnNullInputCall + } + + return v.OnNullInput +} + +// Equals compares two versions for semantic equality: the default +// on-null-input behavior compares equal to an explicit "call". +func (v *DefinitionVersion) Equals(other *DefinitionVersion) bool { + return v.VersionID == other.VersionID && + v.Deterministic == other.Deterministic && + v.NullInputBehavior() == other.NullInputBehavior() && + v.TimestampMS == other.TimestampMS && + slices.EqualFunc(v.Representations, other.Representations, representationsEqual) +} + +func (v *DefinitionVersion) Clone() *DefinitionVersion { + if v == nil { + return nil + } + + cloned := *v + cloned.Representations = slices.Clone(v.Representations) + + return &cloned +} + +func (v *DefinitionVersion) validate(definitionID string) error { + if v.VersionID == -1 { + return fmt.Errorf("%w: definition %q has a version missing version-id", + ErrInvalidUDFMetadata, definitionID) + } + if v.VersionID < 0 { + return fmt.Errorf("%w: definition %q has invalid negative version-id %d", + ErrInvalidUDFMetadata, definitionID, v.VersionID) + } + + if v.TimestampMS == -1 { + return fmt.Errorf("%w: definition %q version %d is missing timestamp-ms", + ErrInvalidUDFMetadata, definitionID, v.VersionID) + } + if v.TimestampMS < 0 { + return fmt.Errorf("%w: definition %q version %d has invalid negative timestamp-ms %d", + ErrInvalidUDFMetadata, definitionID, v.VersionID, v.TimestampMS) + } + + if len(v.Representations) == 0 { + return fmt.Errorf("%w: definition %q version %d must have at least one representation", + ErrInvalidUDFMetadata, definitionID, v.VersionID) + } + + switch v.OnNullInput { + case "", OnNullInputCall, OnNullInputReturnNull: + default: + return fmt.Errorf("%w: definition %q version %d has invalid on-null-input %q", + ErrInvalidUDFMetadata, definitionID, v.VersionID, v.OnNullInput) + } + + seenDialects := make(map[string]bool) + for _, repr := range v.Representations { + if repr == nil { + return fmt.Errorf("%w: definition %q version %d representations must not contain null entries", + ErrInvalidUDFMetadata, definitionID, v.VersionID) + } + if unknown, ok := repr.(UnknownRepresentation); ok && len(unknown.raw) == 0 { + return fmt.Errorf("%w: definition %q version %d has an unknown representation without raw JSON; unknown representations must originate from parsed metadata", + ErrInvalidUDFMetadata, definitionID, v.VersionID) + } + sqlRepr, ok := repr.(SQLRepresentation) + if !ok { + continue + } + if sqlRepr.Dialect == "" { + return fmt.Errorf("%w: definition %q version %d has a sql representation without a dialect", + ErrInvalidUDFMetadata, definitionID, v.VersionID) + } + if sqlRepr.SQL == "" { + return fmt.Errorf("%w: definition %q version %d has a sql representation without a sql expression", + ErrInvalidUDFMetadata, definitionID, v.VersionID) + } + dialect := strings.ToLower(sqlRepr.Dialect) + if seenDialects[dialect] { + return fmt.Errorf("%w: definition %q version %d has duplicate sql dialect %s", + ErrInvalidUDFMetadata, definitionID, v.VersionID, sqlRepr.Dialect) + } + seenDialects[dialect] = true + } + + return nil +} + +// Definition represents one function signature (e.g. add_one(int) vs +// add_one(float)). A definition is uniquely identified by its signature: +// the ordered list of parameter types, canonicalized as its DefinitionID. +type Definition struct { + // DefinitionID is the canonical string derived from the parameter + // types (see CanonicalDefinitionID). + DefinitionID string `json:"definition-id"` + // SpecificName is an optional user-assignable name for this + // definition, unique among all definitions of the function. + SpecificName string `json:"specific-name,omitempty"` + // Parameters is the ordered list of function parameters. Invocation + // order must match this list. + Parameters []Parameter `json:"parameters"` + // ReturnType is the declared return type. All versions must produce + // values of this type. + ReturnType Type `json:"return-type"` + // ReturnNullable hints whether the return value is nullable. + // Defaults to true when absent. + ReturnNullable *bool `json:"return-nullable,omitempty"` + // Versions lists the versioned implementations of this definition. + Versions []*DefinitionVersion `json:"versions"` + // CurrentVersionID identifies the current version of this definition. + CurrentVersionID int `json:"current-version-id"` + // FunctionType is "udf" for scalar functions or "udtf" for table + // functions. For "udtf", ReturnType must be a struct describing the + // output schema. + FunctionType FunctionType `json:"function-type"` + // Doc is an optional documentation string. + Doc string `json:"doc,omitempty"` +} + +func (d *Definition) UnmarshalJSON(b []byte) error { + type Alias Definition + // 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 + // parameters. + aux := struct { + DefinitionID *string `json:"definition-id"` + Parameters json.RawMessage `json:"parameters"` + ReturnType json.RawMessage `json:"return-type"` + *Alias + }{Alias: (*Alias)(d)} + + d.CurrentVersionID = -1 + + if err := json.Unmarshal(b, &aux); err != nil { + return err + } + + if aux.DefinitionID == nil { + return fmt.Errorf("%w: definition is missing definition-id", ErrInvalidUDFMetadata) + } + d.DefinitionID = *aux.DefinitionID + + if len(aux.Parameters) == 0 { + return fmt.Errorf("%w: definition %q is missing parameters", ErrInvalidUDFMetadata, d.DefinitionID) + } + if err := json.Unmarshal(aux.Parameters, &d.Parameters); err != nil { + return err + } + if d.Parameters == nil { + return fmt.Errorf("%w: definition %q is missing parameters", ErrInvalidUDFMetadata, d.DefinitionID) + } + + if len(aux.ReturnType) == 0 { + return fmt.Errorf("%w: definition %q requires a return-type", ErrInvalidUDFMetadata, d.DefinitionID) + } + + typ, err := unmarshalType(aux.ReturnType) + if err != nil { + return err + } + d.ReturnType = typ + + return nil +} + +// ReturnsNullable returns the effective return-nullable hint, applying +// the spec default (true) when the field is absent. +func (d *Definition) ReturnsNullable() bool { + return d.ReturnNullable == nil || *d.ReturnNullable +} + +// CurrentVersion returns the definition version identified by +// CurrentVersionID, or nil if no such version exists. +func (d *Definition) CurrentVersion() *DefinitionVersion { + return d.Version(d.CurrentVersionID) +} + +// Version returns the definition version with the given id, or nil if no +// such version exists. +func (d *Definition) Version(versionID int) *DefinitionVersion { + for _, v := range d.Versions { + if v.VersionID == versionID { + return v + } + } + + return nil +} + +// Equals compares two definitions for semantic equality: absent optional +// fields compare equal to their spec defaults. +func (d *Definition) Equals(other *Definition) bool { + return d.DefinitionID == other.DefinitionID && + d.SpecificName == other.SpecificName && + slices.EqualFunc(d.Parameters, other.Parameters, Parameter.equals) && + d.ReturnType != nil && other.ReturnType != nil && d.ReturnType.Equals(other.ReturnType) && + d.ReturnsNullable() == other.ReturnsNullable() && + d.CurrentVersionID == other.CurrentVersionID && + d.FunctionType == other.FunctionType && + d.Doc == other.Doc && + slices.EqualFunc(d.Versions, other.Versions, (*DefinitionVersion).Equals) +} + +func (d *Definition) Clone() *Definition { + if d == nil { + return nil + } + + cloned := *d + // Type values are immutable, so sharing them across clones is safe. + cloned.Parameters = slices.Clone(d.Parameters) + cloned.Versions = cloneSlice(d.Versions) + if d.ReturnNullable != nil { + nullable := *d.ReturnNullable + cloned.ReturnNullable = &nullable + } + + return &cloned +} + +func (d *Definition) validate() error { + seenNames := make(map[string]bool) + for _, p := range d.Parameters { + if p.Name == "" { + return fmt.Errorf("%w: definition %q has a parameter without a name", + ErrInvalidUDFMetadata, d.DefinitionID) + } + if seenNames[p.Name] { + return fmt.Errorf("%w: definition %q has duplicate parameter name %q", + ErrInvalidUDFMetadata, d.DefinitionID, p.Name) + } + seenNames[p.Name] = true + if err := validateType(p.Type); err != nil { + return fmt.Errorf("%w: definition %q parameter %q: %w", + ErrInvalidUDFMetadata, d.DefinitionID, p.Name, err) + } + } + + canonicalID, err := CanonicalDefinitionID(d.Parameters) + if err != nil { + return fmt.Errorf("%w: definition %q: %w", ErrInvalidUDFMetadata, d.DefinitionID, err) + } + if d.DefinitionID != canonicalID { + return fmt.Errorf("%w: definition-id %q does not match canonical form %q of its parameters", + ErrInvalidUDFMetadata, d.DefinitionID, canonicalID) + } + + if err := validateType(d.ReturnType); err != nil { + return fmt.Errorf("%w: definition %q return-type: %w", ErrInvalidUDFMetadata, d.DefinitionID, err) + } + + switch d.FunctionType { + case FunctionTypeUDF: + case FunctionTypeUDTF: + if _, ok := d.ReturnType.(StructType); !ok { + return fmt.Errorf("%w: definition %q is a udtf but its return-type is not a struct", + ErrInvalidUDFMetadata, d.DefinitionID) + } + case "": + return fmt.Errorf("%w: definition %q is missing function-type", ErrInvalidUDFMetadata, d.DefinitionID) + default: + return fmt.Errorf("%w: definition %q has invalid function-type %q (must be %q or %q)", + ErrInvalidUDFMetadata, d.DefinitionID, d.FunctionType, FunctionTypeUDF, FunctionTypeUDTF) + } + + if len(d.Versions) == 0 { + return fmt.Errorf("%w: definition %q must have at least one version", ErrInvalidUDFMetadata, d.DefinitionID) + } + + seenVersions := make(map[int]bool) + for _, v := range d.Versions { + if v == nil { + return fmt.Errorf("%w: definition %q versions must not contain null entries", + ErrInvalidUDFMetadata, d.DefinitionID) + } + if err := v.validate(d.DefinitionID); err != nil { + return err + } + if seenVersions[v.VersionID] { + return fmt.Errorf("%w: definition %q has duplicate version-id %d", + ErrInvalidUDFMetadata, d.DefinitionID, v.VersionID) + } + seenVersions[v.VersionID] = true + } + + if d.CurrentVersionID == -1 { + return fmt.Errorf("%w: definition %q is missing current-version-id", ErrInvalidUDFMetadata, d.DefinitionID) + } + if d.CurrentVersionID < 0 { + return fmt.Errorf("%w: definition %q has invalid negative current-version-id %d", + ErrInvalidUDFMetadata, d.DefinitionID, d.CurrentVersionID) + } + if !seenVersions[d.CurrentVersionID] { + return fmt.Errorf("%w: definition %q current-version-id %d not found in versions", + ErrInvalidUDFMetadata, d.DefinitionID, d.CurrentVersionID) + } + + return nil +} + +// DefinitionVersionRef maps a definition to a selected version. +type DefinitionVersionRef struct { + // DefinitionID may be empty: a zero-parameter definition has the + // empty string as its definition-id. + DefinitionID string `json:"definition-id"` + VersionID int `json:"version-id"` +} + +func (r *DefinitionVersionRef) UnmarshalJSON(b []byte) error { + type Alias DefinitionVersionRef + // 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 + + if err := json.Unmarshal(b, &aux); err != nil { + return err + } + + if aux.DefinitionID == nil { + return fmt.Errorf("%w: definition-log reference is missing definition-id", ErrInvalidUDFMetadata) + } + r.DefinitionID = *aux.DefinitionID + + return nil +} + +// DefinitionLogEntry records the selected version of every definition at +// the time of a change, enabling rollback without external state. +type DefinitionLogEntry struct { + // TimestampMS is when the function was updated to use these + // definition versions (ms from epoch). + TimestampMS int64 `json:"timestamp-ms"` + // DefinitionVersions maps each definition to its selected version at + // this time. + DefinitionVersions []DefinitionVersionRef `json:"definition-versions"` +} + +func (e *DefinitionLogEntry) UnmarshalJSON(b []byte) error { + type Alias DefinitionLogEntry + aux := (*Alias)(e) + + e.TimestampMS = -1 + + return json.Unmarshal(b, aux) +} + +func (e DefinitionLogEntry) validate() error { + if e.TimestampMS == -1 { + return fmt.Errorf("%w: definition-log entry is missing timestamp-ms", ErrInvalidUDFMetadata) + } + if e.TimestampMS < 0 { + return fmt.Errorf("%w: definition-log entry has invalid negative timestamp-ms %d", + ErrInvalidUDFMetadata, e.TimestampMS) + } + + if len(e.DefinitionVersions) == 0 { + return fmt.Errorf("%w: definition-log entry at timestamp %d must have definition-versions", + ErrInvalidUDFMetadata, e.TimestampMS) + } + + seen := make(map[string]bool) + for _, ref := range e.DefinitionVersions { + if ref.VersionID == -1 { + return fmt.Errorf("%w: definition-log entry at timestamp %d has a reference to definition %q missing version-id", + ErrInvalidUDFMetadata, e.TimestampMS, ref.DefinitionID) + } + if ref.VersionID < 0 { + return fmt.Errorf("%w: definition-log entry at timestamp %d has invalid negative version-id %d for definition %q", + ErrInvalidUDFMetadata, e.TimestampMS, ref.VersionID, ref.DefinitionID) + } + if seen[ref.DefinitionID] { + return fmt.Errorf("%w: definition-log entry at timestamp %d references definition %q more than once", + ErrInvalidUDFMetadata, e.TimestampMS, ref.DefinitionID) + } + seen[ref.DefinitionID] = true + } + + return nil +} + +func (e DefinitionLogEntry) equals(other DefinitionLogEntry) bool { + return e.TimestampMS == other.TimestampMS && + slices.Equal(e.DefinitionVersions, other.DefinitionVersions) +} + +// Metadata is the self-contained metadata of an Iceberg SQL UDF as +// specified by the UDF spec. Metadata files are immutable: any change +// creates a new file, and catalogs atomically swap the file linked to a +// catalog identifier. UDF names are not part of the metadata; mapping +// names to metadata locations is the catalog's responsibility. +type Metadata interface { + // FormatVersion returns the UDF spec format version (1). + FormatVersion() int + // FunctionUUID returns the UUID identifying this UDF, generated once + // at creation. + FunctionUUID() uuid.UUID + // Location returns the function's base location, used to create + // metadata file locations. It may be empty. + Location() string + // Definitions returns the function's definitions, one per signature. + Definitions() []*Definition + // DefinitionByID returns the definition with the given definition-id. + DefinitionByID(definitionID string) (*Definition, bool) + // DefinitionLog returns the history of definition version selections. + DefinitionLog() []DefinitionLogEntry + // Properties returns the function's properties. Entries are treated + // as hints, not strict rules. + Properties() iceberg.Properties + // Secure reports whether this is a secure function whose sensitive + // information engines should not leak to end users. + Secure() bool + // Doc returns the function's documentation string. + Doc() string + + Equals(Metadata) bool +} + +// ParseMetadata parses UDF metadata JSON provided by the passed in +// reader, returning an error if one is encountered. +func ParseMetadata(r io.Reader) (Metadata, error) { + data, err := io.ReadAll(r) + if err != nil { + return nil, err + } + + return ParseMetadataBytes(data) +} + +// ParseMetadataString is like [ParseMetadata], but for a string rather +// than an io.Reader. +func ParseMetadataString(s string) (Metadata, error) { + return ParseMetadataBytes([]byte(s)) +} + +// ParseMetadataBytes is like [ParseMetadataString] but for a byte slice. +func ParseMetadataBytes(b []byte) (Metadata, error) { + var ret Metadata = &metadata{} + + return ret, json.Unmarshal(b, ret) +} + +type metadata struct { + UUID *uuid.UUID `json:"function-uuid"` + FormatVersionValue int `json:"format-version"` + DefinitionList []*Definition `json:"definitions"` + DefinitionLogList []DefinitionLogEntry `json:"definition-log"` + Loc string `json:"location,omitempty"` + Props iceberg.Properties `json:"properties,omitempty"` + SecureValue bool `json:"secure,omitempty"` + DocValue string `json:"doc,omitempty"` + + // cached lookup helpers, must be initialized in init() + lazyDefinitionsByID func() map[string]*Definition +} + +func (m *metadata) FormatVersion() int { return m.FormatVersionValue } +func (m *metadata) FunctionUUID() uuid.UUID { return *m.UUID } +func (m *metadata) Location() string { return m.Loc } +func (m *metadata) Definitions() []*Definition { return m.DefinitionList } +func (m *metadata) DefinitionLog() []DefinitionLogEntry { return m.DefinitionLogList } +func (m *metadata) Properties() iceberg.Properties { return m.Props } +func (m *metadata) Secure() bool { return m.SecureValue } +func (m *metadata) Doc() string { return m.DocValue } + +func (m *metadata) DefinitionByID(definitionID string) (*Definition, bool) { + def, ok := m.lazyDefinitionsByID()[definitionID] + + return def, ok +} + +func (m *metadata) Equals(other Metadata) bool { + if other == nil { Review Comment: This guards an untyped-nil interface, but a typed-nil Metadata (e.g. `(*metadata)(nil)`) slips through and panics at `other.FunctionUUID()` when it dereferences `*m.UUID`. Not reachable through the public API today, but given you're already nil-checking here it may be worth hardening. ########## udf/metadata.go: ########## @@ -0,0 +1,887 @@ +// 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 udf provides the metadata model for Iceberg SQL UDFs as +// specified by the Iceberg UDF spec. +// https://iceberg.apache.org/udf-spec/ +package udf + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "io" + "maps" + "slices" + "strings" + "sync" + + "github.com/apache/iceberg-go" + + "github.com/google/uuid" +) + +var ( + ErrInvalidUDFMetadata = errors.New("invalid UDF metadata") + ErrInvalidUDFMetadataFormatVersion = errors.New("invalid or missing format-version in UDF metadata") +) + +const ( + SupportedUDFFormatVersion = 1 + + initialVersionID = 1 +) + +// FunctionType indicates whether a definition is a scalar function (udf) +// or a table function (udtf). +type FunctionType string + +const ( + FunctionTypeUDF FunctionType = "udf" + FunctionTypeUDTF FunctionType = "udtf" +) + +// OnNullInput defines how a UDF behaves when any input parameter is NULL. +type OnNullInput string + +const ( + // OnNullInputCall means the function may handle NULLs internally, so + // engines must execute it even if some inputs are NULL. This is the + // default behavior when on-null-input is absent. + OnNullInputCall OnNullInput = "call" + // OnNullInputReturnNull means the function always returns NULL if any + // input argument is NULL, allowing engines to skip evaluation. + OnNullInputReturnNull OnNullInput = "return-null" +) + +// Representation is a single implementation of a definition version. The +// only representation type defined by the UDF spec is SQL; unrecognized +// types round-trip through UnknownRepresentation. +type Representation interface { + // RepresentationType returns the representation's type discriminator, + // e.g. "sql". + RepresentationType() string + + // isRepresentation seals the interface to the implementations in + // this package. + isRepresentation() +} + +// SQLRepresentation stores the function body as a SQL expression in a +// specific dialect. The SQL must reference parameters using the names +// declared in the definition's parameters. +type SQLRepresentation struct { + Dialect string `json:"dialect"` + SQL string `json:"sql"` +} + +// NewSQLRepresentation creates a SQL representation from a dialect +// identifier (e.g. "spark", "trino") and a SQL expression. +func NewSQLRepresentation(dialect, sql string) (SQLRepresentation, error) { + if dialect == "" { + return SQLRepresentation{}, fmt.Errorf("%w: sql representation requires a dialect", ErrInvalidUDFMetadata) + } + if sql == "" { + return SQLRepresentation{}, fmt.Errorf("%w: sql representation requires a sql expression", ErrInvalidUDFMetadata) + } + + return SQLRepresentation{Dialect: dialect, SQL: sql}, nil +} + +func (SQLRepresentation) RepresentationType() string { return "sql" } +func (SQLRepresentation) isRepresentation() {} + +func (r SQLRepresentation) MarshalJSON() ([]byte, error) { + return json.Marshal(struct { + Type string `json:"type"` + Dialect string `json:"dialect"` + SQL string `json:"sql"` + }{Type: "sql", Dialect: r.Dialect, SQL: r.SQL}) +} + +// UnknownRepresentation preserves a representation of an unrecognized type +// so that metadata written by newer or extended writers round-trips intact. +type UnknownRepresentation struct { + TypeName string + + raw json.RawMessage +} + +// Raw returns the representation's original JSON. +func (r UnknownRepresentation) Raw() json.RawMessage { return r.raw } + +func (r UnknownRepresentation) RepresentationType() string { return r.TypeName } +func (UnknownRepresentation) isRepresentation() {} + +func (r UnknownRepresentation) MarshalJSON() ([]byte, error) { + return slices.Clone(r.raw), nil +} + +func representationsEqual(a, b Representation) bool { + switch av := a.(type) { + case SQLRepresentation: + bv, ok := b.(SQLRepresentation) + + return ok && av == bv + case UnknownRepresentation: + bv, ok := b.(UnknownRepresentation) + + return ok && av.TypeName == bv.TypeName && bytes.Equal(av.raw, bv.raw) + default: + return false + } +} + +func unmarshalRepresentation(b json.RawMessage) (Representation, error) { + aux := struct { + Type string `json:"type"` + }{} + if err := json.Unmarshal(b, &aux); err != nil { + return nil, err + } + if aux.Type == "" { + return nil, fmt.Errorf("%w: representation requires a type", ErrInvalidUDFMetadata) + } + + if aux.Type == "sql" { + var r SQLRepresentation + if err := json.Unmarshal(b, &r); err != nil { + return nil, err + } + + return r, nil + } + + // Store the raw JSON compacted so that round-tripped representations + // compare equal: encoding/json compacts MarshalJSON output. + var compacted bytes.Buffer + if err := json.Compact(&compacted, b); err != nil { + return nil, err + } + + return UnknownRepresentation{TypeName: aux.Type, raw: compacted.Bytes()}, nil +} + +// Parameter is a single function parameter. +type Parameter struct { + Name string `json:"name"` + Type Type `json:"type"` + Doc string `json:"doc,omitempty"` +} + +func (p *Parameter) UnmarshalJSON(b []byte) error { + type Alias Parameter + aux := struct { + Type json.RawMessage `json:"type"` + *Alias + }{Alias: (*Alias)(p)} + + 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) + } + + typ, err := unmarshalType(aux.Type) + if err != nil { + return err + } + p.Type = typ + + return nil +} + +func (p Parameter) equals(other Parameter) bool { + return p.Name == other.Name && p.Doc == other.Doc && + p.Type != nil && other.Type != nil && p.Type.Equals(other.Type) +} + +// DefinitionVersion is a specific implementation of a definition at a +// given point in time. Versions are immutable: changes to a definition +// introduce a new version. +type DefinitionVersion struct { + // VersionID is a monotonically increasing identifier of the + // definition version. + VersionID int `json:"version-id"` + // Representations lists the UDF implementations of this version. + Representations []Representation `json:"representations"` + // Deterministic indicates whether the function is deterministic. + // Defaults to false. + Deterministic bool `json:"deterministic,omitempty"` + // OnNullInput defines how the UDF behaves when any input parameter + // is NULL. Defaults to OnNullInputCall when empty. + OnNullInput OnNullInput `json:"on-null-input,omitempty"` + // TimestampMS is the creation timestamp of this version (ms from epoch). + TimestampMS int64 `json:"timestamp-ms"` +} + +func (v *DefinitionVersion) UnmarshalJSON(b []byte) error { + type Alias DefinitionVersion + aux := struct { + Representations []json.RawMessage `json:"representations"` + *Alias + }{Alias: (*Alias)(v)} + + v.VersionID = -1 + v.TimestampMS = -1 + + if err := json.Unmarshal(b, &aux); err != nil { + return err + } + + v.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 + } + + return nil +} + +// NullInputBehavior returns the effective on-null-input behavior, +// applying the spec default (OnNullInputCall) when the field is absent. +func (v *DefinitionVersion) NullInputBehavior() OnNullInput { + if v.OnNullInput == "" { + return OnNullInputCall + } + + return v.OnNullInput +} + +// Equals compares two versions for semantic equality: the default +// on-null-input behavior compares equal to an explicit "call". +func (v *DefinitionVersion) Equals(other *DefinitionVersion) bool { + return v.VersionID == other.VersionID && + v.Deterministic == other.Deterministic && + v.NullInputBehavior() == other.NullInputBehavior() && + v.TimestampMS == other.TimestampMS && + slices.EqualFunc(v.Representations, other.Representations, representationsEqual) +} + +func (v *DefinitionVersion) Clone() *DefinitionVersion { + if v == nil { + return nil + } + + cloned := *v + cloned.Representations = slices.Clone(v.Representations) + + return &cloned +} + +func (v *DefinitionVersion) validate(definitionID string) error { + if v.VersionID == -1 { + return fmt.Errorf("%w: definition %q has a version missing version-id", + ErrInvalidUDFMetadata, definitionID) + } + if v.VersionID < 0 { + return fmt.Errorf("%w: definition %q has invalid negative version-id %d", + ErrInvalidUDFMetadata, definitionID, v.VersionID) + } + + if v.TimestampMS == -1 { + return fmt.Errorf("%w: definition %q version %d is missing timestamp-ms", + ErrInvalidUDFMetadata, definitionID, v.VersionID) + } + if v.TimestampMS < 0 { + return fmt.Errorf("%w: definition %q version %d has invalid negative timestamp-ms %d", + ErrInvalidUDFMetadata, definitionID, v.VersionID, v.TimestampMS) + } + + if len(v.Representations) == 0 { + return fmt.Errorf("%w: definition %q version %d must have at least one representation", + ErrInvalidUDFMetadata, definitionID, v.VersionID) + } + + switch v.OnNullInput { + case "", OnNullInputCall, OnNullInputReturnNull: + default: + return fmt.Errorf("%w: definition %q version %d has invalid on-null-input %q", + ErrInvalidUDFMetadata, definitionID, v.VersionID, v.OnNullInput) + } + + seenDialects := make(map[string]bool) + for _, repr := range v.Representations { + if repr == nil { + return fmt.Errorf("%w: definition %q version %d representations must not contain null entries", + ErrInvalidUDFMetadata, definitionID, v.VersionID) + } + if unknown, ok := repr.(UnknownRepresentation); ok && len(unknown.raw) == 0 { + return fmt.Errorf("%w: definition %q version %d has an unknown representation without raw JSON; unknown representations must originate from parsed metadata", + ErrInvalidUDFMetadata, definitionID, v.VersionID) + } + sqlRepr, ok := repr.(SQLRepresentation) + if !ok { + continue + } + if sqlRepr.Dialect == "" { + return fmt.Errorf("%w: definition %q version %d has a sql representation without a dialect", + ErrInvalidUDFMetadata, definitionID, v.VersionID) + } + if sqlRepr.SQL == "" { + return fmt.Errorf("%w: definition %q version %d has a sql representation without a sql expression", + ErrInvalidUDFMetadata, definitionID, v.VersionID) + } + dialect := strings.ToLower(sqlRepr.Dialect) + if seenDialects[dialect] { + return fmt.Errorf("%w: definition %q version %d has duplicate sql dialect %s", + ErrInvalidUDFMetadata, definitionID, v.VersionID, sqlRepr.Dialect) + } + seenDialects[dialect] = true + } + + return nil +} + +// Definition represents one function signature (e.g. add_one(int) vs +// add_one(float)). A definition is uniquely identified by its signature: +// the ordered list of parameter types, canonicalized as its DefinitionID. +type Definition struct { + // DefinitionID is the canonical string derived from the parameter + // types (see CanonicalDefinitionID). + DefinitionID string `json:"definition-id"` + // SpecificName is an optional user-assignable name for this + // definition, unique among all definitions of the function. + SpecificName string `json:"specific-name,omitempty"` + // Parameters is the ordered list of function parameters. Invocation + // order must match this list. + Parameters []Parameter `json:"parameters"` + // ReturnType is the declared return type. All versions must produce + // values of this type. + ReturnType Type `json:"return-type"` + // ReturnNullable hints whether the return value is nullable. + // Defaults to true when absent. + ReturnNullable *bool `json:"return-nullable,omitempty"` + // Versions lists the versioned implementations of this definition. + Versions []*DefinitionVersion `json:"versions"` + // CurrentVersionID identifies the current version of this definition. + CurrentVersionID int `json:"current-version-id"` + // FunctionType is "udf" for scalar functions or "udtf" for table + // functions. For "udtf", ReturnType must be a struct describing the + // output schema. + FunctionType FunctionType `json:"function-type"` + // Doc is an optional documentation string. + Doc string `json:"doc,omitempty"` +} + +func (d *Definition) UnmarshalJSON(b []byte) error { + type Alias Definition + // 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 + // parameters. + aux := struct { + DefinitionID *string `json:"definition-id"` + Parameters json.RawMessage `json:"parameters"` + ReturnType json.RawMessage `json:"return-type"` + *Alias + }{Alias: (*Alias)(d)} + + d.CurrentVersionID = -1 + + if err := json.Unmarshal(b, &aux); err != nil { + return err + } + + if aux.DefinitionID == nil { + return fmt.Errorf("%w: definition is missing definition-id", ErrInvalidUDFMetadata) + } + d.DefinitionID = *aux.DefinitionID + + if len(aux.Parameters) == 0 { + return fmt.Errorf("%w: definition %q is missing parameters", ErrInvalidUDFMetadata, d.DefinitionID) + } + if err := json.Unmarshal(aux.Parameters, &d.Parameters); err != nil { + return err + } + if d.Parameters == nil { + return fmt.Errorf("%w: definition %q is missing parameters", ErrInvalidUDFMetadata, d.DefinitionID) + } + + if len(aux.ReturnType) == 0 { + return fmt.Errorf("%w: definition %q requires a return-type", ErrInvalidUDFMetadata, d.DefinitionID) + } + + typ, err := unmarshalType(aux.ReturnType) + if err != nil { + return err + } + d.ReturnType = typ + + return nil +} + +// ReturnsNullable returns the effective return-nullable hint, applying +// the spec default (true) when the field is absent. +func (d *Definition) ReturnsNullable() bool { + return d.ReturnNullable == nil || *d.ReturnNullable +} + +// CurrentVersion returns the definition version identified by +// CurrentVersionID, or nil if no such version exists. +func (d *Definition) CurrentVersion() *DefinitionVersion { + return d.Version(d.CurrentVersionID) +} + +// Version returns the definition version with the given id, or nil if no +// such version exists. +func (d *Definition) Version(versionID int) *DefinitionVersion { + for _, v := range d.Versions { + if v.VersionID == versionID { + return v + } + } + + return nil +} + +// Equals compares two definitions for semantic equality: absent optional +// fields compare equal to their spec defaults. +func (d *Definition) Equals(other *Definition) bool { + return d.DefinitionID == other.DefinitionID && + d.SpecificName == other.SpecificName && + slices.EqualFunc(d.Parameters, other.Parameters, Parameter.equals) && + d.ReturnType != nil && other.ReturnType != nil && d.ReturnType.Equals(other.ReturnType) && + d.ReturnsNullable() == other.ReturnsNullable() && + d.CurrentVersionID == other.CurrentVersionID && + d.FunctionType == other.FunctionType && + d.Doc == other.Doc && + slices.EqualFunc(d.Versions, other.Versions, (*DefinitionVersion).Equals) +} + +func (d *Definition) Clone() *Definition { + if d == nil { + return nil + } + + cloned := *d + // Type values are immutable, so sharing them across clones is safe. + cloned.Parameters = slices.Clone(d.Parameters) + cloned.Versions = cloneSlice(d.Versions) + if d.ReturnNullable != nil { + nullable := *d.ReturnNullable + cloned.ReturnNullable = &nullable + } + + return &cloned +} + +func (d *Definition) validate() error { + seenNames := make(map[string]bool) + for _, p := range d.Parameters { + if p.Name == "" { + return fmt.Errorf("%w: definition %q has a parameter without a name", + ErrInvalidUDFMetadata, d.DefinitionID) + } + if seenNames[p.Name] { + return fmt.Errorf("%w: definition %q has duplicate parameter name %q", + ErrInvalidUDFMetadata, d.DefinitionID, p.Name) + } + seenNames[p.Name] = true + if err := validateType(p.Type); err != nil { + return fmt.Errorf("%w: definition %q parameter %q: %w", + ErrInvalidUDFMetadata, d.DefinitionID, p.Name, err) + } + } + + canonicalID, err := CanonicalDefinitionID(d.Parameters) + if err != nil { + return fmt.Errorf("%w: definition %q: %w", ErrInvalidUDFMetadata, d.DefinitionID, err) + } + if d.DefinitionID != canonicalID { + return fmt.Errorf("%w: definition-id %q does not match canonical form %q of its parameters", + ErrInvalidUDFMetadata, d.DefinitionID, canonicalID) + } + + if err := validateType(d.ReturnType); err != nil { + return fmt.Errorf("%w: definition %q return-type: %w", ErrInvalidUDFMetadata, d.DefinitionID, err) + } + + switch d.FunctionType { + case FunctionTypeUDF: + case FunctionTypeUDTF: + if _, ok := d.ReturnType.(StructType); !ok { + return fmt.Errorf("%w: definition %q is a udtf but its return-type is not a struct", + ErrInvalidUDFMetadata, d.DefinitionID) Review Comment: validateType accepts an empty struct, so a udtf can declare a zero-column output struct and still validate. Worth requiring a non-empty struct here to match the output-schema intent? -- 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]
