[PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 opened a new pull request, #629:
URL: https://github.com/apache/iceberg-go/pull/629

   ## Description
   This PR implements full support for updating Iceberg views for the REST catalog
   
   ## Changes
   - Implements an `UpdateView` method on the REST catalog
   - Implements `view.MetadataBuilder` to generate updates for use in `UpdateView` on the REST catalog
     - I did my best to blend the style of the existing codebase and use idiomatic Go, while still being faithful to the Java reference code
     - I transcribed every applicable unit test from the Java `ViewMetadataBuilder` tests to validate behavior
   - Refactors the existing state of the `view` package substantially
     - I apologize for the diff here, I was working off of a fork for a few weeks and had not previously seen the recent additions to view metadata support. Did my best to reconcile them
   - Refactors the REST catalog `CreateView` method to take variadic `CreateViewOpt`s and now return a `view.View`, similar to the behavior of the `CreateTable` method
   - Added unit and integration tests to the `rest` package to cover the new changes to the REST view methods
   - Small refactor to extract `sliceEqualHelper` from `table/metadata.go` to `internal/utils.go` for use in the `view` package
   - Refactors the CreateMetadata function used by the SQL catalog to return a `*View`, and extract the location from it


-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2542829721


##########
view/metadata.go:
##########
@@ -18,135 +18,523 @@
 package view
 
 import (
-	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"maps"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
 	"time"
 
 	"github.com/apache/iceberg-go"
-	"github.com/apache/iceberg-go/internal"
-	"github.com/apache/iceberg-go/io"
+	iceinternal "github.com/apache/iceberg-go/internal"
+	"github.com/apache/iceberg-go/table"
+
 	"github.com/google/uuid"
 )
 
-// LoadMetadata reads and parses a view metadata file from the specified location.
+var (
+	ErrInvalidViewMetadata              = errors.New("invalid view metadata")
+	ErrInvalidViewMetadataFormatVersion = errors.New("invalid or missing format-version in view metadata")
+)
+
+const (
+	// LastAddedID is used in place of ID fields (e.g. schema, version) to indicate that
+	// the last added instance of that type should be used.
+	LastAddedID                = -1
+	SupportedViewFormatVersion = 1
+	DefaultViewFormatVersion   = SupportedViewFormatVersion
+)
+
+const (
+	initialSchemaID = 0
+
+	viewEngineProperty = "engine-name"
+	defaultViewEngine  = "iceberg-go"
+)
+
+// Metadata for an iceberg view as specified in the Iceberg spec
+// https://iceberg.apache.org/view-spec/
+type Metadata interface {
+	// FormatVersion indicates the version of this metadata, 1 for V1
+	FormatVersion() int
+	// ViewUUID returns a UUID that identifies the view, generated when the
+	// view is created. Implementations must throw an exception if a view's
+	// UUID does not match the expected UUID after refreshing metadata.
+	ViewUUID() uuid.UUID
+	// Location is the table's base location. This is used by writers to determine
+	// where to store data files, manifest files, and table metadata files.
+	Location() string
+	// Schemas returns the list of view schemas
+	Schemas() []*iceberg.Schema
+	// CurrentVersionID returns the ID of the current version of the view (version-id)
+	CurrentVersionID() int64
+	// CurrentVersion returns the current version of the view
+	CurrentVersion() *Version
+	// CurrentSchemaID returns the ID of the current schema
+	CurrentSchemaID() int
+	// CurrentSchema returns the current schema of the view
+	CurrentSchema() *iceberg.Schema
+	// SchemasByID returns a map of schema IDs to schemas
+	SchemasByID() map[int]*iceberg.Schema
+	// Versions returns the list of view versions
+	Versions() []*Version
+	// VersionLog returns a list of version log entries
+	// with the timestamp and version-id for every change to current-version-id
+	VersionLog() []VersionLogEntry
+	// Properties is a string to string map of view properties.
+	Properties() iceberg.Properties
+
+	// Updates returns the list of metadata updates used to build this metadata
+	Updates() Updates
+
+	Equals(Metadata) bool
+}
+
+// VersionSummary is string to string map of summary metadata about a view's version
+type VersionSummary map[string]string
+
+// Representation is a struct containing information about a view's representation
+// https://iceberg.apache.org/view-spec/#sql-representation
+type Representation struct {
+	// Must be sql
+	Type string `json:"type"`
+	// A SQL SELECT statement
+	Sql string `json:"sql"`
+	// The dialect of the sql SELECT statement (e.g., "trino" or "spark")
+	Dialect string `json:"dialect"`
+}
+type Representations []Representation
+
+func NewRepresentation(sql string, dialect string) Representation {
+	return Representation{
+		Type:    "sql",
+		Sql:     sql,
+		Dialect: dialect,
+	}
+}
+
+type Version struct {
+	// ID for the version
+	VersionID int64 `json:"version-id"`
+	// ID of the schema for the view version
+	SchemaID int `json:"schema-id"`
+	// Timestamp when the version was created (ms from epoch)
+	TimestampMS int64 `json:"timestamp-ms"`
+	// A string to string map of summary metadata about the version
+	Summary VersionSummary `json:"summary"`
+	// A list of representations for the view definition
+	Representations []Representation `json:"representations"`
+	// The default view namespace (as a list of strings)
+	DefaultNamespace table.Identifier `json:"default-namespace"`
+	// An (optional) default catalog name for querying the view
+	DefaultCatalog string `json:"default-catalog,omitempty"`
+}
+
+type VersionOpt func(*Version)
+
+func WithVersionSummary(summary VersionSummary) VersionOpt {
+	return func(v *Version) {
+		if v.Summary == nil {
+			v.Summary = summary
+		} else {
+			maps.Copy(v.Summary, summary)
+		}
+	}
+}
+
+func WithDefaultViewCatalog(catalogName string) func(*Version) {
+	return func(version *Version) {
+		version.DefaultCatalog = catalogName
+	}
+}
+
+func WithTimestampMS(timestampMS int64) VersionOpt {
+	return func(version *Version) {
+		version.TimestampMS = timestampMS
+	}
+}
+
+// NewVersion creates a Version instance from the provided parameters.
+// If building updates using the MetadataBuilder, and one desires to use the last
+// added schema ID, one should use the LastAddedID constant as the provided schemaID
 //
-// Returns the loaded Metadata or an error if the file cannot be read or parsed.
-func LoadMetadata(ctx context.Context,
-	props iceberg.Properties,
-	metadataLocation string,
-	name string,
-	namespace string,
-) (_ Metadata, err error) {
-	fs, err := io.LoadFS(ctx, props, metadataLocation)
-	if err != nil {
-		return nil, fmt.Errorf("error loading view metadata: %w", err)
+// Note that NewVersion automatically seeds TimestampMS to time.Now().UnixMilli(),
+// and one should use the option WithTimestampMS to override this behavior.
+func NewVersion(id int64, schemaID int, representations []Representation, defaultNS table.Identifier, opts ...VersionOpt) (*Version, error) {
+	if id < 1 {
+		return nil, errors.New("id should be greater than 0")
 	}
 
-	inputFile, err := fs.Open(metadataLocation)
-	if err != nil {
-		return nil, fmt.Errorf("%s.%s: error encountered loading view metadata: %w", namespace, name, err)
+	if len(representations) == 0 {
+		return nil, errors.New("invalid view version: must have at least one representation")
 	}
-	defer internal.CheckedClose(inputFile, &err)
 
-	var m metadata
-	if err := json.NewDecoder(inputFile).Decode(&m); err != nil {
-		return nil, fmt.Errorf("error encountered decoding view metadata: %w", err)
+	version := &Version{
+		VersionID:        id,
+		SchemaID:         schemaID,
+		Representations:  representations,
+		Summary:          VersionSummary{viewEngineProperty: defaultViewEngine},
+		TimestampMS:      time.Now().UnixMilli(),
+		DefaultNamespace: defaultNS,
 	}
 
-	return &m, nil
+	for _, opt := range opts {
+		opt(version)
+	}
+
+	return version, nil
 }
 
-// NewMetadata returns a view metadata for a given version.
-func NewMetadata(schema *iceberg.Schema,
-	version Version,
-	loc string,
-	props iceberg.Properties,
-) (Metadata, error) {
-	timestampMs := time.Now().UnixMilli()
+// NewVersionFromSQL creates a new Version with a single representation
+// using the provided SQL and dialect "default".
+func NewVersionFromSQL(id int64, schemaID int, sql string, defaultNS table.Identifier, opts ...VersionOpt) (*Version, error) {
+	return NewVersion(id, schemaID, Representations{NewRepresentation(sql, "default")}, defaultNS, opts...)
+}
 
-	if version.SchemaID != schema.ID {
-		return nil, fmt.Errorf("%w: version.SchemaID does not match schema.ID", iceberg.ErrInvalidArgument)
+// Equals checks whether the other Version would behave the same
+// while ignoring the view version id, and the creation timestamp
+func (v *Version) Equals(other *Version) bool {
+	return v.SchemaID == other.SchemaID &&
+		v.DefaultCatalog == other.DefaultCatalog &&
+		slices.Equal(v.DefaultNamespace, other.DefaultNamespace) &&
+		maps.Equal(v.Summary, other.Summary) &&
+		slices.Equal(v.Representations, other.Representations)
+}
+
+func (v *Version) Clone() *Version {
+	if v == nil {
+		return nil
 	}
 
-	return &metadata{
-		UUID:             uuid.NewString(),
-		FmtVersion:       1,
-		Loc:              loc,
-		SchemaList:       []*iceberg.Schema{schema},
-		CurrentVersionId: version.VersionID,
-		VersionList:      []Version{version},
-		VersionLogList: []VersionLogEntry{
-			{
-				TimestampMs: timestampMs,
-				VersionID:   version.VersionID,
-			},
-		},
-		Props: props,
-	}, nil
+	cloned := *v
+	cloned.Summary = maps.Clone(v.Summary)
+	cloned.Representations = slices.Clone(v.Representations)
+	cloned.DefaultNamespace = slices.Clone(v.DefaultNamespace)
+	return &cloned
 }
 
-// CreateMetadata creates a new view metadata file and writes it to storage.
-//
-// Returns the full path to the created metadata file, or an error if creation fails.
-//
-// Note: This function only supports creating new views with format version 1.
-// It does not support updating existing view metadata.
-func CreateMetadata(
-	ctx context.Context,
-	catalogName string,
-	nsIdent []string,
-	schema *iceberg.Schema,
-	viewSQL string,
-	loc string,
-	props iceberg.Properties,
-) (metadataLocation string, err error) {
-	versionId := int64(1)
-	timestampMs := time.Now().UnixMilli()
-
-	viewVersion := Version{
-		VersionID:   versionId,
-		TimestampMs: timestampMs,
-		SchemaID:    schema.ID,
-		Summary:     map[string]string{"sql": viewSQL},
-		Representations: []SQLRepresentation{
-			{Type: "sql", SQL: viewSQL, Dialect: "default"},
-		},
-		DefaultCatalog:   &catalogName,
-		DefaultNamespace: nsIdent,
-	}
-
-	metadataLocation = loc + "/metadata/view-" + uuid.New().String() + ".metadata.json"
-	viewMetadata, err := NewMetadata(schema, viewVersion, loc, props)
+type VersionLogEntry struct {
+	// Timestamp when the view's current-version-id was updated (ms from epoch)
+	TimestampMS int64 `json:"timestamp-ms"`
+	// ID that current-version-id was set to
+	VersionID int64 `json:"version-id"`
+}
+
+// ParseMetadata parses json metadata 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 "", fmt.Errorf("failed to create view metadata: %w", err)
+		return nil, err
 	}
 
-	viewMetadataBytes, err := json.Marshal(viewMetadata)
-	if err != nil {
-		return "", fmt.Errorf("failed to marshal view metadata: %w", 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) {
+	ver := struct {
+		FormatVersion int `json:"format-version"`
+	}{}
+	if err := json.Unmarshal(b, &ver); err != nil {
+		return nil, err
 	}
 
-	fs, err := io.LoadFS(ctx, props, metadataLocation)
-	if err != nil {
-		return "", fmt.Errorf("failed to load filesystem for view metadata: %w", err)
+	var ret Metadata
+	ret = &metadata{}
+	return ret, json.Unmarshal(b, ret)
+}
+
+// indexBy indexes a slice into a map, using a provided extractKey function
+// The extractKey function will be called on each item in the slice and assign the
+// item as a value for that key in the resultant map.
+func indexBy[T any, K comparable](s []T, extractKey func(T) K) map[K]T {
+	index := make(map[K]T)
+	for _, v := range s {
+		index[extractKey(v)] = v
+	}
+	return index
+}
+
+// Cloner is an interface which implements a Clone method for deep copying itself
+type cloner[T any] interface {
+	// Clone returns a deep copy of the underlying object
+	Clone() T
+}
+
+// cloneSlice returns a deep-clone of a Slice of elements implementing Cloner
+func cloneSlice[T cloner[T]](val []T) []T {
+	cloned := make([]T, len(val))
+	for i, elem := range val {
+		cloned[i] = elem.Clone()
+	}
+	return cloned
+}
+
+// https://iceberg.apache.org/view-spec/
+type metadata struct {
+	FormatVersionValue    int                `json:"format-version"`
+	UUID                  uuid.UUID          `json:"view-uuid"`
+	Loc                   string             `json:"location"`
+	CurrentVersionIDValue int64              `json:"current-version-id"`
+	VersionList           []*Version         `json:"versions"`
+	SchemaList            []*iceberg.Schema  `json:"schemas"`
+	VersionLogList        []VersionLogEntry  `json:"version-log"`
+	Props                 iceberg.Properties `json:"properties,omitempty"`
+
+	// updates from builder if applicable
+	updates Updates
+
+	// cached lookup helpers, must be initialized in init()
+	lazyVersionsByID func() map[int64]*Version
+	lazySchemasByID  func() map[int]*iceberg.Schema
+}
+
+func (m *metadata) Equals(other Metadata) bool {
+	if other == nil {
+		return false
+	}
+
+	if m == other {
+		return true
+	}
+
+	return m.UUID == other.ViewUUID() &&
+		m.Loc == other.Location() &&
+		m.FormatVersionValue == other.FormatVersion() &&
+		iceinternal.SliceEqualHelper(m.Schemas(), other.Schemas()) &&
+		iceinternal.SliceEqualHelper(m.Versions(), other.Versions()) &&
+		m.CurrentVersionIDValue == other.CurrentVersionID() &&
+		slices.Equal(m.VersionLogList, other.VersionLog())
+}
+
+func (m *metadata) Updates() Updates { return m.updates }
+
+func (m *metadata) FormatVersion() int         { return m.FormatVersionValue }
+func (m *metadata) ViewUUID() uuid.UUID        { return m.UUID }
+func (m *metadata) Location() string           { return m.Loc }
+func (m *metadata) Versions() []*Version       { return m.VersionList }
+func (m *metadata) Schemas() []*iceberg.Schema { return m.SchemaList }
+func (m *metadata) SchemasByID() map[int]*iceberg.Schema {
+	return maps.Clone(m.lazySchemasByID())
+}
+
+func (m *metadata) CurrentVersionID() int64 {
+	return m.CurrentVersionIDValue
+}
+
+func (m *metadata) CurrentVersion() *Version {
+	version, ok := m.lazyVersionsByID()[m.CurrentVersionIDValue]
+	if !ok {
+		panic("current version not found")
 	}

Review Comment:
   This should never happen. Initialization of `metadata` should always be done with a call to `init` and `validate`. `validate` specifically checks that the version list and schema lists are nonempty, and also that the current version ID exists in the list of versions. Similar logic for schems. 



##########
view/metadata.go:
##########
@@ -18,135 +18,523 @@
 package view
 
 import (
-	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"maps"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
 	"time"
 
 	"github.com/apache/iceberg-go"
-	"github.com/apache/iceberg-go/internal"
-	"github.com/apache/iceberg-go/io"
+	iceinternal "github.com/apache/iceberg-go/internal"
+	"github.com/apache/iceberg-go/table"
+
 	"github.com/google/uuid"
 )
 
-// LoadMetadata reads and parses a view metadata file from the specified location.
+var (
+	ErrInvalidViewMetadata              = errors.New("invalid view metadata")
+	ErrInvalidViewMetadataFormatVersion = errors.New("invalid or missing format-version in view metadata")
+)
+
+const (
+	// LastAddedID is used in place of ID fields (e.g. schema, version) to indicate that
+	// the last added instance of that type should be used.
+	LastAddedID                = -1
+	SupportedViewFormatVersion = 1
+	DefaultViewFormatVersion   = SupportedViewFormatVersion
+)
+
+const (
+	initialSchemaID = 0
+
+	viewEngineProperty = "engine-name"
+	defaultViewEngine  = "iceberg-go"
+)
+
+// Metadata for an iceberg view as specified in the Iceberg spec
+// https://iceberg.apache.org/view-spec/
+type Metadata interface {
+	// FormatVersion indicates the version of this metadata, 1 for V1
+	FormatVersion() int
+	// ViewUUID returns a UUID that identifies the view, generated when the
+	// view is created. Implementations must throw an exception if a view's
+	// UUID does not match the expected UUID after refreshing metadata.
+	ViewUUID() uuid.UUID
+	// Location is the table's base location. This is used by writers to determine
+	// where to store data files, manifest files, and table metadata files.
+	Location() string
+	// Schemas returns the list of view schemas
+	Schemas() []*iceberg.Schema
+	// CurrentVersionID returns the ID of the current version of the view (version-id)
+	CurrentVersionID() int64
+	// CurrentVersion returns the current version of the view
+	CurrentVersion() *Version
+	// CurrentSchemaID returns the ID of the current schema
+	CurrentSchemaID() int
+	// CurrentSchema returns the current schema of the view
+	CurrentSchema() *iceberg.Schema
+	// SchemasByID returns a map of schema IDs to schemas
+	SchemasByID() map[int]*iceberg.Schema
+	// Versions returns the list of view versions
+	Versions() []*Version
+	// VersionLog returns a list of version log entries
+	// with the timestamp and version-id for every change to current-version-id
+	VersionLog() []VersionLogEntry
+	// Properties is a string to string map of view properties.
+	Properties() iceberg.Properties
+
+	// Updates returns the list of metadata updates used to build this metadata
+	Updates() Updates
+
+	Equals(Metadata) bool
+}
+
+// VersionSummary is string to string map of summary metadata about a view's version
+type VersionSummary map[string]string
+
+// Representation is a struct containing information about a view's representation
+// https://iceberg.apache.org/view-spec/#sql-representation
+type Representation struct {
+	// Must be sql
+	Type string `json:"type"`
+	// A SQL SELECT statement
+	Sql string `json:"sql"`
+	// The dialect of the sql SELECT statement (e.g., "trino" or "spark")
+	Dialect string `json:"dialect"`
+}
+type Representations []Representation
+
+func NewRepresentation(sql string, dialect string) Representation {
+	return Representation{
+		Type:    "sql",
+		Sql:     sql,
+		Dialect: dialect,
+	}
+}
+
+type Version struct {
+	// ID for the version
+	VersionID int64 `json:"version-id"`
+	// ID of the schema for the view version
+	SchemaID int `json:"schema-id"`
+	// Timestamp when the version was created (ms from epoch)
+	TimestampMS int64 `json:"timestamp-ms"`
+	// A string to string map of summary metadata about the version
+	Summary VersionSummary `json:"summary"`
+	// A list of representations for the view definition
+	Representations []Representation `json:"representations"`
+	// The default view namespace (as a list of strings)
+	DefaultNamespace table.Identifier `json:"default-namespace"`
+	// An (optional) default catalog name for querying the view
+	DefaultCatalog string `json:"default-catalog,omitempty"`
+}
+
+type VersionOpt func(*Version)
+
+func WithVersionSummary(summary VersionSummary) VersionOpt {
+	return func(v *Version) {
+		if v.Summary == nil {
+			v.Summary = summary
+		} else {
+			maps.Copy(v.Summary, summary)
+		}
+	}
+}
+
+func WithDefaultViewCatalog(catalogName string) func(*Version) {
+	return func(version *Version) {
+		version.DefaultCatalog = catalogName
+	}
+}
+
+func WithTimestampMS(timestampMS int64) VersionOpt {
+	return func(version *Version) {
+		version.TimestampMS = timestampMS
+	}
+}
+
+// NewVersion creates a Version instance from the provided parameters.
+// If building updates using the MetadataBuilder, and one desires to use the last
+// added schema ID, one should use the LastAddedID constant as the provided schemaID
 //
-// Returns the loaded Metadata or an error if the file cannot be read or parsed.
-func LoadMetadata(ctx context.Context,
-	props iceberg.Properties,
-	metadataLocation string,
-	name string,
-	namespace string,
-) (_ Metadata, err error) {
-	fs, err := io.LoadFS(ctx, props, metadataLocation)
-	if err != nil {
-		return nil, fmt.Errorf("error loading view metadata: %w", err)
+// Note that NewVersion automatically seeds TimestampMS to time.Now().UnixMilli(),
+// and one should use the option WithTimestampMS to override this behavior.
+func NewVersion(id int64, schemaID int, representations []Representation, defaultNS table.Identifier, opts ...VersionOpt) (*Version, error) {
+	if id < 1 {
+		return nil, errors.New("id should be greater than 0")
 	}
 
-	inputFile, err := fs.Open(metadataLocation)
-	if err != nil {
-		return nil, fmt.Errorf("%s.%s: error encountered loading view metadata: %w", namespace, name, err)
+	if len(representations) == 0 {
+		return nil, errors.New("invalid view version: must have at least one representation")
 	}
-	defer internal.CheckedClose(inputFile, &err)
 
-	var m metadata
-	if err := json.NewDecoder(inputFile).Decode(&m); err != nil {
-		return nil, fmt.Errorf("error encountered decoding view metadata: %w", err)
+	version := &Version{
+		VersionID:        id,
+		SchemaID:         schemaID,
+		Representations:  representations,
+		Summary:          VersionSummary{viewEngineProperty: defaultViewEngine},
+		TimestampMS:      time.Now().UnixMilli(),
+		DefaultNamespace: defaultNS,
 	}
 
-	return &m, nil
+	for _, opt := range opts {
+		opt(version)
+	}
+
+	return version, nil
 }
 
-// NewMetadata returns a view metadata for a given version.
-func NewMetadata(schema *iceberg.Schema,
-	version Version,
-	loc string,
-	props iceberg.Properties,
-) (Metadata, error) {
-	timestampMs := time.Now().UnixMilli()
+// NewVersionFromSQL creates a new Version with a single representation
+// using the provided SQL and dialect "default".
+func NewVersionFromSQL(id int64, schemaID int, sql string, defaultNS table.Identifier, opts ...VersionOpt) (*Version, error) {
+	return NewVersion(id, schemaID, Representations{NewRepresentation(sql, "default")}, defaultNS, opts...)
+}
 
-	if version.SchemaID != schema.ID {
-		return nil, fmt.Errorf("%w: version.SchemaID does not match schema.ID", iceberg.ErrInvalidArgument)
+// Equals checks whether the other Version would behave the same
+// while ignoring the view version id, and the creation timestamp
+func (v *Version) Equals(other *Version) bool {
+	return v.SchemaID == other.SchemaID &&
+		v.DefaultCatalog == other.DefaultCatalog &&
+		slices.Equal(v.DefaultNamespace, other.DefaultNamespace) &&
+		maps.Equal(v.Summary, other.Summary) &&
+		slices.Equal(v.Representations, other.Representations)
+}
+
+func (v *Version) Clone() *Version {
+	if v == nil {
+		return nil
 	}
 
-	return &metadata{
-		UUID:             uuid.NewString(),
-		FmtVersion:       1,
-		Loc:              loc,
-		SchemaList:       []*iceberg.Schema{schema},
-		CurrentVersionId: version.VersionID,
-		VersionList:      []Version{version},
-		VersionLogList: []VersionLogEntry{
-			{
-				TimestampMs: timestampMs,
-				VersionID:   version.VersionID,
-			},
-		},
-		Props: props,
-	}, nil
+	cloned := *v
+	cloned.Summary = maps.Clone(v.Summary)
+	cloned.Representations = slices.Clone(v.Representations)
+	cloned.DefaultNamespace = slices.Clone(v.DefaultNamespace)
+	return &cloned
 }
 
-// CreateMetadata creates a new view metadata file and writes it to storage.
-//
-// Returns the full path to the created metadata file, or an error if creation fails.
-//
-// Note: This function only supports creating new views with format version 1.
-// It does not support updating existing view metadata.
-func CreateMetadata(
-	ctx context.Context,
-	catalogName string,
-	nsIdent []string,
-	schema *iceberg.Schema,
-	viewSQL string,
-	loc string,
-	props iceberg.Properties,
-) (metadataLocation string, err error) {
-	versionId := int64(1)
-	timestampMs := time.Now().UnixMilli()
-
-	viewVersion := Version{
-		VersionID:   versionId,
-		TimestampMs: timestampMs,
-		SchemaID:    schema.ID,
-		Summary:     map[string]string{"sql": viewSQL},
-		Representations: []SQLRepresentation{
-			{Type: "sql", SQL: viewSQL, Dialect: "default"},
-		},
-		DefaultCatalog:   &catalogName,
-		DefaultNamespace: nsIdent,
-	}
-
-	metadataLocation = loc + "/metadata/view-" + uuid.New().String() + ".metadata.json"
-	viewMetadata, err := NewMetadata(schema, viewVersion, loc, props)
+type VersionLogEntry struct {
+	// Timestamp when the view's current-version-id was updated (ms from epoch)
+	TimestampMS int64 `json:"timestamp-ms"`
+	// ID that current-version-id was set to
+	VersionID int64 `json:"version-id"`
+}
+
+// ParseMetadata parses json metadata 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 "", fmt.Errorf("failed to create view metadata: %w", err)
+		return nil, err
 	}
 
-	viewMetadataBytes, err := json.Marshal(viewMetadata)
-	if err != nil {
-		return "", fmt.Errorf("failed to marshal view metadata: %w", 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) {
+	ver := struct {
+		FormatVersion int `json:"format-version"`
+	}{}
+	if err := json.Unmarshal(b, &ver); err != nil {
+		return nil, err
 	}
 
-	fs, err := io.LoadFS(ctx, props, metadataLocation)
-	if err != nil {
-		return "", fmt.Errorf("failed to load filesystem for view metadata: %w", err)
+	var ret Metadata
+	ret = &metadata{}
+	return ret, json.Unmarshal(b, ret)
+}
+
+// indexBy indexes a slice into a map, using a provided extractKey function
+// The extractKey function will be called on each item in the slice and assign the
+// item as a value for that key in the resultant map.
+func indexBy[T any, K comparable](s []T, extractKey func(T) K) map[K]T {
+	index := make(map[K]T)
+	for _, v := range s {
+		index[extractKey(v)] = v
+	}
+	return index
+}
+
+// Cloner is an interface which implements a Clone method for deep copying itself
+type cloner[T any] interface {
+	// Clone returns a deep copy of the underlying object
+	Clone() T
+}
+
+// cloneSlice returns a deep-clone of a Slice of elements implementing Cloner
+func cloneSlice[T cloner[T]](val []T) []T {
+	cloned := make([]T, len(val))
+	for i, elem := range val {
+		cloned[i] = elem.Clone()
+	}
+	return cloned
+}
+
+// https://iceberg.apache.org/view-spec/
+type metadata struct {
+	FormatVersionValue    int                `json:"format-version"`
+	UUID                  uuid.UUID          `json:"view-uuid"`
+	Loc                   string             `json:"location"`
+	CurrentVersionIDValue int64              `json:"current-version-id"`
+	VersionList           []*Version         `json:"versions"`
+	SchemaList            []*iceberg.Schema  `json:"schemas"`
+	VersionLogList        []VersionLogEntry  `json:"version-log"`
+	Props                 iceberg.Properties `json:"properties,omitempty"`
+
+	// updates from builder if applicable
+	updates Updates
+
+	// cached lookup helpers, must be initialized in init()
+	lazyVersionsByID func() map[int64]*Version
+	lazySchemasByID  func() map[int]*iceberg.Schema
+}
+
+func (m *metadata) Equals(other Metadata) bool {
+	if other == nil {
+		return false
+	}
+
+	if m == other {
+		return true
+	}
+
+	return m.UUID == other.ViewUUID() &&
+		m.Loc == other.Location() &&
+		m.FormatVersionValue == other.FormatVersion() &&
+		iceinternal.SliceEqualHelper(m.Schemas(), other.Schemas()) &&
+		iceinternal.SliceEqualHelper(m.Versions(), other.Versions()) &&
+		m.CurrentVersionIDValue == other.CurrentVersionID() &&
+		slices.Equal(m.VersionLogList, other.VersionLog())
+}
+
+func (m *metadata) Updates() Updates { return m.updates }
+
+func (m *metadata) FormatVersion() int         { return m.FormatVersionValue }
+func (m *metadata) ViewUUID() uuid.UUID        { return m.UUID }
+func (m *metadata) Location() string           { return m.Loc }
+func (m *metadata) Versions() []*Version       { return m.VersionList }
+func (m *metadata) Schemas() []*iceberg.Schema { return m.SchemaList }
+func (m *metadata) SchemasByID() map[int]*iceberg.Schema {
+	return maps.Clone(m.lazySchemasByID())
+}
+
+func (m *metadata) CurrentVersionID() int64 {
+	return m.CurrentVersionIDValue
+}
+
+func (m *metadata) CurrentVersion() *Version {
+	version, ok := m.lazyVersionsByID()[m.CurrentVersionIDValue]
+	if !ok {
+		panic("current version not found")
 	}

Review Comment:
   This should never happen. Initialization of `metadata` should always be done with a call to `init` and `validate`. `validate` specifically checks that the version list and schema lists are nonempty, and also that the current version ID exists in the list of versions. Similar logic for schemas. 



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2542719245


##########
catalog/rest/rest.go:
##########
@@ -1159,55 +1159,95 @@ type createViewRequest struct {
 	Schema      *iceberg.Schema    `json:"schema"`
 	Location    string             `json:"location,omitempty"`
 	Props       iceberg.Properties `json:"properties,omitempty"`
-	SQL         string             `json:"sql"`
-	ViewVersion view.Version       `json:"view-version"`
+	ViewVersion *view.Version      `json:"view-version"`
 }
 
 type viewResponse struct {
 	MetadataLoc string             `json:"metadata-location"`
 	RawMetadata json.RawMessage    `json:"metadata"`
 	Config      iceberg.Properties `json:"config"`
-	Metadata    table.Metadata     `json:"-"`
+	Metadata    view.Metadata      `json:"-"`
+}
+
+func (v *viewResponse) UnmarshalJSON(b []byte) (err error) {
+	type Alias viewResponse
+	if err = json.Unmarshal(b, (*Alias)(v)); err != nil {
+		return err
+	}
+
+	v.Metadata, err = view.ParseMetadataBytes(v.RawMetadata)
+
+	return
 }
 
 // CreateView creates a new view in the catalog.
-func (r *Catalog) CreateView(ctx context.Context, identifier table.Identifier, schema *iceberg.Schema, sql string, props iceberg.Properties) error {
-	ns, v, err := splitIdentForPath(identifier)
+func (r *Catalog) CreateView(ctx context.Context, identifier table.Identifier, version *view.Version, schema *iceberg.Schema, opts ...catalog.CreateViewOpt) (*view.View, error) {
+	ns, viewName, err := splitIdentForPath(identifier)
 	if err != nil {
-		return err
+		return nil, err
+	}
+
+	if opts == nil {
+		opts = []catalog.CreateViewOpt{}
+	}
+	cfg := catalog.NewCreateViewCfg()
+	for _, opt := range opts {
+		opt(&cfg)
 	}
 
 	freshSchema, err := iceberg.AssignFreshSchemaIDs(schema, nil)
 	if err != nil {
-		return err
+		return nil, err
+	}
+

Review Comment:
   I think one way of doing this is making Version an interface and making the struct private. What do you think? TimestampMS is defaulted to `now` in that constructor. 
   
   If we only take representations we lose the ability to set the summary and default NS, and adding those explicitly to the REST call bloats the parameter list



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2542851441


##########
view/metadata_builder.go:
##########
@@ -1,57 +1,468 @@
-// 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 view
 
 import (
+	"errors"
+	"fmt"
+	"maps"
+	"slices"
+	"strings"
+	"time"
+
 	"github.com/apache/iceberg-go"
+
+	"github.com/google/uuid"
 )
 
-// MetadataBuilder is a struct used for building and updating Iceberg view metadata.
-type MetadataBuilder struct{}
+type MetadataBuilder struct {
+	base    Metadata
+	updates Updates
+
+	// common fields
+	formatVersion    int
+	uuid             uuid.UUID
+	loc              string
+	schemaList       []*iceberg.Schema
+	versionList      []*Version
+	currentVersionID int64
+	versionLog       []VersionLogEntry
+	props            iceberg.Properties
+
+	// lookup maps
+	versionsById map[int64]*Version
+	schemasById  map[int]*iceberg.Schema
+
+	// update tracking
+	versionHistoryEntry *VersionLogEntry
+	lastAddedVersionID  *int64
+	lastAddedSchemaID   *int
+
+	// error tracking for build chaining
+	// if set, subsequent operations become a noop
+	err error
+}
+
+func NewMetadataBuilder() (*MetadataBuilder, error) {
+	return &MetadataBuilder{
+		formatVersion: DefaultViewFormatVersion,
+		updates:       make([]Update, 0),
+		schemaList:    make([]*iceberg.Schema, 0),
+		versionList:   make([]*Version, 0),
+		versionLog:    make([]VersionLogEntry, 0),
+		props:         make(iceberg.Properties),
+		versionsById:  make(map[int64]*Version),
+		schemasById:   make(map[int]*iceberg.Schema),
+	}, nil
+}
+
+func MetadataBuilderFromBase(metadata Metadata) (*MetadataBuilder, error) {
+	b := &MetadataBuilder{}
+	b.base = metadata
+
+	// Copy fields
+	b.formatVersion = metadata.FormatVersion()
+	b.uuid = metadata.ViewUUID()
+	b.loc = metadata.Location()
+	b.schemaList = slices.Clone(metadata.Schemas())
+	b.currentVersionID = metadata.CurrentVersionID()
+	b.versionList = cloneSlice(metadata.Versions())
+	b.versionLog = slices.Clone(metadata.VersionLog())
+	b.props = maps.Clone(metadata.Properties())
+
+	// Build lookup maps
+	b.versionsById = indexBy(b.versionList, func(vl *Version) int64 { return vl.VersionID })
+	b.schemasById = indexBy(b.schemaList, func(sm *iceberg.Schema) int { return sm.ID })
+
+	return b, nil
+}
+
+func (b *MetadataBuilder) HasChanges() bool { return len(b.updates) > 0 }
 
-func (b *MetadataBuilder) AssignUUID(_ string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetCurrentVersion(version *Version, schema *iceberg.Schema) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	newSchemaID, err := b.addSchema(schema)
+	if b.setErr(err) {
+		return b
+	}
+
+	newVersion := version.Clone()
+	newVersion.SchemaID = newSchemaID
+	newVersionID, err := b.addVersion(newVersion)
+	if b.setErr(err) {
+		return b
+	}
+
+	return b.SetCurrentVersionID(newVersionID)
 }
 
-func (b *MetadataBuilder) UpgradeFormatVersion(_ int) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) AddVersion(newVersion *Version) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	_, err := b.addVersion(newVersion)
+	if b.setErr(err) {
+		return b
+	}
+	return b
 }
 
-func (b *MetadataBuilder) AddSchema(_ *iceberg.Schema) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) addVersion(newVersion *Version) (int64, error) {
+	newVersionID := b.reuseOrCreateNewVersionID(newVersion)
+	version := newVersion.Clone()
+	if newVersionID != version.VersionID {
+		version.VersionID = newVersionID
+	}
+
+	// Check if this version was added in an update already
+	if _, ok := b.versionsById[newVersionID]; ok {
+		for _, upd := range b.updates {
+			if vvu, ok := upd.(*addViewVersionUpdate); ok && vvu.Version.VersionID == newVersionID {
+				b.lastAddedVersionID = &newVersionID
+				return newVersionID, nil
+			}
+		}
+	}
+
+	// If the SchemaID of the version is unset (nil), we attach the lastAddedSchemaID to it.
+	// If we have no lastAddedSchemaID, we fail
+	if version.SchemaID == LastAddedID {
+		if b.lastAddedSchemaID == nil {
+			return 0, errors.New("cannot set last added schema: no schema has been added")
+		}
+		version.SchemaID = *b.lastAddedSchemaID
+	}
+
+	if _, ok := b.schemasById[version.SchemaID]; !ok {
+		return 0, fmt.Errorf("cannot add version with unknown schema: %d", version.SchemaID)
+	}
+
+	if len(version.Representations) == 0 {
+		return 0, fmt.Errorf("cannot add version with no representations")
+	}
+	dialects := make(map[string]struct{})
+	for _, repr := range version.Representations {
+		normalizedDialect := strings.ToLower(repr.Dialect)
+		if _, ok := dialects[normalizedDialect]; ok {
+			return 0, fmt.Errorf("invalid view version: cannot add multiple queries for dialect %s", normalizedDialect)
+		}
+		dialects[normalizedDialect] = struct{}{}
+	}
+
+	b.versionList = append(b.versionList, version)
+	b.versionsById[version.VersionID] = version
+
+	if b.lastAddedSchemaID != nil && version.SchemaID == *b.lastAddedSchemaID {
+		updateVersion := version.Clone()
+		updateVersion.SchemaID = LastAddedID
+		b.updates = append(b.updates, NewAddViewVersionUpdate(updateVersion))
+	} else {
+		b.updates = append(b.updates, NewAddViewVersionUpdate(version))
+	}
+
+	b.lastAddedVersionID = &newVersionID
+	return newVersionID, nil
 }
 
-func (b *MetadataBuilder) SetLocation(_ string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) AddSchema(schema *iceberg.Schema) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	_, err := b.addSchema(schema)
+	if b.setErr(err) {
+		return b
+	}
+	return b
 }
 
-func (b *MetadataBuilder) SetProperties(_ map[string]string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) addSchema(schema *iceberg.Schema) (int, error) {
+	newSchemaID := b.reuseOrCreateNewSchemaID(schema)
+
+	if _, ok := b.schemasById[newSchemaID]; ok {
+		return newSchemaID, nil
+	}
+
+	newSchema := schema
+	// Build a fresh schema if we reset the ID
+	if schema.ID != newSchemaID {
+		newSchema = iceberg.NewSchemaWithIdentifiers(newSchemaID, schema.IdentifierFieldIDs, schema.Fields()...)
+	}
+
+	b.schemaList = append(b.schemaList, newSchema)
+	b.schemasById[newSchema.ID] = newSchema
+	b.updates = append(b.updates, NewAddSchemaUpdate(newSchema))
+	b.lastAddedSchemaID = &newSchemaID
+
+	return newSchemaID, nil
 }
 
-func (b *MetadataBuilder) RemoveProperties(_ []string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) RemoveProperties(keys []string) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if len(keys) == 0 {
+		return b
+	}
+
+	b.updates = append(b.updates, NewRemovePropertiesUpdate(keys))
+	for _, key := range keys {
+		delete(b.props, key)
+	}
+
+	return b
 }
 
-func (b *MetadataBuilder) AddVersion(_ *Version) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetCurrentVersionID(newVersionID int64) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if newVersionID == LastAddedID {
+		if b.lastAddedVersionID == nil {
+			b.err = errors.New("cannot set current version to last added, no version has been added")
+			return b
+		}
+		newVersionID = *b.lastAddedVersionID
+	}
+
+	if newVersionID == b.currentVersionID {
+		return b
+	}
+
+	_, ok := b.versionsById[newVersionID]
+	if !ok {
+		b.err = fmt.Errorf("cannot set current version to unknown version with id %d", newVersionID)
+		return b
+	}
+
+	if b.lastAddedVersionID != nil && *b.lastAddedVersionID == newVersionID {
+		b.updates = append(b.updates, NewSetCurrentVersionUpdate(LastAddedID))
+	} else {
+		b.updates = append(b.updates, NewSetCurrentVersionUpdate(newVersionID))
+	}
+	b.currentVersionID = newVersionID
+
+	// Set the current history entry.
+	// If the version was added in the current set of changes, use its timestamp,
+	// otherwise use system time.
+	updateTimestampMS := time.Now().UnixMilli()
+	for _, update := range b.updates {
+		if v, ok := update.(*addViewVersionUpdate); ok && v.Version.VersionID == newVersionID {
+			updateTimestampMS = v.Version.TimestampMS
+			break
+		}
+	}
+
+	b.versionHistoryEntry = &VersionLogEntry{
+		VersionID:   newVersionID,
+		TimestampMS: updateTimestampMS,
+	}
+
+	return b
 }
 
-func (b *MetadataBuilder) SetCurrentVersionID(_ int64) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetFormatVersion(formatVersion int) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if formatVersion < b.formatVersion {
+		b.err = fmt.Errorf("downgrading format version from %d to %d is not allowed",
+			b.formatVersion, formatVersion)
+		return b
+	}
+
+	if formatVersion > SupportedViewFormatVersion {
+		b.err = fmt.Errorf("unsupported format version %d", formatVersion)
+		return b
+	}
+
+	if formatVersion == b.formatVersion {
+		return b
+	}
+
+	b.updates = append(b.updates, NewUpgradeFormatVersionUpdate(formatVersion))
+	b.formatVersion = formatVersion
+
+	return b
+}
+
+func (b *MetadataBuilder) SetLoc(loc string) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if b.loc == loc {
+		return b
+	}
+
+	b.updates = append(b.updates, NewSetLocationUpdate(loc))
+	b.loc = loc
+
+	return b
+}
+
+func (b *MetadataBuilder) SetProperties(props iceberg.Properties) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if len(props) == 0 {
+		return b
+	}
+
+	b.updates = append(b.updates, NewSetPropertiesUpdate(props))
+	if b.props == nil {
+		b.props = props
+	} else {
+		maps.Copy(b.props, props)
+	}
+
+	return b
+}
+
+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
+	}
+
+	// Noop - same UUID does not generate a change.
+	if b.uuid == newUUID {
+		return b
+	}
+
+	if b.uuid != uuid.Nil {
+		b.err = errors.New("cannot reassign uuid")
+		return b
+	}
+
+	b.updates = append(b.updates, NewAssignUUIDUpdate(newUUID))
+	b.uuid = newUUID
+
+	return b
+}
+
+func (b *MetadataBuilder) Err() error {
+	return b.err
+}
+
+func (b *MetadataBuilder) buildMetadata(withUpdates bool) (*metadata, error) {
+	uuid_ := b.uuid
+	if uuid_ == uuid.Nil {
+		uuid_ = uuid.New()
+	}
+
+	md := &metadata{
+		FormatVersionValue:    b.formatVersion,
+		UUID:                  uuid_,
+		Loc:                   b.loc,
+		CurrentVersionIDValue: b.currentVersionID,
+		VersionList:           b.versionList,
+		VersionLogList:        b.versionLog,
+		SchemaList:            b.schemaList,
+		Props:                 b.props,
+	}
+	if withUpdates {
+		md.updates = b.updates
+	}
+	md.init()
+
+	return md, nil
+}
+
+// build builds the view metadata updates from the builder
+// if withUpdates is set to false, updates will not be included
+// in the returned Metadata, thus its Updates() will return an empty slice
+func (b *MetadataBuilder) build(withUpdates bool) (Metadata, error) {

Review Comment:
   Yes sorry I should have left a comment about this in my summary, I intentionally omitted this for now. It's my understanding that at least the rest catalog will manage history retention, so I made the decision to omit this for an MVP to avoid further bloat in the PR if catalogs will ultimately handle retention (since there is no Go catalog implementation I know of).
   
   What are your thoughts? 



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["twuebi (via GitHub)" <gi...@apache.org>]

twuebi commented on PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#issuecomment-3552641353

   > It looks like this PR has some overlaps with what @twuebi is working on. Could we limit the changes in this PR to the REST catalog?
   
   I didn't realize this PR existed one, I'm not attached to my PR, I'll close mine.
   
   


-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["alessandro-nori (via GitHub)" <gi...@apache.org>]

alessandro-nori commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2542822569


##########
catalog/rest/rest.go:
##########
@@ -1159,55 +1159,95 @@ type createViewRequest struct {
 	Schema      *iceberg.Schema    `json:"schema"`
 	Location    string             `json:"location,omitempty"`
 	Props       iceberg.Properties `json:"properties,omitempty"`
-	SQL         string             `json:"sql"`
-	ViewVersion view.Version       `json:"view-version"`
+	ViewVersion *view.Version      `json:"view-version"`
 }
 
 type viewResponse struct {
 	MetadataLoc string             `json:"metadata-location"`
 	RawMetadata json.RawMessage    `json:"metadata"`
 	Config      iceberg.Properties `json:"config"`
-	Metadata    table.Metadata     `json:"-"`
+	Metadata    view.Metadata      `json:"-"`
+}
+
+func (v *viewResponse) UnmarshalJSON(b []byte) (err error) {
+	type Alias viewResponse
+	if err = json.Unmarshal(b, (*Alias)(v)); err != nil {
+		return err
+	}
+
+	v.Metadata, err = view.ParseMetadataBytes(v.RawMetadata)
+
+	return
 }
 
 // CreateView creates a new view in the catalog.
-func (r *Catalog) CreateView(ctx context.Context, identifier table.Identifier, schema *iceberg.Schema, sql string, props iceberg.Properties) error {
-	ns, v, err := splitIdentForPath(identifier)
+func (r *Catalog) CreateView(ctx context.Context, identifier table.Identifier, version *view.Version, schema *iceberg.Schema, opts ...catalog.CreateViewOpt) (*view.View, error) {
+	ns, viewName, err := splitIdentForPath(identifier)
 	if err != nil {
-		return err
+		return nil, err
+	}
+
+	if opts == nil {
+		opts = []catalog.CreateViewOpt{}
+	}
+	cfg := catalog.NewCreateViewCfg()
+	for _, opt := range opts {
+		opt(&cfg)
 	}
 
 	freshSchema, err := iceberg.AssignFreshSchemaIDs(schema, nil)
 	if err != nil {
-		return err
+		return nil, err
+	}
+

Review Comment:
   Makes sense, thanks for the clarification.
   About `Version`, I would be consistent with how `Snapshot` is handled for `table.Metadata` so I think the current implementation is fine.



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["twuebi (via GitHub)" <gi...@apache.org>]

twuebi commented on PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#issuecomment-3607604064

   sorry for the delay @jnagel12, I would like us to move it to the constructor of the view metadata since catalog applications will have to do this server-side as clients are generally not to be trusted 


-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["zeroshade (via GitHub)" <gi...@apache.org>]

zeroshade commented on PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#issuecomment-3597384326

   @alessandro-nori @twuebi any further comments on this?


-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#issuecomment-3608095386

   Seems like macos 1.13.6 is flaky, not sure how to retrigger


-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#issuecomment-3555083360

   An update for maintainer(s) on this - after discussion with @twuebi , will make the following changes to the `MetadataBuilder` in this PR.
   
   - Implement history retention management
   - Implement enforcement of policy around dropping dialects
   - Will return View MD and Updates as separate entities from the builder packaged in a container struct
   
   As I mentioned above will be unavailable tomorrow but will get back to this Friday. Hopefully the rest of the PR should be reviewable as is in the meantime 🙇 


-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["alessandro-nori (via GitHub)" <gi...@apache.org>]

alessandro-nori commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2541936246


##########
catalog/rest/rest.go:
##########
@@ -1159,55 +1159,95 @@ type createViewRequest struct {
 	Schema      *iceberg.Schema    `json:"schema"`
 	Location    string             `json:"location,omitempty"`
 	Props       iceberg.Properties `json:"properties,omitempty"`

Review Comment:
   `properties` is marked as required field in the spec so we should remove the `omitempty`
   
   https://github.com/apache/iceberg/blob/main/open-api/rest-catalog-open-api.yaml#L3556-L3560



##########
catalog/rest/rest.go:
##########
@@ -1159,55 +1159,95 @@ type createViewRequest struct {
 	Schema      *iceberg.Schema    `json:"schema"`
 	Location    string             `json:"location,omitempty"`
 	Props       iceberg.Properties `json:"properties,omitempty"`
-	SQL         string             `json:"sql"`
-	ViewVersion view.Version       `json:"view-version"`
+	ViewVersion *view.Version      `json:"view-version"`
 }
 
 type viewResponse struct {
 	MetadataLoc string             `json:"metadata-location"`
 	RawMetadata json.RawMessage    `json:"metadata"`
 	Config      iceberg.Properties `json:"config"`
-	Metadata    table.Metadata     `json:"-"`
+	Metadata    view.Metadata      `json:"-"`
+}
+
+func (v *viewResponse) UnmarshalJSON(b []byte) (err error) {
+	type Alias viewResponse
+	if err = json.Unmarshal(b, (*Alias)(v)); err != nil {
+		return err
+	}
+
+	v.Metadata, err = view.ParseMetadataBytes(v.RawMetadata)
+
+	return
 }
 
 // CreateView creates a new view in the catalog.
-func (r *Catalog) CreateView(ctx context.Context, identifier table.Identifier, schema *iceberg.Schema, sql string, props iceberg.Properties) error {
-	ns, v, err := splitIdentForPath(identifier)
+func (r *Catalog) CreateView(ctx context.Context, identifier table.Identifier, version *view.Version, schema *iceberg.Schema, opts ...catalog.CreateViewOpt) (*view.View, error) {
+	ns, viewName, err := splitIdentForPath(identifier)
 	if err != nil {
-		return err
+		return nil, err
+	}
+
+	if opts == nil {
+		opts = []catalog.CreateViewOpt{}
+	}
+	cfg := catalog.NewCreateViewCfg()
+	for _, opt := range opts {
+		opt(&cfg)
 	}
 
 	freshSchema, err := iceberg.AssignFreshSchemaIDs(schema, nil)
 	if err != nil {
-		return err
+		return nil, err
+	}
+

Review Comment:
   since `Version` is now coming from the caller, I think we should enforce that fields like VersionID, TimestampMs are correctly set before the request is sent to the server.
   
   Alternatively we could change the method signature to receive a slice of `Representation`s instead of `view.Version`



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#issuecomment-3570868551

   Ok, implemented the items discussed above, sorry for the delay!


-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["twuebi (via GitHub)" <gi...@apache.org>]

twuebi commented on PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#issuecomment-3602820197

   Sure, I'm wondering if there's a good reason not to do it here in the view constructor, as in if there's any harm in it.


-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2542816834


##########
view/metadata.go:
##########
@@ -18,135 +18,523 @@
 package view
 
 import (
-	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"maps"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
 	"time"
 
 	"github.com/apache/iceberg-go"
-	"github.com/apache/iceberg-go/internal"
-	"github.com/apache/iceberg-go/io"
+	iceinternal "github.com/apache/iceberg-go/internal"
+	"github.com/apache/iceberg-go/table"
+
 	"github.com/google/uuid"
 )
 
-// LoadMetadata reads and parses a view metadata file from the specified location.
+var (
+	ErrInvalidViewMetadata              = errors.New("invalid view metadata")
+	ErrInvalidViewMetadataFormatVersion = errors.New("invalid or missing format-version in view metadata")
+)
+
+const (
+	// LastAddedID is used in place of ID fields (e.g. schema, version) to indicate that
+	// the last added instance of that type should be used.
+	LastAddedID                = -1
+	SupportedViewFormatVersion = 1
+	DefaultViewFormatVersion   = SupportedViewFormatVersion
+)
+
+const (
+	initialSchemaID = 0
+
+	viewEngineProperty = "engine-name"
+	defaultViewEngine  = "iceberg-go"
+)
+
+// Metadata for an iceberg view as specified in the Iceberg spec
+// https://iceberg.apache.org/view-spec/
+type Metadata interface {
+	// FormatVersion indicates the version of this metadata, 1 for V1
+	FormatVersion() int
+	// ViewUUID returns a UUID that identifies the view, generated when the
+	// view is created. Implementations must throw an exception if a view's
+	// UUID does not match the expected UUID after refreshing metadata.
+	ViewUUID() uuid.UUID
+	// Location is the table's base location. This is used by writers to determine
+	// where to store data files, manifest files, and table metadata files.
+	Location() string
+	// Schemas returns the list of view schemas
+	Schemas() []*iceberg.Schema
+	// CurrentVersionID returns the ID of the current version of the view (version-id)
+	CurrentVersionID() int64
+	// CurrentVersion returns the current version of the view
+	CurrentVersion() *Version
+	// CurrentSchemaID returns the ID of the current schema
+	CurrentSchemaID() int
+	// CurrentSchema returns the current schema of the view
+	CurrentSchema() *iceberg.Schema
+	// SchemasByID returns a map of schema IDs to schemas
+	SchemasByID() map[int]*iceberg.Schema
+	// Versions returns the list of view versions
+	Versions() []*Version
+	// VersionLog returns a list of version log entries
+	// with the timestamp and version-id for every change to current-version-id
+	VersionLog() []VersionLogEntry
+	// Properties is a string to string map of view properties.
+	Properties() iceberg.Properties
+
+	// Updates returns the list of metadata updates used to build this metadata
+	Updates() Updates

Review Comment:
   I also was not a fan of it being on the metadata itself, but it's what the Java implementation does. I more or less view it as a reference implementation so figured I'd replicate it, but obviously `Build()` and `BuildWithoutChanges()` is awkward as you mentioned.
   
   This seems like a fine compromise. What would you think about embedding metadata in the `BuildResult` for the convenience of method promotion? 



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["twuebi (via GitHub)" <gi...@apache.org>]

twuebi commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2543700071


##########
view/metadata_builder.go:
##########
@@ -1,57 +1,468 @@
-// 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 view
 
 import (
+	"errors"
+	"fmt"
+	"maps"
+	"slices"
+	"strings"
+	"time"
+
 	"github.com/apache/iceberg-go"
+
+	"github.com/google/uuid"
 )
 
-// MetadataBuilder is a struct used for building and updating Iceberg view metadata.
-type MetadataBuilder struct{}
+type MetadataBuilder struct {
+	base    Metadata
+	updates Updates
+
+	// common fields
+	formatVersion    int
+	uuid             uuid.UUID
+	loc              string
+	schemaList       []*iceberg.Schema
+	versionList      []*Version
+	currentVersionID int64
+	versionLog       []VersionLogEntry
+	props            iceberg.Properties
+
+	// lookup maps
+	versionsById map[int64]*Version
+	schemasById  map[int]*iceberg.Schema
+
+	// update tracking
+	versionHistoryEntry *VersionLogEntry
+	lastAddedVersionID  *int64
+	lastAddedSchemaID   *int
+
+	// error tracking for build chaining
+	// if set, subsequent operations become a noop
+	err error
+}
+
+func NewMetadataBuilder() (*MetadataBuilder, error) {
+	return &MetadataBuilder{
+		formatVersion: DefaultViewFormatVersion,
+		updates:       make([]Update, 0),
+		schemaList:    make([]*iceberg.Schema, 0),
+		versionList:   make([]*Version, 0),
+		versionLog:    make([]VersionLogEntry, 0),
+		props:         make(iceberg.Properties),
+		versionsById:  make(map[int64]*Version),
+		schemasById:   make(map[int]*iceberg.Schema),
+	}, nil
+}
+
+func MetadataBuilderFromBase(metadata Metadata) (*MetadataBuilder, error) {
+	b := &MetadataBuilder{}
+	b.base = metadata
+
+	// Copy fields
+	b.formatVersion = metadata.FormatVersion()
+	b.uuid = metadata.ViewUUID()
+	b.loc = metadata.Location()
+	b.schemaList = slices.Clone(metadata.Schemas())
+	b.currentVersionID = metadata.CurrentVersionID()
+	b.versionList = cloneSlice(metadata.Versions())
+	b.versionLog = slices.Clone(metadata.VersionLog())
+	b.props = maps.Clone(metadata.Properties())
+
+	// Build lookup maps
+	b.versionsById = indexBy(b.versionList, func(vl *Version) int64 { return vl.VersionID })
+	b.schemasById = indexBy(b.schemaList, func(sm *iceberg.Schema) int { return sm.ID })
+
+	return b, nil
+}
+
+func (b *MetadataBuilder) HasChanges() bool { return len(b.updates) > 0 }
 
-func (b *MetadataBuilder) AssignUUID(_ string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetCurrentVersion(version *Version, schema *iceberg.Schema) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	newSchemaID, err := b.addSchema(schema)
+	if b.setErr(err) {
+		return b
+	}
+
+	newVersion := version.Clone()
+	newVersion.SchemaID = newSchemaID
+	newVersionID, err := b.addVersion(newVersion)
+	if b.setErr(err) {
+		return b
+	}
+
+	return b.SetCurrentVersionID(newVersionID)
 }
 
-func (b *MetadataBuilder) UpgradeFormatVersion(_ int) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) AddVersion(newVersion *Version) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	_, err := b.addVersion(newVersion)
+	if b.setErr(err) {
+		return b
+	}
+	return b
 }
 
-func (b *MetadataBuilder) AddSchema(_ *iceberg.Schema) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) addVersion(newVersion *Version) (int64, error) {
+	newVersionID := b.reuseOrCreateNewVersionID(newVersion)
+	version := newVersion.Clone()
+	if newVersionID != version.VersionID {
+		version.VersionID = newVersionID
+	}
+
+	// Check if this version was added in an update already
+	if _, ok := b.versionsById[newVersionID]; ok {
+		for _, upd := range b.updates {
+			if vvu, ok := upd.(*addViewVersionUpdate); ok && vvu.Version.VersionID == newVersionID {
+				b.lastAddedVersionID = &newVersionID
+				return newVersionID, nil
+			}
+		}
+	}
+
+	// If the SchemaID of the version is unset (nil), we attach the lastAddedSchemaID to it.
+	// If we have no lastAddedSchemaID, we fail
+	if version.SchemaID == LastAddedID {
+		if b.lastAddedSchemaID == nil {
+			return 0, errors.New("cannot set last added schema: no schema has been added")
+		}
+		version.SchemaID = *b.lastAddedSchemaID
+	}
+
+	if _, ok := b.schemasById[version.SchemaID]; !ok {
+		return 0, fmt.Errorf("cannot add version with unknown schema: %d", version.SchemaID)
+	}
+
+	if len(version.Representations) == 0 {
+		return 0, fmt.Errorf("cannot add version with no representations")
+	}
+	dialects := make(map[string]struct{})
+	for _, repr := range version.Representations {
+		normalizedDialect := strings.ToLower(repr.Dialect)
+		if _, ok := dialects[normalizedDialect]; ok {
+			return 0, fmt.Errorf("invalid view version: cannot add multiple queries for dialect %s", normalizedDialect)
+		}
+		dialects[normalizedDialect] = struct{}{}
+	}
+
+	b.versionList = append(b.versionList, version)
+	b.versionsById[version.VersionID] = version
+
+	if b.lastAddedSchemaID != nil && version.SchemaID == *b.lastAddedSchemaID {
+		updateVersion := version.Clone()
+		updateVersion.SchemaID = LastAddedID
+		b.updates = append(b.updates, NewAddViewVersionUpdate(updateVersion))
+	} else {
+		b.updates = append(b.updates, NewAddViewVersionUpdate(version))
+	}
+
+	b.lastAddedVersionID = &newVersionID
+	return newVersionID, nil
 }
 
-func (b *MetadataBuilder) SetLocation(_ string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) AddSchema(schema *iceberg.Schema) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	_, err := b.addSchema(schema)
+	if b.setErr(err) {
+		return b
+	}
+	return b
 }
 
-func (b *MetadataBuilder) SetProperties(_ map[string]string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) addSchema(schema *iceberg.Schema) (int, error) {
+	newSchemaID := b.reuseOrCreateNewSchemaID(schema)
+
+	if _, ok := b.schemasById[newSchemaID]; ok {
+		return newSchemaID, nil
+	}
+
+	newSchema := schema
+	// Build a fresh schema if we reset the ID
+	if schema.ID != newSchemaID {
+		newSchema = iceberg.NewSchemaWithIdentifiers(newSchemaID, schema.IdentifierFieldIDs, schema.Fields()...)
+	}
+
+	b.schemaList = append(b.schemaList, newSchema)
+	b.schemasById[newSchema.ID] = newSchema
+	b.updates = append(b.updates, NewAddSchemaUpdate(newSchema))
+	b.lastAddedSchemaID = &newSchemaID
+
+	return newSchemaID, nil
 }
 
-func (b *MetadataBuilder) RemoveProperties(_ []string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) RemoveProperties(keys []string) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if len(keys) == 0 {
+		return b
+	}
+
+	b.updates = append(b.updates, NewRemovePropertiesUpdate(keys))
+	for _, key := range keys {
+		delete(b.props, key)
+	}
+
+	return b
 }
 
-func (b *MetadataBuilder) AddVersion(_ *Version) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetCurrentVersionID(newVersionID int64) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if newVersionID == LastAddedID {
+		if b.lastAddedVersionID == nil {
+			b.err = errors.New("cannot set current version to last added, no version has been added")
+			return b
+		}
+		newVersionID = *b.lastAddedVersionID
+	}
+
+	if newVersionID == b.currentVersionID {
+		return b
+	}
+
+	_, ok := b.versionsById[newVersionID]
+	if !ok {
+		b.err = fmt.Errorf("cannot set current version to unknown version with id %d", newVersionID)
+		return b
+	}
+
+	if b.lastAddedVersionID != nil && *b.lastAddedVersionID == newVersionID {
+		b.updates = append(b.updates, NewSetCurrentVersionUpdate(LastAddedID))
+	} else {
+		b.updates = append(b.updates, NewSetCurrentVersionUpdate(newVersionID))
+	}
+	b.currentVersionID = newVersionID
+
+	// Set the current history entry.
+	// If the version was added in the current set of changes, use its timestamp,
+	// otherwise use system time.
+	updateTimestampMS := time.Now().UnixMilli()
+	for _, update := range b.updates {
+		if v, ok := update.(*addViewVersionUpdate); ok && v.Version.VersionID == newVersionID {
+			updateTimestampMS = v.Version.TimestampMS
+			break
+		}
+	}
+
+	b.versionHistoryEntry = &VersionLogEntry{
+		VersionID:   newVersionID,
+		TimestampMS: updateTimestampMS,
+	}
+
+	return b
 }
 
-func (b *MetadataBuilder) SetCurrentVersionID(_ int64) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetFormatVersion(formatVersion int) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if formatVersion < b.formatVersion {
+		b.err = fmt.Errorf("downgrading format version from %d to %d is not allowed",
+			b.formatVersion, formatVersion)
+		return b
+	}
+
+	if formatVersion > SupportedViewFormatVersion {
+		b.err = fmt.Errorf("unsupported format version %d", formatVersion)
+		return b
+	}
+
+	if formatVersion == b.formatVersion {
+		return b
+	}
+
+	b.updates = append(b.updates, NewUpgradeFormatVersionUpdate(formatVersion))
+	b.formatVersion = formatVersion
+
+	return b
+}
+
+func (b *MetadataBuilder) SetLoc(loc string) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if b.loc == loc {
+		return b
+	}
+
+	b.updates = append(b.updates, NewSetLocationUpdate(loc))
+	b.loc = loc
+
+	return b
+}
+
+func (b *MetadataBuilder) SetProperties(props iceberg.Properties) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if len(props) == 0 {
+		return b
+	}
+
+	b.updates = append(b.updates, NewSetPropertiesUpdate(props))
+	if b.props == nil {
+		b.props = props
+	} else {
+		maps.Copy(b.props, props)
+	}
+
+	return b
+}
+
+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
+	}
+
+	// Noop - same UUID does not generate a change.
+	if b.uuid == newUUID {
+		return b
+	}
+
+	if b.uuid != uuid.Nil {
+		b.err = errors.New("cannot reassign uuid")
+		return b
+	}
+
+	b.updates = append(b.updates, NewAssignUUIDUpdate(newUUID))
+	b.uuid = newUUID
+
+	return b
+}
+
+func (b *MetadataBuilder) Err() error {
+	return b.err
+}
+
+func (b *MetadataBuilder) buildMetadata(withUpdates bool) (*metadata, error) {
+	uuid_ := b.uuid
+	if uuid_ == uuid.Nil {
+		uuid_ = uuid.New()
+	}
+
+	md := &metadata{
+		FormatVersionValue:    b.formatVersion,
+		UUID:                  uuid_,
+		Loc:                   b.loc,
+		CurrentVersionIDValue: b.currentVersionID,
+		VersionList:           b.versionList,
+		VersionLogList:        b.versionLog,
+		SchemaList:            b.schemaList,
+		Props:                 b.props,
+	}
+	if withUpdates {
+		md.updates = b.updates
+	}
+	md.init()
+
+	return md, nil
+}
+
+// build builds the view metadata updates from the builder
+// if withUpdates is set to false, updates will not be included
+// in the returned Metadata, thus its Updates() will return an empty slice
+func (b *MetadataBuilder) build(withUpdates bool) (Metadata, error) {

Review Comment:
   I'd keep it in here to avoid duplicating this logic across catalog implementations, the metadata builder is the shared thing amongst all of them, we're also managing history in the table metadata builder, for sake of consistency it'd also make sense to do it here



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["twuebi (via GitHub)" <gi...@apache.org>]

twuebi commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2542003627


##########
view/metadata.go:
##########
@@ -18,135 +18,523 @@
 package view
 
 import (
-	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"maps"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
 	"time"
 
 	"github.com/apache/iceberg-go"
-	"github.com/apache/iceberg-go/internal"
-	"github.com/apache/iceberg-go/io"
+	iceinternal "github.com/apache/iceberg-go/internal"
+	"github.com/apache/iceberg-go/table"
+
 	"github.com/google/uuid"
 )
 
-// LoadMetadata reads and parses a view metadata file from the specified location.
+var (
+	ErrInvalidViewMetadata              = errors.New("invalid view metadata")
+	ErrInvalidViewMetadataFormatVersion = errors.New("invalid or missing format-version in view metadata")
+)
+
+const (
+	// LastAddedID is used in place of ID fields (e.g. schema, version) to indicate that
+	// the last added instance of that type should be used.
+	LastAddedID                = -1
+	SupportedViewFormatVersion = 1
+	DefaultViewFormatVersion   = SupportedViewFormatVersion
+)
+
+const (
+	initialSchemaID = 0
+
+	viewEngineProperty = "engine-name"
+	defaultViewEngine  = "iceberg-go"
+)
+
+// Metadata for an iceberg view as specified in the Iceberg spec
+// https://iceberg.apache.org/view-spec/
+type Metadata interface {
+	// FormatVersion indicates the version of this metadata, 1 for V1
+	FormatVersion() int
+	// ViewUUID returns a UUID that identifies the view, generated when the
+	// view is created. Implementations must throw an exception if a view's
+	// UUID does not match the expected UUID after refreshing metadata.
+	ViewUUID() uuid.UUID
+	// Location is the table's base location. This is used by writers to determine
+	// where to store data files, manifest files, and table metadata files.
+	Location() string
+	// Schemas returns the list of view schemas
+	Schemas() []*iceberg.Schema
+	// CurrentVersionID returns the ID of the current version of the view (version-id)
+	CurrentVersionID() int64
+	// CurrentVersion returns the current version of the view
+	CurrentVersion() *Version
+	// CurrentSchemaID returns the ID of the current schema
+	CurrentSchemaID() int
+	// CurrentSchema returns the current schema of the view
+	CurrentSchema() *iceberg.Schema
+	// SchemasByID returns a map of schema IDs to schemas
+	SchemasByID() map[int]*iceberg.Schema
+	// Versions returns the list of view versions
+	Versions() []*Version
+	// VersionLog returns a list of version log entries
+	// with the timestamp and version-id for every change to current-version-id
+	VersionLog() []VersionLogEntry
+	// Properties is a string to string map of view properties.
+	Properties() iceberg.Properties
+
+	// Updates returns the list of metadata updates used to build this metadata
+	Updates() Updates
+
+	Equals(Metadata) bool
+}
+
+// VersionSummary is string to string map of summary metadata about a view's version
+type VersionSummary map[string]string
+
+// Representation is a struct containing information about a view's representation
+// https://iceberg.apache.org/view-spec/#sql-representation
+type Representation struct {
+	// Must be sql
+	Type string `json:"type"`
+	// A SQL SELECT statement
+	Sql string `json:"sql"`
+	// The dialect of the sql SELECT statement (e.g., "trino" or "spark")
+	Dialect string `json:"dialect"`
+}
+type Representations []Representation
+
+func NewRepresentation(sql string, dialect string) Representation {
+	return Representation{
+		Type:    "sql",
+		Sql:     sql,
+		Dialect: dialect,
+	}
+}
+
+type Version struct {
+	// ID for the version
+	VersionID int64 `json:"version-id"`
+	// ID of the schema for the view version
+	SchemaID int `json:"schema-id"`
+	// Timestamp when the version was created (ms from epoch)
+	TimestampMS int64 `json:"timestamp-ms"`
+	// A string to string map of summary metadata about the version
+	Summary VersionSummary `json:"summary"`
+	// A list of representations for the view definition
+	Representations []Representation `json:"representations"`
+	// The default view namespace (as a list of strings)
+	DefaultNamespace table.Identifier `json:"default-namespace"`
+	// An (optional) default catalog name for querying the view
+	DefaultCatalog string `json:"default-catalog,omitempty"`
+}
+
+type VersionOpt func(*Version)
+
+func WithVersionSummary(summary VersionSummary) VersionOpt {
+	return func(v *Version) {
+		if v.Summary == nil {
+			v.Summary = summary
+		} else {
+			maps.Copy(v.Summary, summary)
+		}
+	}
+}
+
+func WithDefaultViewCatalog(catalogName string) func(*Version) {
+	return func(version *Version) {
+		version.DefaultCatalog = catalogName
+	}
+}
+
+func WithTimestampMS(timestampMS int64) VersionOpt {
+	return func(version *Version) {
+		version.TimestampMS = timestampMS
+	}
+}
+
+// NewVersion creates a Version instance from the provided parameters.
+// If building updates using the MetadataBuilder, and one desires to use the last
+// added schema ID, one should use the LastAddedID constant as the provided schemaID
 //
-// Returns the loaded Metadata or an error if the file cannot be read or parsed.
-func LoadMetadata(ctx context.Context,
-	props iceberg.Properties,
-	metadataLocation string,
-	name string,
-	namespace string,
-) (_ Metadata, err error) {
-	fs, err := io.LoadFS(ctx, props, metadataLocation)
-	if err != nil {
-		return nil, fmt.Errorf("error loading view metadata: %w", err)
+// Note that NewVersion automatically seeds TimestampMS to time.Now().UnixMilli(),
+// and one should use the option WithTimestampMS to override this behavior.
+func NewVersion(id int64, schemaID int, representations []Representation, defaultNS table.Identifier, opts ...VersionOpt) (*Version, error) {
+	if id < 1 {
+		return nil, errors.New("id should be greater than 0")
 	}
 
-	inputFile, err := fs.Open(metadataLocation)
-	if err != nil {
-		return nil, fmt.Errorf("%s.%s: error encountered loading view metadata: %w", namespace, name, err)
+	if len(representations) == 0 {
+		return nil, errors.New("invalid view version: must have at least one representation")
 	}
-	defer internal.CheckedClose(inputFile, &err)
 
-	var m metadata
-	if err := json.NewDecoder(inputFile).Decode(&m); err != nil {
-		return nil, fmt.Errorf("error encountered decoding view metadata: %w", err)
+	version := &Version{
+		VersionID:        id,
+		SchemaID:         schemaID,
+		Representations:  representations,
+		Summary:          VersionSummary{viewEngineProperty: defaultViewEngine},
+		TimestampMS:      time.Now().UnixMilli(),
+		DefaultNamespace: defaultNS,
 	}
 
-	return &m, nil
+	for _, opt := range opts {
+		opt(version)
+	}
+
+	return version, nil
 }
 
-// NewMetadata returns a view metadata for a given version.
-func NewMetadata(schema *iceberg.Schema,
-	version Version,
-	loc string,
-	props iceberg.Properties,
-) (Metadata, error) {
-	timestampMs := time.Now().UnixMilli()
+// NewVersionFromSQL creates a new Version with a single representation
+// using the provided SQL and dialect "default".
+func NewVersionFromSQL(id int64, schemaID int, sql string, defaultNS table.Identifier, opts ...VersionOpt) (*Version, error) {
+	return NewVersion(id, schemaID, Representations{NewRepresentation(sql, "default")}, defaultNS, opts...)
+}
 
-	if version.SchemaID != schema.ID {
-		return nil, fmt.Errorf("%w: version.SchemaID does not match schema.ID", iceberg.ErrInvalidArgument)
+// Equals checks whether the other Version would behave the same
+// while ignoring the view version id, and the creation timestamp
+func (v *Version) Equals(other *Version) bool {
+	return v.SchemaID == other.SchemaID &&
+		v.DefaultCatalog == other.DefaultCatalog &&
+		slices.Equal(v.DefaultNamespace, other.DefaultNamespace) &&
+		maps.Equal(v.Summary, other.Summary) &&
+		slices.Equal(v.Representations, other.Representations)
+}
+
+func (v *Version) Clone() *Version {
+	if v == nil {
+		return nil
 	}
 
-	return &metadata{
-		UUID:             uuid.NewString(),
-		FmtVersion:       1,
-		Loc:              loc,
-		SchemaList:       []*iceberg.Schema{schema},
-		CurrentVersionId: version.VersionID,
-		VersionList:      []Version{version},
-		VersionLogList: []VersionLogEntry{
-			{
-				TimestampMs: timestampMs,
-				VersionID:   version.VersionID,
-			},
-		},
-		Props: props,
-	}, nil
+	cloned := *v
+	cloned.Summary = maps.Clone(v.Summary)
+	cloned.Representations = slices.Clone(v.Representations)
+	cloned.DefaultNamespace = slices.Clone(v.DefaultNamespace)
+	return &cloned
 }
 
-// CreateMetadata creates a new view metadata file and writes it to storage.
-//
-// Returns the full path to the created metadata file, or an error if creation fails.
-//
-// Note: This function only supports creating new views with format version 1.
-// It does not support updating existing view metadata.
-func CreateMetadata(
-	ctx context.Context,
-	catalogName string,
-	nsIdent []string,
-	schema *iceberg.Schema,
-	viewSQL string,
-	loc string,
-	props iceberg.Properties,
-) (metadataLocation string, err error) {
-	versionId := int64(1)
-	timestampMs := time.Now().UnixMilli()
-
-	viewVersion := Version{
-		VersionID:   versionId,
-		TimestampMs: timestampMs,
-		SchemaID:    schema.ID,
-		Summary:     map[string]string{"sql": viewSQL},
-		Representations: []SQLRepresentation{
-			{Type: "sql", SQL: viewSQL, Dialect: "default"},
-		},
-		DefaultCatalog:   &catalogName,
-		DefaultNamespace: nsIdent,
-	}
-
-	metadataLocation = loc + "/metadata/view-" + uuid.New().String() + ".metadata.json"
-	viewMetadata, err := NewMetadata(schema, viewVersion, loc, props)
+type VersionLogEntry struct {
+	// Timestamp when the view's current-version-id was updated (ms from epoch)
+	TimestampMS int64 `json:"timestamp-ms"`
+	// ID that current-version-id was set to
+	VersionID int64 `json:"version-id"`
+}
+
+// ParseMetadata parses json metadata 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 "", fmt.Errorf("failed to create view metadata: %w", err)
+		return nil, err
 	}
 
-	viewMetadataBytes, err := json.Marshal(viewMetadata)
-	if err != nil {
-		return "", fmt.Errorf("failed to marshal view metadata: %w", 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) {
+	ver := struct {
+		FormatVersion int `json:"format-version"`
+	}{}
+	if err := json.Unmarshal(b, &ver); err != nil {
+		return nil, err
 	}
 
-	fs, err := io.LoadFS(ctx, props, metadataLocation)
-	if err != nil {
-		return "", fmt.Errorf("failed to load filesystem for view metadata: %w", err)
+	var ret Metadata
+	ret = &metadata{}
+	return ret, json.Unmarshal(b, ret)
+}
+
+// indexBy indexes a slice into a map, using a provided extractKey function
+// The extractKey function will be called on each item in the slice and assign the
+// item as a value for that key in the resultant map.
+func indexBy[T any, K comparable](s []T, extractKey func(T) K) map[K]T {
+	index := make(map[K]T)
+	for _, v := range s {
+		index[extractKey(v)] = v
+	}
+	return index
+}
+
+// Cloner is an interface which implements a Clone method for deep copying itself
+type cloner[T any] interface {
+	// Clone returns a deep copy of the underlying object
+	Clone() T
+}
+
+// cloneSlice returns a deep-clone of a Slice of elements implementing Cloner
+func cloneSlice[T cloner[T]](val []T) []T {
+	cloned := make([]T, len(val))
+	for i, elem := range val {
+		cloned[i] = elem.Clone()
+	}
+	return cloned
+}
+
+// https://iceberg.apache.org/view-spec/
+type metadata struct {
+	FormatVersionValue    int                `json:"format-version"`
+	UUID                  uuid.UUID          `json:"view-uuid"`
+	Loc                   string             `json:"location"`
+	CurrentVersionIDValue int64              `json:"current-version-id"`
+	VersionList           []*Version         `json:"versions"`
+	SchemaList            []*iceberg.Schema  `json:"schemas"`
+	VersionLogList        []VersionLogEntry  `json:"version-log"`
+	Props                 iceberg.Properties `json:"properties,omitempty"`
+
+	// updates from builder if applicable
+	updates Updates
+
+	// cached lookup helpers, must be initialized in init()
+	lazyVersionsByID func() map[int64]*Version
+	lazySchemasByID  func() map[int]*iceberg.Schema
+}
+
+func (m *metadata) Equals(other Metadata) bool {
+	if other == nil {
+		return false
+	}
+
+	if m == other {
+		return true
+	}
+
+	return m.UUID == other.ViewUUID() &&
+		m.Loc == other.Location() &&
+		m.FormatVersionValue == other.FormatVersion() &&
+		iceinternal.SliceEqualHelper(m.Schemas(), other.Schemas()) &&
+		iceinternal.SliceEqualHelper(m.Versions(), other.Versions()) &&
+		m.CurrentVersionIDValue == other.CurrentVersionID() &&
+		slices.Equal(m.VersionLogList, other.VersionLog())
+}
+
+func (m *metadata) Updates() Updates { return m.updates }
+
+func (m *metadata) FormatVersion() int         { return m.FormatVersionValue }
+func (m *metadata) ViewUUID() uuid.UUID        { return m.UUID }
+func (m *metadata) Location() string           { return m.Loc }
+func (m *metadata) Versions() []*Version       { return m.VersionList }
+func (m *metadata) Schemas() []*iceberg.Schema { return m.SchemaList }
+func (m *metadata) SchemasByID() map[int]*iceberg.Schema {
+	return maps.Clone(m.lazySchemasByID())
+}
+
+func (m *metadata) CurrentVersionID() int64 {
+	return m.CurrentVersionIDValue
+}
+
+func (m *metadata) CurrentVersion() *Version {
+	version, ok := m.lazyVersionsByID()[m.CurrentVersionIDValue]
+	if !ok {
+		panic("current version not found")
 	}

Review Comment:
   Is this always a bug in iceberg-go if we arrive here? Can a user of the library provoke this panic? If so, we should either return an error or nil, we should avoid runtime panics



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2542976387


##########
view/metadata.go:
##########
@@ -18,135 +18,523 @@
 package view
 
 import (
-	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"maps"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
 	"time"
 
 	"github.com/apache/iceberg-go"
-	"github.com/apache/iceberg-go/internal"
-	"github.com/apache/iceberg-go/io"
+	iceinternal "github.com/apache/iceberg-go/internal"
+	"github.com/apache/iceberg-go/table"
+
 	"github.com/google/uuid"
 )
 
-// LoadMetadata reads and parses a view metadata file from the specified location.
+var (
+	ErrInvalidViewMetadata              = errors.New("invalid view metadata")
+	ErrInvalidViewMetadataFormatVersion = errors.New("invalid or missing format-version in view metadata")
+)
+
+const (
+	// LastAddedID is used in place of ID fields (e.g. schema, version) to indicate that
+	// the last added instance of that type should be used.
+	LastAddedID                = -1
+	SupportedViewFormatVersion = 1
+	DefaultViewFormatVersion   = SupportedViewFormatVersion
+)
+
+const (
+	initialSchemaID = 0
+
+	viewEngineProperty = "engine-name"
+	defaultViewEngine  = "iceberg-go"
+)
+
+// Metadata for an iceberg view as specified in the Iceberg spec
+// https://iceberg.apache.org/view-spec/
+type Metadata interface {
+	// FormatVersion indicates the version of this metadata, 1 for V1
+	FormatVersion() int
+	// ViewUUID returns a UUID that identifies the view, generated when the
+	// view is created. Implementations must throw an exception if a view's
+	// UUID does not match the expected UUID after refreshing metadata.
+	ViewUUID() uuid.UUID
+	// Location is the table's base location. This is used by writers to determine
+	// where to store data files, manifest files, and table metadata files.
+	Location() string
+	// Schemas returns the list of view schemas
+	Schemas() []*iceberg.Schema
+	// CurrentVersionID returns the ID of the current version of the view (version-id)
+	CurrentVersionID() int64
+	// CurrentVersion returns the current version of the view
+	CurrentVersion() *Version
+	// CurrentSchemaID returns the ID of the current schema
+	CurrentSchemaID() int
+	// CurrentSchema returns the current schema of the view
+	CurrentSchema() *iceberg.Schema
+	// SchemasByID returns a map of schema IDs to schemas
+	SchemasByID() map[int]*iceberg.Schema
+	// Versions returns the list of view versions
+	Versions() []*Version
+	// VersionLog returns a list of version log entries
+	// with the timestamp and version-id for every change to current-version-id
+	VersionLog() []VersionLogEntry
+	// Properties is a string to string map of view properties.
+	Properties() iceberg.Properties
+
+	// Updates returns the list of metadata updates used to build this metadata
+	Updates() Updates
+
+	Equals(Metadata) bool
+}
+
+// VersionSummary is string to string map of summary metadata about a view's version
+type VersionSummary map[string]string
+
+// Representation is a struct containing information about a view's representation
+// https://iceberg.apache.org/view-spec/#sql-representation
+type Representation struct {
+	// Must be sql
+	Type string `json:"type"`
+	// A SQL SELECT statement
+	Sql string `json:"sql"`
+	// The dialect of the sql SELECT statement (e.g., "trino" or "spark")
+	Dialect string `json:"dialect"`
+}
+type Representations []Representation
+
+func NewRepresentation(sql string, dialect string) Representation {
+	return Representation{
+		Type:    "sql",
+		Sql:     sql,
+		Dialect: dialect,
+	}
+}
+
+type Version struct {
+	// ID for the version
+	VersionID int64 `json:"version-id"`
+	// ID of the schema for the view version
+	SchemaID int `json:"schema-id"`
+	// Timestamp when the version was created (ms from epoch)
+	TimestampMS int64 `json:"timestamp-ms"`
+	// A string to string map of summary metadata about the version
+	Summary VersionSummary `json:"summary"`
+	// A list of representations for the view definition
+	Representations []Representation `json:"representations"`
+	// The default view namespace (as a list of strings)
+	DefaultNamespace table.Identifier `json:"default-namespace"`
+	// An (optional) default catalog name for querying the view
+	DefaultCatalog string `json:"default-catalog,omitempty"`
+}
+
+type VersionOpt func(*Version)
+
+func WithVersionSummary(summary VersionSummary) VersionOpt {
+	return func(v *Version) {
+		if v.Summary == nil {
+			v.Summary = summary
+		} else {
+			maps.Copy(v.Summary, summary)
+		}
+	}
+}
+
+func WithDefaultViewCatalog(catalogName string) func(*Version) {
+	return func(version *Version) {
+		version.DefaultCatalog = catalogName
+	}
+}
+
+func WithTimestampMS(timestampMS int64) VersionOpt {
+	return func(version *Version) {
+		version.TimestampMS = timestampMS
+	}
+}
+
+// NewVersion creates a Version instance from the provided parameters.
+// If building updates using the MetadataBuilder, and one desires to use the last
+// added schema ID, one should use the LastAddedID constant as the provided schemaID
 //
-// Returns the loaded Metadata or an error if the file cannot be read or parsed.
-func LoadMetadata(ctx context.Context,
-	props iceberg.Properties,
-	metadataLocation string,
-	name string,
-	namespace string,
-) (_ Metadata, err error) {
-	fs, err := io.LoadFS(ctx, props, metadataLocation)
-	if err != nil {
-		return nil, fmt.Errorf("error loading view metadata: %w", err)
+// Note that NewVersion automatically seeds TimestampMS to time.Now().UnixMilli(),
+// and one should use the option WithTimestampMS to override this behavior.
+func NewVersion(id int64, schemaID int, representations []Representation, defaultNS table.Identifier, opts ...VersionOpt) (*Version, error) {
+	if id < 1 {
+		return nil, errors.New("id should be greater than 0")
 	}
 
-	inputFile, err := fs.Open(metadataLocation)
-	if err != nil {
-		return nil, fmt.Errorf("%s.%s: error encountered loading view metadata: %w", namespace, name, err)
+	if len(representations) == 0 {
+		return nil, errors.New("invalid view version: must have at least one representation")
 	}
-	defer internal.CheckedClose(inputFile, &err)
 
-	var m metadata
-	if err := json.NewDecoder(inputFile).Decode(&m); err != nil {
-		return nil, fmt.Errorf("error encountered decoding view metadata: %w", err)
+	version := &Version{
+		VersionID:        id,
+		SchemaID:         schemaID,
+		Representations:  representations,
+		Summary:          VersionSummary{viewEngineProperty: defaultViewEngine},
+		TimestampMS:      time.Now().UnixMilli(),
+		DefaultNamespace: defaultNS,
 	}
 
-	return &m, nil
+	for _, opt := range opts {
+		opt(version)
+	}
+
+	return version, nil
 }
 
-// NewMetadata returns a view metadata for a given version.
-func NewMetadata(schema *iceberg.Schema,
-	version Version,
-	loc string,
-	props iceberg.Properties,
-) (Metadata, error) {
-	timestampMs := time.Now().UnixMilli()
+// NewVersionFromSQL creates a new Version with a single representation
+// using the provided SQL and dialect "default".
+func NewVersionFromSQL(id int64, schemaID int, sql string, defaultNS table.Identifier, opts ...VersionOpt) (*Version, error) {
+	return NewVersion(id, schemaID, Representations{NewRepresentation(sql, "default")}, defaultNS, opts...)
+}
 
-	if version.SchemaID != schema.ID {
-		return nil, fmt.Errorf("%w: version.SchemaID does not match schema.ID", iceberg.ErrInvalidArgument)
+// Equals checks whether the other Version would behave the same
+// while ignoring the view version id, and the creation timestamp
+func (v *Version) Equals(other *Version) bool {
+	return v.SchemaID == other.SchemaID &&
+		v.DefaultCatalog == other.DefaultCatalog &&
+		slices.Equal(v.DefaultNamespace, other.DefaultNamespace) &&
+		maps.Equal(v.Summary, other.Summary) &&
+		slices.Equal(v.Representations, other.Representations)
+}
+
+func (v *Version) Clone() *Version {
+	if v == nil {
+		return nil
 	}
 
-	return &metadata{
-		UUID:             uuid.NewString(),
-		FmtVersion:       1,
-		Loc:              loc,
-		SchemaList:       []*iceberg.Schema{schema},
-		CurrentVersionId: version.VersionID,
-		VersionList:      []Version{version},
-		VersionLogList: []VersionLogEntry{
-			{
-				TimestampMs: timestampMs,
-				VersionID:   version.VersionID,
-			},
-		},
-		Props: props,
-	}, nil
+	cloned := *v
+	cloned.Summary = maps.Clone(v.Summary)
+	cloned.Representations = slices.Clone(v.Representations)
+	cloned.DefaultNamespace = slices.Clone(v.DefaultNamespace)
+	return &cloned
 }
 
-// CreateMetadata creates a new view metadata file and writes it to storage.
-//
-// Returns the full path to the created metadata file, or an error if creation fails.
-//
-// Note: This function only supports creating new views with format version 1.
-// It does not support updating existing view metadata.
-func CreateMetadata(
-	ctx context.Context,
-	catalogName string,
-	nsIdent []string,
-	schema *iceberg.Schema,
-	viewSQL string,
-	loc string,
-	props iceberg.Properties,
-) (metadataLocation string, err error) {
-	versionId := int64(1)
-	timestampMs := time.Now().UnixMilli()
-
-	viewVersion := Version{
-		VersionID:   versionId,
-		TimestampMs: timestampMs,
-		SchemaID:    schema.ID,
-		Summary:     map[string]string{"sql": viewSQL},
-		Representations: []SQLRepresentation{
-			{Type: "sql", SQL: viewSQL, Dialect: "default"},
-		},
-		DefaultCatalog:   &catalogName,
-		DefaultNamespace: nsIdent,
-	}
-
-	metadataLocation = loc + "/metadata/view-" + uuid.New().String() + ".metadata.json"
-	viewMetadata, err := NewMetadata(schema, viewVersion, loc, props)
+type VersionLogEntry struct {
+	// Timestamp when the view's current-version-id was updated (ms from epoch)
+	TimestampMS int64 `json:"timestamp-ms"`
+	// ID that current-version-id was set to
+	VersionID int64 `json:"version-id"`
+}
+
+// ParseMetadata parses json metadata 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 "", fmt.Errorf("failed to create view metadata: %w", err)
+		return nil, err
 	}
 
-	viewMetadataBytes, err := json.Marshal(viewMetadata)
-	if err != nil {
-		return "", fmt.Errorf("failed to marshal view metadata: %w", 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) {
+	ver := struct {
+		FormatVersion int `json:"format-version"`
+	}{}
+	if err := json.Unmarshal(b, &ver); err != nil {
+		return nil, err
 	}
 
-	fs, err := io.LoadFS(ctx, props, metadataLocation)
-	if err != nil {
-		return "", fmt.Errorf("failed to load filesystem for view metadata: %w", err)
+	var ret Metadata
+	ret = &metadata{}
+	return ret, json.Unmarshal(b, ret)
+}
+
+// indexBy indexes a slice into a map, using a provided extractKey function
+// The extractKey function will be called on each item in the slice and assign the
+// item as a value for that key in the resultant map.
+func indexBy[T any, K comparable](s []T, extractKey func(T) K) map[K]T {
+	index := make(map[K]T)
+	for _, v := range s {
+		index[extractKey(v)] = v
+	}
+	return index
+}
+
+// Cloner is an interface which implements a Clone method for deep copying itself
+type cloner[T any] interface {
+	// Clone returns a deep copy of the underlying object
+	Clone() T
+}
+
+// cloneSlice returns a deep-clone of a Slice of elements implementing Cloner
+func cloneSlice[T cloner[T]](val []T) []T {
+	cloned := make([]T, len(val))
+	for i, elem := range val {
+		cloned[i] = elem.Clone()
+	}
+	return cloned
+}
+
+// https://iceberg.apache.org/view-spec/
+type metadata struct {
+	FormatVersionValue    int                `json:"format-version"`
+	UUID                  uuid.UUID          `json:"view-uuid"`
+	Loc                   string             `json:"location"`
+	CurrentVersionIDValue int64              `json:"current-version-id"`
+	VersionList           []*Version         `json:"versions"`
+	SchemaList            []*iceberg.Schema  `json:"schemas"`
+	VersionLogList        []VersionLogEntry  `json:"version-log"`
+	Props                 iceberg.Properties `json:"properties,omitempty"`
+
+	// updates from builder if applicable
+	updates Updates
+
+	// cached lookup helpers, must be initialized in init()
+	lazyVersionsByID func() map[int64]*Version
+	lazySchemasByID  func() map[int]*iceberg.Schema
+}
+
+func (m *metadata) Equals(other Metadata) bool {
+	if other == nil {
+		return false
+	}
+
+	if m == other {
+		return true
+	}
+
+	return m.UUID == other.ViewUUID() &&
+		m.Loc == other.Location() &&
+		m.FormatVersionValue == other.FormatVersion() &&
+		iceinternal.SliceEqualHelper(m.Schemas(), other.Schemas()) &&
+		iceinternal.SliceEqualHelper(m.Versions(), other.Versions()) &&
+		m.CurrentVersionIDValue == other.CurrentVersionID() &&
+		slices.Equal(m.VersionLogList, other.VersionLog())
+}
+
+func (m *metadata) Updates() Updates { return m.updates }
+
+func (m *metadata) FormatVersion() int         { return m.FormatVersionValue }
+func (m *metadata) ViewUUID() uuid.UUID        { return m.UUID }
+func (m *metadata) Location() string           { return m.Loc }
+func (m *metadata) Versions() []*Version       { return m.VersionList }
+func (m *metadata) Schemas() []*iceberg.Schema { return m.SchemaList }
+func (m *metadata) SchemasByID() map[int]*iceberg.Schema {
+	return maps.Clone(m.lazySchemasByID())
+}
+
+func (m *metadata) CurrentVersionID() int64 {
+	return m.CurrentVersionIDValue
+}
+
+func (m *metadata) CurrentVersion() *Version {
+	version, ok := m.lazyVersionsByID()[m.CurrentVersionIDValue]
+	if !ok {
+		panic("current version not found")
 	}
+	return version
+}
+
+func (m *metadata) CurrentSchemaID() int {
+	return m.CurrentVersion().SchemaID
+}
 
-	wfs, ok := fs.(io.WriteFileIO)
+func (m *metadata) CurrentSchema() *iceberg.Schema {
+	schema, ok := m.lazySchemasByID()[m.CurrentSchemaID()]
 	if !ok {
-		return "", errors.New("filesystem IO does not support writing")
+		panic("current schema not found")
+	}
+	return schema
+}
+
+func (m *metadata) VersionLog() []VersionLogEntry {
+	return m.VersionLogList
+}
+
+func (m *metadata) Properties() iceberg.Properties {
+	return m.Props
+}
+
+func (m *metadata) validate() error {
+	if m.Loc == "" {
+		return fmt.Errorf("%w: location is required", ErrInvalidViewMetadata)
+	}
+
+	if m.UUID == uuid.Nil {
+		return fmt.Errorf("%w: view-uuid is required", ErrInvalidViewMetadata)
+	}

Review Comment:
   Update - just turned it into a pointer so we can distinguish between unset vs "000..." on deserialization



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["twuebi (via GitHub)" <gi...@apache.org>]

twuebi commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2543694646


##########
view/metadata_builder.go:
##########
@@ -1,57 +1,468 @@
-// 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 view
 
 import (
+	"errors"
+	"fmt"
+	"maps"
+	"slices"
+	"strings"
+	"time"
+
 	"github.com/apache/iceberg-go"
+
+	"github.com/google/uuid"
 )
 
-// MetadataBuilder is a struct used for building and updating Iceberg view metadata.
-type MetadataBuilder struct{}
+type MetadataBuilder struct {
+	base    Metadata
+	updates Updates
+
+	// common fields
+	formatVersion    int
+	uuid             uuid.UUID
+	loc              string
+	schemaList       []*iceberg.Schema
+	versionList      []*Version
+	currentVersionID int64
+	versionLog       []VersionLogEntry
+	props            iceberg.Properties
+
+	// lookup maps
+	versionsById map[int64]*Version
+	schemasById  map[int]*iceberg.Schema
+
+	// update tracking
+	versionHistoryEntry *VersionLogEntry
+	lastAddedVersionID  *int64
+	lastAddedSchemaID   *int
+
+	// error tracking for build chaining
+	// if set, subsequent operations become a noop
+	err error
+}
+
+func NewMetadataBuilder() (*MetadataBuilder, error) {
+	return &MetadataBuilder{
+		formatVersion: DefaultViewFormatVersion,
+		updates:       make([]Update, 0),
+		schemaList:    make([]*iceberg.Schema, 0),
+		versionList:   make([]*Version, 0),
+		versionLog:    make([]VersionLogEntry, 0),
+		props:         make(iceberg.Properties),
+		versionsById:  make(map[int64]*Version),
+		schemasById:   make(map[int]*iceberg.Schema),
+	}, nil
+}
+
+func MetadataBuilderFromBase(metadata Metadata) (*MetadataBuilder, error) {
+	b := &MetadataBuilder{}
+	b.base = metadata
+
+	// Copy fields
+	b.formatVersion = metadata.FormatVersion()
+	b.uuid = metadata.ViewUUID()
+	b.loc = metadata.Location()
+	b.schemaList = slices.Clone(metadata.Schemas())
+	b.currentVersionID = metadata.CurrentVersionID()
+	b.versionList = cloneSlice(metadata.Versions())
+	b.versionLog = slices.Clone(metadata.VersionLog())
+	b.props = maps.Clone(metadata.Properties())
+
+	// Build lookup maps
+	b.versionsById = indexBy(b.versionList, func(vl *Version) int64 { return vl.VersionID })
+	b.schemasById = indexBy(b.schemaList, func(sm *iceberg.Schema) int { return sm.ID })
+
+	return b, nil
+}
+
+func (b *MetadataBuilder) HasChanges() bool { return len(b.updates) > 0 }
 
-func (b *MetadataBuilder) AssignUUID(_ string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetCurrentVersion(version *Version, schema *iceberg.Schema) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	newSchemaID, err := b.addSchema(schema)
+	if b.setErr(err) {
+		return b
+	}
+
+	newVersion := version.Clone()
+	newVersion.SchemaID = newSchemaID
+	newVersionID, err := b.addVersion(newVersion)
+	if b.setErr(err) {
+		return b
+	}
+
+	return b.SetCurrentVersionID(newVersionID)
 }
 
-func (b *MetadataBuilder) UpgradeFormatVersion(_ int) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) AddVersion(newVersion *Version) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	_, err := b.addVersion(newVersion)
+	if b.setErr(err) {
+		return b
+	}
+	return b
 }
 
-func (b *MetadataBuilder) AddSchema(_ *iceberg.Schema) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) addVersion(newVersion *Version) (int64, error) {
+	newVersionID := b.reuseOrCreateNewVersionID(newVersion)
+	version := newVersion.Clone()
+	if newVersionID != version.VersionID {
+		version.VersionID = newVersionID
+	}
+
+	// Check if this version was added in an update already
+	if _, ok := b.versionsById[newVersionID]; ok {
+		for _, upd := range b.updates {
+			if vvu, ok := upd.(*addViewVersionUpdate); ok && vvu.Version.VersionID == newVersionID {
+				b.lastAddedVersionID = &newVersionID
+				return newVersionID, nil
+			}
+		}
+	}
+
+	// If the SchemaID of the version is unset (nil), we attach the lastAddedSchemaID to it.
+	// If we have no lastAddedSchemaID, we fail
+	if version.SchemaID == LastAddedID {
+		if b.lastAddedSchemaID == nil {
+			return 0, errors.New("cannot set last added schema: no schema has been added")
+		}
+		version.SchemaID = *b.lastAddedSchemaID
+	}
+
+	if _, ok := b.schemasById[version.SchemaID]; !ok {
+		return 0, fmt.Errorf("cannot add version with unknown schema: %d", version.SchemaID)
+	}
+
+	if len(version.Representations) == 0 {
+		return 0, fmt.Errorf("cannot add version with no representations")
+	}
+	dialects := make(map[string]struct{})
+	for _, repr := range version.Representations {
+		normalizedDialect := strings.ToLower(repr.Dialect)
+		if _, ok := dialects[normalizedDialect]; ok {
+			return 0, fmt.Errorf("invalid view version: cannot add multiple queries for dialect %s", normalizedDialect)
+		}
+		dialects[normalizedDialect] = struct{}{}
+	}
+
+	b.versionList = append(b.versionList, version)
+	b.versionsById[version.VersionID] = version
+
+	if b.lastAddedSchemaID != nil && version.SchemaID == *b.lastAddedSchemaID {
+		updateVersion := version.Clone()
+		updateVersion.SchemaID = LastAddedID
+		b.updates = append(b.updates, NewAddViewVersionUpdate(updateVersion))
+	} else {
+		b.updates = append(b.updates, NewAddViewVersionUpdate(version))
+	}
+
+	b.lastAddedVersionID = &newVersionID
+	return newVersionID, nil
 }
 
-func (b *MetadataBuilder) SetLocation(_ string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) AddSchema(schema *iceberg.Schema) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	_, err := b.addSchema(schema)
+	if b.setErr(err) {
+		return b
+	}
+	return b
 }
 
-func (b *MetadataBuilder) SetProperties(_ map[string]string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) addSchema(schema *iceberg.Schema) (int, error) {
+	newSchemaID := b.reuseOrCreateNewSchemaID(schema)
+
+	if _, ok := b.schemasById[newSchemaID]; ok {
+		return newSchemaID, nil
+	}
+
+	newSchema := schema
+	// Build a fresh schema if we reset the ID
+	if schema.ID != newSchemaID {
+		newSchema = iceberg.NewSchemaWithIdentifiers(newSchemaID, schema.IdentifierFieldIDs, schema.Fields()...)
+	}
+
+	b.schemaList = append(b.schemaList, newSchema)
+	b.schemasById[newSchema.ID] = newSchema
+	b.updates = append(b.updates, NewAddSchemaUpdate(newSchema))
+	b.lastAddedSchemaID = &newSchemaID
+
+	return newSchemaID, nil
 }
 
-func (b *MetadataBuilder) RemoveProperties(_ []string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) RemoveProperties(keys []string) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if len(keys) == 0 {
+		return b
+	}
+
+	b.updates = append(b.updates, NewRemovePropertiesUpdate(keys))
+	for _, key := range keys {
+		delete(b.props, key)
+	}
+
+	return b
 }
 
-func (b *MetadataBuilder) AddVersion(_ *Version) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetCurrentVersionID(newVersionID int64) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if newVersionID == LastAddedID {
+		if b.lastAddedVersionID == nil {
+			b.err = errors.New("cannot set current version to last added, no version has been added")
+			return b
+		}
+		newVersionID = *b.lastAddedVersionID
+	}
+
+	if newVersionID == b.currentVersionID {
+		return b
+	}
+
+	_, ok := b.versionsById[newVersionID]
+	if !ok {
+		b.err = fmt.Errorf("cannot set current version to unknown version with id %d", newVersionID)
+		return b
+	}
+
+	if b.lastAddedVersionID != nil && *b.lastAddedVersionID == newVersionID {
+		b.updates = append(b.updates, NewSetCurrentVersionUpdate(LastAddedID))
+	} else {
+		b.updates = append(b.updates, NewSetCurrentVersionUpdate(newVersionID))
+	}
+	b.currentVersionID = newVersionID
+
+	// Set the current history entry.
+	// If the version was added in the current set of changes, use its timestamp,
+	// otherwise use system time.
+	updateTimestampMS := time.Now().UnixMilli()
+	for _, update := range b.updates {
+		if v, ok := update.(*addViewVersionUpdate); ok && v.Version.VersionID == newVersionID {
+			updateTimestampMS = v.Version.TimestampMS
+			break
+		}
+	}
+
+	b.versionHistoryEntry = &VersionLogEntry{
+		VersionID:   newVersionID,
+		TimestampMS: updateTimestampMS,
+	}
+
+	return b
 }
 
-func (b *MetadataBuilder) SetCurrentVersionID(_ int64) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetFormatVersion(formatVersion int) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if formatVersion < b.formatVersion {
+		b.err = fmt.Errorf("downgrading format version from %d to %d is not allowed",
+			b.formatVersion, formatVersion)
+		return b
+	}
+
+	if formatVersion > SupportedViewFormatVersion {
+		b.err = fmt.Errorf("unsupported format version %d", formatVersion)
+		return b
+	}
+
+	if formatVersion == b.formatVersion {
+		return b
+	}
+
+	b.updates = append(b.updates, NewUpgradeFormatVersionUpdate(formatVersion))
+	b.formatVersion = formatVersion
+
+	return b
+}
+
+func (b *MetadataBuilder) SetLoc(loc string) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if b.loc == loc {
+		return b
+	}
+
+	b.updates = append(b.updates, NewSetLocationUpdate(loc))
+	b.loc = loc
+
+	return b
+}
+
+func (b *MetadataBuilder) SetProperties(props iceberg.Properties) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if len(props) == 0 {
+		return b
+	}
+
+	b.updates = append(b.updates, NewSetPropertiesUpdate(props))
+	if b.props == nil {
+		b.props = props
+	} else {
+		maps.Copy(b.props, props)
+	}
+
+	return b
+}
+
+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
+	}
+
+	// Noop - same UUID does not generate a change.
+	if b.uuid == newUUID {
+		return b
+	}
+
+	if b.uuid != uuid.Nil {
+		b.err = errors.New("cannot reassign uuid")
+		return b
+	}
+
+	b.updates = append(b.updates, NewAssignUUIDUpdate(newUUID))
+	b.uuid = newUUID
+
+	return b
+}
+
+func (b *MetadataBuilder) Err() error {
+	return b.err
+}
+
+func (b *MetadataBuilder) buildMetadata(withUpdates bool) (*metadata, error) {
+	uuid_ := b.uuid
+	if uuid_ == uuid.Nil {
+		uuid_ = uuid.New()
+	}
+
+	md := &metadata{
+		FormatVersionValue:    b.formatVersion,
+		UUID:                  uuid_,
+		Loc:                   b.loc,
+		CurrentVersionIDValue: b.currentVersionID,
+		VersionList:           b.versionList,
+		VersionLogList:        b.versionLog,
+		SchemaList:            b.schemaList,
+		Props:                 b.props,
+	}
+	if withUpdates {
+		md.updates = b.updates
+	}
+	md.init()
+
+	return md, nil
+}
+
+// build builds the view metadata updates from the builder
+// if withUpdates is set to false, updates will not be included
+// in the returned Metadata, thus its Updates() will return an empty slice
+func (b *MetadataBuilder) build(withUpdates bool) (Metadata, error) {

Review Comment:
   I'd keep it, they also apply to non-rest catalogs, there this may be the only place this is validated, I'd try to avoid reimplementing these checks in all catalogs



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2542831824


##########
view/metadata.go:
##########
@@ -18,135 +18,523 @@
 package view
 
 import (
-	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"maps"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
 	"time"
 
 	"github.com/apache/iceberg-go"
-	"github.com/apache/iceberg-go/internal"
-	"github.com/apache/iceberg-go/io"
+	iceinternal "github.com/apache/iceberg-go/internal"
+	"github.com/apache/iceberg-go/table"
+
 	"github.com/google/uuid"
 )
 
-// LoadMetadata reads and parses a view metadata file from the specified location.
+var (
+	ErrInvalidViewMetadata              = errors.New("invalid view metadata")
+	ErrInvalidViewMetadataFormatVersion = errors.New("invalid or missing format-version in view metadata")
+)
+
+const (
+	// LastAddedID is used in place of ID fields (e.g. schema, version) to indicate that
+	// the last added instance of that type should be used.
+	LastAddedID                = -1
+	SupportedViewFormatVersion = 1
+	DefaultViewFormatVersion   = SupportedViewFormatVersion
+)
+
+const (
+	initialSchemaID = 0
+
+	viewEngineProperty = "engine-name"
+	defaultViewEngine  = "iceberg-go"
+)
+
+// Metadata for an iceberg view as specified in the Iceberg spec
+// https://iceberg.apache.org/view-spec/
+type Metadata interface {
+	// FormatVersion indicates the version of this metadata, 1 for V1
+	FormatVersion() int
+	// ViewUUID returns a UUID that identifies the view, generated when the
+	// view is created. Implementations must throw an exception if a view's
+	// UUID does not match the expected UUID after refreshing metadata.
+	ViewUUID() uuid.UUID
+	// Location is the table's base location. This is used by writers to determine
+	// where to store data files, manifest files, and table metadata files.
+	Location() string
+	// Schemas returns the list of view schemas
+	Schemas() []*iceberg.Schema
+	// CurrentVersionID returns the ID of the current version of the view (version-id)
+	CurrentVersionID() int64
+	// CurrentVersion returns the current version of the view
+	CurrentVersion() *Version
+	// CurrentSchemaID returns the ID of the current schema
+	CurrentSchemaID() int
+	// CurrentSchema returns the current schema of the view
+	CurrentSchema() *iceberg.Schema
+	// SchemasByID returns a map of schema IDs to schemas
+	SchemasByID() map[int]*iceberg.Schema
+	// Versions returns the list of view versions
+	Versions() []*Version
+	// VersionLog returns a list of version log entries
+	// with the timestamp and version-id for every change to current-version-id
+	VersionLog() []VersionLogEntry
+	// Properties is a string to string map of view properties.
+	Properties() iceberg.Properties
+
+	// Updates returns the list of metadata updates used to build this metadata
+	Updates() Updates
+
+	Equals(Metadata) bool
+}
+
+// VersionSummary is string to string map of summary metadata about a view's version
+type VersionSummary map[string]string
+
+// Representation is a struct containing information about a view's representation
+// https://iceberg.apache.org/view-spec/#sql-representation
+type Representation struct {
+	// Must be sql
+	Type string `json:"type"`
+	// A SQL SELECT statement
+	Sql string `json:"sql"`
+	// The dialect of the sql SELECT statement (e.g., "trino" or "spark")
+	Dialect string `json:"dialect"`
+}
+type Representations []Representation
+
+func NewRepresentation(sql string, dialect string) Representation {
+	return Representation{
+		Type:    "sql",
+		Sql:     sql,
+		Dialect: dialect,
+	}
+}
+
+type Version struct {
+	// ID for the version
+	VersionID int64 `json:"version-id"`
+	// ID of the schema for the view version
+	SchemaID int `json:"schema-id"`
+	// Timestamp when the version was created (ms from epoch)
+	TimestampMS int64 `json:"timestamp-ms"`
+	// A string to string map of summary metadata about the version
+	Summary VersionSummary `json:"summary"`
+	// A list of representations for the view definition
+	Representations []Representation `json:"representations"`
+	// The default view namespace (as a list of strings)
+	DefaultNamespace table.Identifier `json:"default-namespace"`
+	// An (optional) default catalog name for querying the view
+	DefaultCatalog string `json:"default-catalog,omitempty"`
+}
+
+type VersionOpt func(*Version)
+
+func WithVersionSummary(summary VersionSummary) VersionOpt {
+	return func(v *Version) {
+		if v.Summary == nil {
+			v.Summary = summary
+		} else {
+			maps.Copy(v.Summary, summary)
+		}
+	}
+}
+
+func WithDefaultViewCatalog(catalogName string) func(*Version) {
+	return func(version *Version) {
+		version.DefaultCatalog = catalogName
+	}
+}
+
+func WithTimestampMS(timestampMS int64) VersionOpt {
+	return func(version *Version) {
+		version.TimestampMS = timestampMS
+	}
+}
+
+// NewVersion creates a Version instance from the provided parameters.
+// If building updates using the MetadataBuilder, and one desires to use the last
+// added schema ID, one should use the LastAddedID constant as the provided schemaID
 //
-// Returns the loaded Metadata or an error if the file cannot be read or parsed.
-func LoadMetadata(ctx context.Context,
-	props iceberg.Properties,
-	metadataLocation string,
-	name string,
-	namespace string,
-) (_ Metadata, err error) {
-	fs, err := io.LoadFS(ctx, props, metadataLocation)
-	if err != nil {
-		return nil, fmt.Errorf("error loading view metadata: %w", err)
+// Note that NewVersion automatically seeds TimestampMS to time.Now().UnixMilli(),
+// and one should use the option WithTimestampMS to override this behavior.
+func NewVersion(id int64, schemaID int, representations []Representation, defaultNS table.Identifier, opts ...VersionOpt) (*Version, error) {
+	if id < 1 {
+		return nil, errors.New("id should be greater than 0")
 	}
 
-	inputFile, err := fs.Open(metadataLocation)
-	if err != nil {
-		return nil, fmt.Errorf("%s.%s: error encountered loading view metadata: %w", namespace, name, err)
+	if len(representations) == 0 {
+		return nil, errors.New("invalid view version: must have at least one representation")
 	}
-	defer internal.CheckedClose(inputFile, &err)
 
-	var m metadata
-	if err := json.NewDecoder(inputFile).Decode(&m); err != nil {
-		return nil, fmt.Errorf("error encountered decoding view metadata: %w", err)
+	version := &Version{
+		VersionID:        id,
+		SchemaID:         schemaID,
+		Representations:  representations,
+		Summary:          VersionSummary{viewEngineProperty: defaultViewEngine},
+		TimestampMS:      time.Now().UnixMilli(),
+		DefaultNamespace: defaultNS,
 	}
 
-	return &m, nil
+	for _, opt := range opts {
+		opt(version)
+	}
+
+	return version, nil
 }
 
-// NewMetadata returns a view metadata for a given version.
-func NewMetadata(schema *iceberg.Schema,
-	version Version,
-	loc string,
-	props iceberg.Properties,
-) (Metadata, error) {
-	timestampMs := time.Now().UnixMilli()
+// NewVersionFromSQL creates a new Version with a single representation
+// using the provided SQL and dialect "default".
+func NewVersionFromSQL(id int64, schemaID int, sql string, defaultNS table.Identifier, opts ...VersionOpt) (*Version, error) {
+	return NewVersion(id, schemaID, Representations{NewRepresentation(sql, "default")}, defaultNS, opts...)
+}
 
-	if version.SchemaID != schema.ID {
-		return nil, fmt.Errorf("%w: version.SchemaID does not match schema.ID", iceberg.ErrInvalidArgument)
+// Equals checks whether the other Version would behave the same
+// while ignoring the view version id, and the creation timestamp
+func (v *Version) Equals(other *Version) bool {
+	return v.SchemaID == other.SchemaID &&
+		v.DefaultCatalog == other.DefaultCatalog &&
+		slices.Equal(v.DefaultNamespace, other.DefaultNamespace) &&
+		maps.Equal(v.Summary, other.Summary) &&
+		slices.Equal(v.Representations, other.Representations)
+}
+
+func (v *Version) Clone() *Version {
+	if v == nil {
+		return nil
 	}
 
-	return &metadata{
-		UUID:             uuid.NewString(),
-		FmtVersion:       1,
-		Loc:              loc,
-		SchemaList:       []*iceberg.Schema{schema},
-		CurrentVersionId: version.VersionID,
-		VersionList:      []Version{version},
-		VersionLogList: []VersionLogEntry{
-			{
-				TimestampMs: timestampMs,
-				VersionID:   version.VersionID,
-			},
-		},
-		Props: props,
-	}, nil
+	cloned := *v
+	cloned.Summary = maps.Clone(v.Summary)
+	cloned.Representations = slices.Clone(v.Representations)
+	cloned.DefaultNamespace = slices.Clone(v.DefaultNamespace)
+	return &cloned
 }
 
-// CreateMetadata creates a new view metadata file and writes it to storage.
-//
-// Returns the full path to the created metadata file, or an error if creation fails.
-//
-// Note: This function only supports creating new views with format version 1.
-// It does not support updating existing view metadata.
-func CreateMetadata(
-	ctx context.Context,
-	catalogName string,
-	nsIdent []string,
-	schema *iceberg.Schema,
-	viewSQL string,
-	loc string,
-	props iceberg.Properties,
-) (metadataLocation string, err error) {
-	versionId := int64(1)
-	timestampMs := time.Now().UnixMilli()
-
-	viewVersion := Version{
-		VersionID:   versionId,
-		TimestampMs: timestampMs,
-		SchemaID:    schema.ID,
-		Summary:     map[string]string{"sql": viewSQL},
-		Representations: []SQLRepresentation{
-			{Type: "sql", SQL: viewSQL, Dialect: "default"},
-		},
-		DefaultCatalog:   &catalogName,
-		DefaultNamespace: nsIdent,
-	}
-
-	metadataLocation = loc + "/metadata/view-" + uuid.New().String() + ".metadata.json"
-	viewMetadata, err := NewMetadata(schema, viewVersion, loc, props)
+type VersionLogEntry struct {
+	// Timestamp when the view's current-version-id was updated (ms from epoch)
+	TimestampMS int64 `json:"timestamp-ms"`
+	// ID that current-version-id was set to
+	VersionID int64 `json:"version-id"`
+}
+
+// ParseMetadata parses json metadata 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 "", fmt.Errorf("failed to create view metadata: %w", err)
+		return nil, err
 	}
 
-	viewMetadataBytes, err := json.Marshal(viewMetadata)
-	if err != nil {
-		return "", fmt.Errorf("failed to marshal view metadata: %w", 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) {
+	ver := struct {
+		FormatVersion int `json:"format-version"`
+	}{}
+	if err := json.Unmarshal(b, &ver); err != nil {
+		return nil, err
 	}
 
-	fs, err := io.LoadFS(ctx, props, metadataLocation)
-	if err != nil {
-		return "", fmt.Errorf("failed to load filesystem for view metadata: %w", err)
+	var ret Metadata
+	ret = &metadata{}
+	return ret, json.Unmarshal(b, ret)
+}
+
+// indexBy indexes a slice into a map, using a provided extractKey function
+// The extractKey function will be called on each item in the slice and assign the
+// item as a value for that key in the resultant map.
+func indexBy[T any, K comparable](s []T, extractKey func(T) K) map[K]T {
+	index := make(map[K]T)
+	for _, v := range s {
+		index[extractKey(v)] = v
+	}
+	return index
+}
+
+// Cloner is an interface which implements a Clone method for deep copying itself
+type cloner[T any] interface {
+	// Clone returns a deep copy of the underlying object
+	Clone() T
+}
+
+// cloneSlice returns a deep-clone of a Slice of elements implementing Cloner
+func cloneSlice[T cloner[T]](val []T) []T {
+	cloned := make([]T, len(val))
+	for i, elem := range val {
+		cloned[i] = elem.Clone()
+	}
+	return cloned
+}
+
+// https://iceberg.apache.org/view-spec/
+type metadata struct {
+	FormatVersionValue    int                `json:"format-version"`
+	UUID                  uuid.UUID          `json:"view-uuid"`
+	Loc                   string             `json:"location"`
+	CurrentVersionIDValue int64              `json:"current-version-id"`
+	VersionList           []*Version         `json:"versions"`
+	SchemaList            []*iceberg.Schema  `json:"schemas"`
+	VersionLogList        []VersionLogEntry  `json:"version-log"`
+	Props                 iceberg.Properties `json:"properties,omitempty"`
+
+	// updates from builder if applicable
+	updates Updates
+
+	// cached lookup helpers, must be initialized in init()
+	lazyVersionsByID func() map[int64]*Version
+	lazySchemasByID  func() map[int]*iceberg.Schema
+}
+
+func (m *metadata) Equals(other Metadata) bool {
+	if other == nil {
+		return false
+	}
+
+	if m == other {
+		return true
+	}
+
+	return m.UUID == other.ViewUUID() &&
+		m.Loc == other.Location() &&
+		m.FormatVersionValue == other.FormatVersion() &&
+		iceinternal.SliceEqualHelper(m.Schemas(), other.Schemas()) &&
+		iceinternal.SliceEqualHelper(m.Versions(), other.Versions()) &&
+		m.CurrentVersionIDValue == other.CurrentVersionID() &&
+		slices.Equal(m.VersionLogList, other.VersionLog())
+}
+
+func (m *metadata) Updates() Updates { return m.updates }
+
+func (m *metadata) FormatVersion() int         { return m.FormatVersionValue }
+func (m *metadata) ViewUUID() uuid.UUID        { return m.UUID }
+func (m *metadata) Location() string           { return m.Loc }
+func (m *metadata) Versions() []*Version       { return m.VersionList }
+func (m *metadata) Schemas() []*iceberg.Schema { return m.SchemaList }
+func (m *metadata) SchemasByID() map[int]*iceberg.Schema {
+	return maps.Clone(m.lazySchemasByID())
+}
+
+func (m *metadata) CurrentVersionID() int64 {
+	return m.CurrentVersionIDValue
+}
+
+func (m *metadata) CurrentVersion() *Version {
+	version, ok := m.lazyVersionsByID()[m.CurrentVersionIDValue]
+	if !ok {
+		panic("current version not found")
 	}
+	return version
+}
+
+func (m *metadata) CurrentSchemaID() int {
+	return m.CurrentVersion().SchemaID
+}
 
-	wfs, ok := fs.(io.WriteFileIO)
+func (m *metadata) CurrentSchema() *iceberg.Schema {
+	schema, ok := m.lazySchemasByID()[m.CurrentSchemaID()]
 	if !ok {
-		return "", errors.New("filesystem IO does not support writing")
+		panic("current schema not found")
+	}
+	return schema
+}
+
+func (m *metadata) VersionLog() []VersionLogEntry {
+	return m.VersionLogList
+}
+
+func (m *metadata) Properties() iceberg.Properties {
+	return m.Props
+}
+
+func (m *metadata) validate() error {
+	if m.Loc == "" {
+		return fmt.Errorf("%w: location is required", ErrInvalidViewMetadata)
+	}
+
+	if m.UUID == uuid.Nil {
+		return fmt.Errorf("%w: view-uuid is required", ErrInvalidViewMetadata)
+	}

Review Comment:
   Ack, I'll remove



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["twuebi (via GitHub)" <gi...@apache.org>]

twuebi commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2542005062


##########
view/metadata.go:
##########
@@ -18,135 +18,523 @@
 package view
 
 import (
-	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"maps"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
 	"time"
 
 	"github.com/apache/iceberg-go"
-	"github.com/apache/iceberg-go/internal"
-	"github.com/apache/iceberg-go/io"
+	iceinternal "github.com/apache/iceberg-go/internal"
+	"github.com/apache/iceberg-go/table"
+
 	"github.com/google/uuid"
 )
 
-// LoadMetadata reads and parses a view metadata file from the specified location.
+var (
+	ErrInvalidViewMetadata              = errors.New("invalid view metadata")
+	ErrInvalidViewMetadataFormatVersion = errors.New("invalid or missing format-version in view metadata")
+)
+
+const (
+	// LastAddedID is used in place of ID fields (e.g. schema, version) to indicate that
+	// the last added instance of that type should be used.
+	LastAddedID                = -1
+	SupportedViewFormatVersion = 1
+	DefaultViewFormatVersion   = SupportedViewFormatVersion
+)
+
+const (
+	initialSchemaID = 0
+
+	viewEngineProperty = "engine-name"
+	defaultViewEngine  = "iceberg-go"
+)
+
+// Metadata for an iceberg view as specified in the Iceberg spec
+// https://iceberg.apache.org/view-spec/
+type Metadata interface {
+	// FormatVersion indicates the version of this metadata, 1 for V1
+	FormatVersion() int
+	// ViewUUID returns a UUID that identifies the view, generated when the
+	// view is created. Implementations must throw an exception if a view's
+	// UUID does not match the expected UUID after refreshing metadata.
+	ViewUUID() uuid.UUID
+	// Location is the table's base location. This is used by writers to determine
+	// where to store data files, manifest files, and table metadata files.
+	Location() string
+	// Schemas returns the list of view schemas
+	Schemas() []*iceberg.Schema
+	// CurrentVersionID returns the ID of the current version of the view (version-id)
+	CurrentVersionID() int64
+	// CurrentVersion returns the current version of the view
+	CurrentVersion() *Version
+	// CurrentSchemaID returns the ID of the current schema
+	CurrentSchemaID() int
+	// CurrentSchema returns the current schema of the view
+	CurrentSchema() *iceberg.Schema
+	// SchemasByID returns a map of schema IDs to schemas
+	SchemasByID() map[int]*iceberg.Schema
+	// Versions returns the list of view versions
+	Versions() []*Version
+	// VersionLog returns a list of version log entries
+	// with the timestamp and version-id for every change to current-version-id
+	VersionLog() []VersionLogEntry
+	// Properties is a string to string map of view properties.
+	Properties() iceberg.Properties
+
+	// Updates returns the list of metadata updates used to build this metadata
+	Updates() Updates
+
+	Equals(Metadata) bool
+}
+
+// VersionSummary is string to string map of summary metadata about a view's version
+type VersionSummary map[string]string
+
+// Representation is a struct containing information about a view's representation
+// https://iceberg.apache.org/view-spec/#sql-representation
+type Representation struct {
+	// Must be sql
+	Type string `json:"type"`
+	// A SQL SELECT statement
+	Sql string `json:"sql"`
+	// The dialect of the sql SELECT statement (e.g., "trino" or "spark")
+	Dialect string `json:"dialect"`
+}
+type Representations []Representation
+
+func NewRepresentation(sql string, dialect string) Representation {
+	return Representation{
+		Type:    "sql",
+		Sql:     sql,
+		Dialect: dialect,
+	}
+}
+
+type Version struct {
+	// ID for the version
+	VersionID int64 `json:"version-id"`
+	// ID of the schema for the view version
+	SchemaID int `json:"schema-id"`
+	// Timestamp when the version was created (ms from epoch)
+	TimestampMS int64 `json:"timestamp-ms"`
+	// A string to string map of summary metadata about the version
+	Summary VersionSummary `json:"summary"`
+	// A list of representations for the view definition
+	Representations []Representation `json:"representations"`
+	// The default view namespace (as a list of strings)
+	DefaultNamespace table.Identifier `json:"default-namespace"`
+	// An (optional) default catalog name for querying the view
+	DefaultCatalog string `json:"default-catalog,omitempty"`
+}
+
+type VersionOpt func(*Version)
+
+func WithVersionSummary(summary VersionSummary) VersionOpt {
+	return func(v *Version) {
+		if v.Summary == nil {
+			v.Summary = summary
+		} else {
+			maps.Copy(v.Summary, summary)
+		}
+	}
+}
+
+func WithDefaultViewCatalog(catalogName string) func(*Version) {
+	return func(version *Version) {
+		version.DefaultCatalog = catalogName
+	}
+}
+
+func WithTimestampMS(timestampMS int64) VersionOpt {
+	return func(version *Version) {
+		version.TimestampMS = timestampMS
+	}
+}
+
+// NewVersion creates a Version instance from the provided parameters.
+// If building updates using the MetadataBuilder, and one desires to use the last
+// added schema ID, one should use the LastAddedID constant as the provided schemaID
 //
-// Returns the loaded Metadata or an error if the file cannot be read or parsed.
-func LoadMetadata(ctx context.Context,
-	props iceberg.Properties,
-	metadataLocation string,
-	name string,
-	namespace string,
-) (_ Metadata, err error) {
-	fs, err := io.LoadFS(ctx, props, metadataLocation)
-	if err != nil {
-		return nil, fmt.Errorf("error loading view metadata: %w", err)
+// Note that NewVersion automatically seeds TimestampMS to time.Now().UnixMilli(),
+// and one should use the option WithTimestampMS to override this behavior.
+func NewVersion(id int64, schemaID int, representations []Representation, defaultNS table.Identifier, opts ...VersionOpt) (*Version, error) {
+	if id < 1 {
+		return nil, errors.New("id should be greater than 0")
 	}
 
-	inputFile, err := fs.Open(metadataLocation)
-	if err != nil {
-		return nil, fmt.Errorf("%s.%s: error encountered loading view metadata: %w", namespace, name, err)
+	if len(representations) == 0 {
+		return nil, errors.New("invalid view version: must have at least one representation")
 	}
-	defer internal.CheckedClose(inputFile, &err)
 
-	var m metadata
-	if err := json.NewDecoder(inputFile).Decode(&m); err != nil {
-		return nil, fmt.Errorf("error encountered decoding view metadata: %w", err)
+	version := &Version{
+		VersionID:        id,
+		SchemaID:         schemaID,
+		Representations:  representations,
+		Summary:          VersionSummary{viewEngineProperty: defaultViewEngine},
+		TimestampMS:      time.Now().UnixMilli(),
+		DefaultNamespace: defaultNS,
 	}
 
-	return &m, nil
+	for _, opt := range opts {
+		opt(version)
+	}
+
+	return version, nil
 }
 
-// NewMetadata returns a view metadata for a given version.
-func NewMetadata(schema *iceberg.Schema,
-	version Version,
-	loc string,
-	props iceberg.Properties,
-) (Metadata, error) {
-	timestampMs := time.Now().UnixMilli()
+// NewVersionFromSQL creates a new Version with a single representation
+// using the provided SQL and dialect "default".
+func NewVersionFromSQL(id int64, schemaID int, sql string, defaultNS table.Identifier, opts ...VersionOpt) (*Version, error) {
+	return NewVersion(id, schemaID, Representations{NewRepresentation(sql, "default")}, defaultNS, opts...)
+}
 
-	if version.SchemaID != schema.ID {
-		return nil, fmt.Errorf("%w: version.SchemaID does not match schema.ID", iceberg.ErrInvalidArgument)
+// Equals checks whether the other Version would behave the same
+// while ignoring the view version id, and the creation timestamp
+func (v *Version) Equals(other *Version) bool {
+	return v.SchemaID == other.SchemaID &&
+		v.DefaultCatalog == other.DefaultCatalog &&
+		slices.Equal(v.DefaultNamespace, other.DefaultNamespace) &&
+		maps.Equal(v.Summary, other.Summary) &&
+		slices.Equal(v.Representations, other.Representations)
+}
+
+func (v *Version) Clone() *Version {
+	if v == nil {
+		return nil
 	}
 
-	return &metadata{
-		UUID:             uuid.NewString(),
-		FmtVersion:       1,
-		Loc:              loc,
-		SchemaList:       []*iceberg.Schema{schema},
-		CurrentVersionId: version.VersionID,
-		VersionList:      []Version{version},
-		VersionLogList: []VersionLogEntry{
-			{
-				TimestampMs: timestampMs,
-				VersionID:   version.VersionID,
-			},
-		},
-		Props: props,
-	}, nil
+	cloned := *v
+	cloned.Summary = maps.Clone(v.Summary)
+	cloned.Representations = slices.Clone(v.Representations)
+	cloned.DefaultNamespace = slices.Clone(v.DefaultNamespace)
+	return &cloned
 }
 
-// CreateMetadata creates a new view metadata file and writes it to storage.
-//
-// Returns the full path to the created metadata file, or an error if creation fails.
-//
-// Note: This function only supports creating new views with format version 1.
-// It does not support updating existing view metadata.
-func CreateMetadata(
-	ctx context.Context,
-	catalogName string,
-	nsIdent []string,
-	schema *iceberg.Schema,
-	viewSQL string,
-	loc string,
-	props iceberg.Properties,
-) (metadataLocation string, err error) {
-	versionId := int64(1)
-	timestampMs := time.Now().UnixMilli()
-
-	viewVersion := Version{
-		VersionID:   versionId,
-		TimestampMs: timestampMs,
-		SchemaID:    schema.ID,
-		Summary:     map[string]string{"sql": viewSQL},
-		Representations: []SQLRepresentation{
-			{Type: "sql", SQL: viewSQL, Dialect: "default"},
-		},
-		DefaultCatalog:   &catalogName,
-		DefaultNamespace: nsIdent,
-	}
-
-	metadataLocation = loc + "/metadata/view-" + uuid.New().String() + ".metadata.json"
-	viewMetadata, err := NewMetadata(schema, viewVersion, loc, props)
+type VersionLogEntry struct {
+	// Timestamp when the view's current-version-id was updated (ms from epoch)
+	TimestampMS int64 `json:"timestamp-ms"`
+	// ID that current-version-id was set to
+	VersionID int64 `json:"version-id"`
+}
+
+// ParseMetadata parses json metadata 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 "", fmt.Errorf("failed to create view metadata: %w", err)
+		return nil, err
 	}
 
-	viewMetadataBytes, err := json.Marshal(viewMetadata)
-	if err != nil {
-		return "", fmt.Errorf("failed to marshal view metadata: %w", 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) {
+	ver := struct {
+		FormatVersion int `json:"format-version"`
+	}{}
+	if err := json.Unmarshal(b, &ver); err != nil {
+		return nil, err
 	}
 
-	fs, err := io.LoadFS(ctx, props, metadataLocation)
-	if err != nil {
-		return "", fmt.Errorf("failed to load filesystem for view metadata: %w", err)
+	var ret Metadata
+	ret = &metadata{}
+	return ret, json.Unmarshal(b, ret)
+}
+
+// indexBy indexes a slice into a map, using a provided extractKey function
+// The extractKey function will be called on each item in the slice and assign the
+// item as a value for that key in the resultant map.
+func indexBy[T any, K comparable](s []T, extractKey func(T) K) map[K]T {
+	index := make(map[K]T)
+	for _, v := range s {
+		index[extractKey(v)] = v
+	}
+	return index
+}
+
+// Cloner is an interface which implements a Clone method for deep copying itself
+type cloner[T any] interface {
+	// Clone returns a deep copy of the underlying object
+	Clone() T
+}
+
+// cloneSlice returns a deep-clone of a Slice of elements implementing Cloner
+func cloneSlice[T cloner[T]](val []T) []T {
+	cloned := make([]T, len(val))
+	for i, elem := range val {
+		cloned[i] = elem.Clone()
+	}
+	return cloned
+}
+
+// https://iceberg.apache.org/view-spec/
+type metadata struct {
+	FormatVersionValue    int                `json:"format-version"`
+	UUID                  uuid.UUID          `json:"view-uuid"`
+	Loc                   string             `json:"location"`
+	CurrentVersionIDValue int64              `json:"current-version-id"`
+	VersionList           []*Version         `json:"versions"`
+	SchemaList            []*iceberg.Schema  `json:"schemas"`
+	VersionLogList        []VersionLogEntry  `json:"version-log"`
+	Props                 iceberg.Properties `json:"properties,omitempty"`
+
+	// updates from builder if applicable
+	updates Updates
+
+	// cached lookup helpers, must be initialized in init()
+	lazyVersionsByID func() map[int64]*Version
+	lazySchemasByID  func() map[int]*iceberg.Schema
+}
+
+func (m *metadata) Equals(other Metadata) bool {
+	if other == nil {
+		return false
+	}
+
+	if m == other {
+		return true
+	}
+
+	return m.UUID == other.ViewUUID() &&
+		m.Loc == other.Location() &&
+		m.FormatVersionValue == other.FormatVersion() &&
+		iceinternal.SliceEqualHelper(m.Schemas(), other.Schemas()) &&
+		iceinternal.SliceEqualHelper(m.Versions(), other.Versions()) &&
+		m.CurrentVersionIDValue == other.CurrentVersionID() &&
+		slices.Equal(m.VersionLogList, other.VersionLog())
+}
+
+func (m *metadata) Updates() Updates { return m.updates }
+
+func (m *metadata) FormatVersion() int         { return m.FormatVersionValue }
+func (m *metadata) ViewUUID() uuid.UUID        { return m.UUID }
+func (m *metadata) Location() string           { return m.Loc }
+func (m *metadata) Versions() []*Version       { return m.VersionList }
+func (m *metadata) Schemas() []*iceberg.Schema { return m.SchemaList }
+func (m *metadata) SchemasByID() map[int]*iceberg.Schema {
+	return maps.Clone(m.lazySchemasByID())
+}
+
+func (m *metadata) CurrentVersionID() int64 {
+	return m.CurrentVersionIDValue
+}
+
+func (m *metadata) CurrentVersion() *Version {
+	version, ok := m.lazyVersionsByID()[m.CurrentVersionIDValue]
+	if !ok {
+		panic("current version not found")
 	}
+	return version
+}
+
+func (m *metadata) CurrentSchemaID() int {
+	return m.CurrentVersion().SchemaID
+}
 
-	wfs, ok := fs.(io.WriteFileIO)
+func (m *metadata) CurrentSchema() *iceberg.Schema {
+	schema, ok := m.lazySchemasByID()[m.CurrentSchemaID()]
 	if !ok {
-		return "", errors.New("filesystem IO does not support writing")
+		panic("current schema not found")
+	}

Review Comment:
   same as above



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2550787439


##########
view/requirements.go:
##########


Review Comment:
   thanks for the snippet will incorporate this as well



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["twuebi (via GitHub)" <gi...@apache.org>]

twuebi commented on PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#issuecomment-3602749276

   nevermind, I just realized we're doing the schemaId reassignment with:
   
   ```go
   func (b *MetadataBuilder) addSchema(schema *iceberg.Schema) (int, error) {
   	newSchemaID := b.reuseOrCreateNewSchemaID(schema)
   ```


-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#issuecomment-3602804018

   Also in case it's a concern, per REST API spec, schema ID on create is managed by the catalog, and if reassigned, new schema ID is propagated to the view version's schema ID ([spec](https://github.com/apache/iceberg/blob/16e84356dae1975fa04d8c3ecce30a90df18ca9f/open-api/rest-catalog-open-api.yaml#L3648))


-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2543871283


##########
view/metadata_builder.go:
##########
@@ -1,57 +1,468 @@
-// 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 view
 
 import (
+	"errors"
+	"fmt"
+	"maps"
+	"slices"
+	"strings"
+	"time"
+
 	"github.com/apache/iceberg-go"
+
+	"github.com/google/uuid"
 )
 
-// MetadataBuilder is a struct used for building and updating Iceberg view metadata.
-type MetadataBuilder struct{}
+type MetadataBuilder struct {
+	base    Metadata
+	updates Updates
+
+	// common fields
+	formatVersion    int
+	uuid             uuid.UUID
+	loc              string
+	schemaList       []*iceberg.Schema
+	versionList      []*Version
+	currentVersionID int64
+	versionLog       []VersionLogEntry
+	props            iceberg.Properties
+
+	// lookup maps
+	versionsById map[int64]*Version
+	schemasById  map[int]*iceberg.Schema
+
+	// update tracking
+	versionHistoryEntry *VersionLogEntry
+	lastAddedVersionID  *int64
+	lastAddedSchemaID   *int
+
+	// error tracking for build chaining
+	// if set, subsequent operations become a noop
+	err error
+}
+
+func NewMetadataBuilder() (*MetadataBuilder, error) {
+	return &MetadataBuilder{
+		formatVersion: DefaultViewFormatVersion,
+		updates:       make([]Update, 0),
+		schemaList:    make([]*iceberg.Schema, 0),
+		versionList:   make([]*Version, 0),
+		versionLog:    make([]VersionLogEntry, 0),
+		props:         make(iceberg.Properties),
+		versionsById:  make(map[int64]*Version),
+		schemasById:   make(map[int]*iceberg.Schema),
+	}, nil
+}
+
+func MetadataBuilderFromBase(metadata Metadata) (*MetadataBuilder, error) {
+	b := &MetadataBuilder{}
+	b.base = metadata
+
+	// Copy fields
+	b.formatVersion = metadata.FormatVersion()
+	b.uuid = metadata.ViewUUID()
+	b.loc = metadata.Location()
+	b.schemaList = slices.Clone(metadata.Schemas())
+	b.currentVersionID = metadata.CurrentVersionID()
+	b.versionList = cloneSlice(metadata.Versions())
+	b.versionLog = slices.Clone(metadata.VersionLog())
+	b.props = maps.Clone(metadata.Properties())
+
+	// Build lookup maps
+	b.versionsById = indexBy(b.versionList, func(vl *Version) int64 { return vl.VersionID })
+	b.schemasById = indexBy(b.schemaList, func(sm *iceberg.Schema) int { return sm.ID })
+
+	return b, nil
+}
+
+func (b *MetadataBuilder) HasChanges() bool { return len(b.updates) > 0 }
 
-func (b *MetadataBuilder) AssignUUID(_ string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetCurrentVersion(version *Version, schema *iceberg.Schema) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	newSchemaID, err := b.addSchema(schema)
+	if b.setErr(err) {
+		return b
+	}
+
+	newVersion := version.Clone()
+	newVersion.SchemaID = newSchemaID
+	newVersionID, err := b.addVersion(newVersion)
+	if b.setErr(err) {
+		return b
+	}
+
+	return b.SetCurrentVersionID(newVersionID)
 }
 
-func (b *MetadataBuilder) UpgradeFormatVersion(_ int) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) AddVersion(newVersion *Version) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	_, err := b.addVersion(newVersion)
+	if b.setErr(err) {
+		return b
+	}
+	return b
 }
 
-func (b *MetadataBuilder) AddSchema(_ *iceberg.Schema) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) addVersion(newVersion *Version) (int64, error) {
+	newVersionID := b.reuseOrCreateNewVersionID(newVersion)
+	version := newVersion.Clone()
+	if newVersionID != version.VersionID {
+		version.VersionID = newVersionID
+	}
+
+	// Check if this version was added in an update already
+	if _, ok := b.versionsById[newVersionID]; ok {
+		for _, upd := range b.updates {
+			if vvu, ok := upd.(*addViewVersionUpdate); ok && vvu.Version.VersionID == newVersionID {
+				b.lastAddedVersionID = &newVersionID
+				return newVersionID, nil
+			}
+		}
+	}
+
+	// If the SchemaID of the version is unset (nil), we attach the lastAddedSchemaID to it.
+	// If we have no lastAddedSchemaID, we fail
+	if version.SchemaID == LastAddedID {
+		if b.lastAddedSchemaID == nil {
+			return 0, errors.New("cannot set last added schema: no schema has been added")
+		}
+		version.SchemaID = *b.lastAddedSchemaID
+	}
+
+	if _, ok := b.schemasById[version.SchemaID]; !ok {
+		return 0, fmt.Errorf("cannot add version with unknown schema: %d", version.SchemaID)
+	}
+
+	if len(version.Representations) == 0 {
+		return 0, fmt.Errorf("cannot add version with no representations")
+	}
+	dialects := make(map[string]struct{})
+	for _, repr := range version.Representations {
+		normalizedDialect := strings.ToLower(repr.Dialect)
+		if _, ok := dialects[normalizedDialect]; ok {
+			return 0, fmt.Errorf("invalid view version: cannot add multiple queries for dialect %s", normalizedDialect)
+		}
+		dialects[normalizedDialect] = struct{}{}
+	}
+
+	b.versionList = append(b.versionList, version)
+	b.versionsById[version.VersionID] = version
+
+	if b.lastAddedSchemaID != nil && version.SchemaID == *b.lastAddedSchemaID {
+		updateVersion := version.Clone()
+		updateVersion.SchemaID = LastAddedID
+		b.updates = append(b.updates, NewAddViewVersionUpdate(updateVersion))
+	} else {
+		b.updates = append(b.updates, NewAddViewVersionUpdate(version))
+	}
+
+	b.lastAddedVersionID = &newVersionID
+	return newVersionID, nil
 }
 
-func (b *MetadataBuilder) SetLocation(_ string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) AddSchema(schema *iceberg.Schema) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	_, err := b.addSchema(schema)
+	if b.setErr(err) {
+		return b
+	}
+	return b
 }
 
-func (b *MetadataBuilder) SetProperties(_ map[string]string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) addSchema(schema *iceberg.Schema) (int, error) {
+	newSchemaID := b.reuseOrCreateNewSchemaID(schema)
+
+	if _, ok := b.schemasById[newSchemaID]; ok {
+		return newSchemaID, nil
+	}
+
+	newSchema := schema
+	// Build a fresh schema if we reset the ID
+	if schema.ID != newSchemaID {
+		newSchema = iceberg.NewSchemaWithIdentifiers(newSchemaID, schema.IdentifierFieldIDs, schema.Fields()...)
+	}
+
+	b.schemaList = append(b.schemaList, newSchema)
+	b.schemasById[newSchema.ID] = newSchema
+	b.updates = append(b.updates, NewAddSchemaUpdate(newSchema))
+	b.lastAddedSchemaID = &newSchemaID
+
+	return newSchemaID, nil
 }
 
-func (b *MetadataBuilder) RemoveProperties(_ []string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) RemoveProperties(keys []string) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if len(keys) == 0 {
+		return b
+	}
+
+	b.updates = append(b.updates, NewRemovePropertiesUpdate(keys))
+	for _, key := range keys {
+		delete(b.props, key)
+	}
+
+	return b
 }
 
-func (b *MetadataBuilder) AddVersion(_ *Version) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetCurrentVersionID(newVersionID int64) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if newVersionID == LastAddedID {
+		if b.lastAddedVersionID == nil {
+			b.err = errors.New("cannot set current version to last added, no version has been added")
+			return b
+		}
+		newVersionID = *b.lastAddedVersionID
+	}
+
+	if newVersionID == b.currentVersionID {
+		return b
+	}
+
+	_, ok := b.versionsById[newVersionID]
+	if !ok {
+		b.err = fmt.Errorf("cannot set current version to unknown version with id %d", newVersionID)
+		return b
+	}
+
+	if b.lastAddedVersionID != nil && *b.lastAddedVersionID == newVersionID {
+		b.updates = append(b.updates, NewSetCurrentVersionUpdate(LastAddedID))
+	} else {
+		b.updates = append(b.updates, NewSetCurrentVersionUpdate(newVersionID))
+	}
+	b.currentVersionID = newVersionID
+
+	// Set the current history entry.
+	// If the version was added in the current set of changes, use its timestamp,
+	// otherwise use system time.
+	updateTimestampMS := time.Now().UnixMilli()
+	for _, update := range b.updates {
+		if v, ok := update.(*addViewVersionUpdate); ok && v.Version.VersionID == newVersionID {
+			updateTimestampMS = v.Version.TimestampMS
+			break
+		}
+	}
+
+	b.versionHistoryEntry = &VersionLogEntry{
+		VersionID:   newVersionID,
+		TimestampMS: updateTimestampMS,
+	}
+
+	return b
 }
 
-func (b *MetadataBuilder) SetCurrentVersionID(_ int64) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetFormatVersion(formatVersion int) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if formatVersion < b.formatVersion {
+		b.err = fmt.Errorf("downgrading format version from %d to %d is not allowed",
+			b.formatVersion, formatVersion)
+		return b
+	}
+
+	if formatVersion > SupportedViewFormatVersion {
+		b.err = fmt.Errorf("unsupported format version %d", formatVersion)
+		return b
+	}
+
+	if formatVersion == b.formatVersion {
+		return b
+	}
+
+	b.updates = append(b.updates, NewUpgradeFormatVersionUpdate(formatVersion))
+	b.formatVersion = formatVersion
+
+	return b
+}
+
+func (b *MetadataBuilder) SetLoc(loc string) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if b.loc == loc {
+		return b
+	}
+
+	b.updates = append(b.updates, NewSetLocationUpdate(loc))
+	b.loc = loc
+
+	return b
+}
+
+func (b *MetadataBuilder) SetProperties(props iceberg.Properties) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if len(props) == 0 {
+		return b
+	}
+
+	b.updates = append(b.updates, NewSetPropertiesUpdate(props))
+	if b.props == nil {
+		b.props = props
+	} else {
+		maps.Copy(b.props, props)
+	}
+
+	return b
+}
+
+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
+	}
+
+	// Noop - same UUID does not generate a change.
+	if b.uuid == newUUID {
+		return b
+	}
+
+	if b.uuid != uuid.Nil {
+		b.err = errors.New("cannot reassign uuid")
+		return b
+	}
+
+	b.updates = append(b.updates, NewAssignUUIDUpdate(newUUID))
+	b.uuid = newUUID
+
+	return b
+}
+
+func (b *MetadataBuilder) Err() error {
+	return b.err
+}
+
+func (b *MetadataBuilder) buildMetadata(withUpdates bool) (*metadata, error) {
+	uuid_ := b.uuid
+	if uuid_ == uuid.Nil {
+		uuid_ = uuid.New()
+	}
+
+	md := &metadata{
+		FormatVersionValue:    b.formatVersion,
+		UUID:                  uuid_,
+		Loc:                   b.loc,
+		CurrentVersionIDValue: b.currentVersionID,
+		VersionList:           b.versionList,
+		VersionLogList:        b.versionLog,
+		SchemaList:            b.schemaList,
+		Props:                 b.props,
+	}
+	if withUpdates {
+		md.updates = b.updates
+	}
+	md.init()
+
+	return md, nil
+}
+
+// build builds the view metadata updates from the builder
+// if withUpdates is set to false, updates will not be included
+// in the returned Metadata, thus its Updates() will return an empty slice
+func (b *MetadataBuilder) build(withUpdates bool) (Metadata, error) {

Review Comment:
   Understood, will implement!



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2542737031


##########
catalog/rest/rest.go:
##########
@@ -1159,55 +1159,95 @@ type createViewRequest struct {
 	Schema      *iceberg.Schema    `json:"schema"`
 	Location    string             `json:"location,omitempty"`
 	Props       iceberg.Properties `json:"properties,omitempty"`

Review Comment:
   good catch thanks



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2543866700


##########
view/metadata_builder.go:
##########
@@ -1,57 +1,468 @@
-// 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 view
 
 import (
+	"errors"
+	"fmt"
+	"maps"
+	"slices"
+	"strings"
+	"time"
+
 	"github.com/apache/iceberg-go"
+
+	"github.com/google/uuid"
 )
 
-// MetadataBuilder is a struct used for building and updating Iceberg view metadata.
-type MetadataBuilder struct{}
+type MetadataBuilder struct {
+	base    Metadata
+	updates Updates
+
+	// common fields
+	formatVersion    int
+	uuid             uuid.UUID
+	loc              string
+	schemaList       []*iceberg.Schema
+	versionList      []*Version
+	currentVersionID int64
+	versionLog       []VersionLogEntry
+	props            iceberg.Properties
+
+	// lookup maps
+	versionsById map[int64]*Version
+	schemasById  map[int]*iceberg.Schema
+
+	// update tracking
+	versionHistoryEntry *VersionLogEntry
+	lastAddedVersionID  *int64
+	lastAddedSchemaID   *int
+
+	// error tracking for build chaining
+	// if set, subsequent operations become a noop
+	err error
+}
+
+func NewMetadataBuilder() (*MetadataBuilder, error) {
+	return &MetadataBuilder{
+		formatVersion: DefaultViewFormatVersion,
+		updates:       make([]Update, 0),
+		schemaList:    make([]*iceberg.Schema, 0),
+		versionList:   make([]*Version, 0),
+		versionLog:    make([]VersionLogEntry, 0),
+		props:         make(iceberg.Properties),
+		versionsById:  make(map[int64]*Version),
+		schemasById:   make(map[int]*iceberg.Schema),
+	}, nil
+}
+
+func MetadataBuilderFromBase(metadata Metadata) (*MetadataBuilder, error) {
+	b := &MetadataBuilder{}
+	b.base = metadata
+
+	// Copy fields
+	b.formatVersion = metadata.FormatVersion()
+	b.uuid = metadata.ViewUUID()
+	b.loc = metadata.Location()
+	b.schemaList = slices.Clone(metadata.Schemas())
+	b.currentVersionID = metadata.CurrentVersionID()
+	b.versionList = cloneSlice(metadata.Versions())
+	b.versionLog = slices.Clone(metadata.VersionLog())
+	b.props = maps.Clone(metadata.Properties())
+
+	// Build lookup maps
+	b.versionsById = indexBy(b.versionList, func(vl *Version) int64 { return vl.VersionID })
+	b.schemasById = indexBy(b.schemaList, func(sm *iceberg.Schema) int { return sm.ID })
+
+	return b, nil
+}
+
+func (b *MetadataBuilder) HasChanges() bool { return len(b.updates) > 0 }
 
-func (b *MetadataBuilder) AssignUUID(_ string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetCurrentVersion(version *Version, schema *iceberg.Schema) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	newSchemaID, err := b.addSchema(schema)
+	if b.setErr(err) {
+		return b
+	}
+
+	newVersion := version.Clone()
+	newVersion.SchemaID = newSchemaID
+	newVersionID, err := b.addVersion(newVersion)
+	if b.setErr(err) {
+		return b
+	}
+
+	return b.SetCurrentVersionID(newVersionID)
 }
 
-func (b *MetadataBuilder) UpgradeFormatVersion(_ int) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) AddVersion(newVersion *Version) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	_, err := b.addVersion(newVersion)
+	if b.setErr(err) {
+		return b
+	}
+	return b
 }
 
-func (b *MetadataBuilder) AddSchema(_ *iceberg.Schema) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) addVersion(newVersion *Version) (int64, error) {
+	newVersionID := b.reuseOrCreateNewVersionID(newVersion)
+	version := newVersion.Clone()
+	if newVersionID != version.VersionID {
+		version.VersionID = newVersionID
+	}
+
+	// Check if this version was added in an update already
+	if _, ok := b.versionsById[newVersionID]; ok {
+		for _, upd := range b.updates {
+			if vvu, ok := upd.(*addViewVersionUpdate); ok && vvu.Version.VersionID == newVersionID {
+				b.lastAddedVersionID = &newVersionID
+				return newVersionID, nil
+			}
+		}
+	}
+
+	// If the SchemaID of the version is unset (nil), we attach the lastAddedSchemaID to it.
+	// If we have no lastAddedSchemaID, we fail
+	if version.SchemaID == LastAddedID {
+		if b.lastAddedSchemaID == nil {
+			return 0, errors.New("cannot set last added schema: no schema has been added")
+		}
+		version.SchemaID = *b.lastAddedSchemaID
+	}
+
+	if _, ok := b.schemasById[version.SchemaID]; !ok {
+		return 0, fmt.Errorf("cannot add version with unknown schema: %d", version.SchemaID)
+	}
+
+	if len(version.Representations) == 0 {
+		return 0, fmt.Errorf("cannot add version with no representations")
+	}
+	dialects := make(map[string]struct{})
+	for _, repr := range version.Representations {
+		normalizedDialect := strings.ToLower(repr.Dialect)
+		if _, ok := dialects[normalizedDialect]; ok {
+			return 0, fmt.Errorf("invalid view version: cannot add multiple queries for dialect %s", normalizedDialect)
+		}
+		dialects[normalizedDialect] = struct{}{}
+	}
+
+	b.versionList = append(b.versionList, version)
+	b.versionsById[version.VersionID] = version
+
+	if b.lastAddedSchemaID != nil && version.SchemaID == *b.lastAddedSchemaID {
+		updateVersion := version.Clone()
+		updateVersion.SchemaID = LastAddedID
+		b.updates = append(b.updates, NewAddViewVersionUpdate(updateVersion))
+	} else {
+		b.updates = append(b.updates, NewAddViewVersionUpdate(version))
+	}
+
+	b.lastAddedVersionID = &newVersionID
+	return newVersionID, nil
 }
 
-func (b *MetadataBuilder) SetLocation(_ string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) AddSchema(schema *iceberg.Schema) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	_, err := b.addSchema(schema)
+	if b.setErr(err) {
+		return b
+	}
+	return b
 }
 
-func (b *MetadataBuilder) SetProperties(_ map[string]string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) addSchema(schema *iceberg.Schema) (int, error) {
+	newSchemaID := b.reuseOrCreateNewSchemaID(schema)
+
+	if _, ok := b.schemasById[newSchemaID]; ok {
+		return newSchemaID, nil
+	}
+
+	newSchema := schema
+	// Build a fresh schema if we reset the ID
+	if schema.ID != newSchemaID {
+		newSchema = iceberg.NewSchemaWithIdentifiers(newSchemaID, schema.IdentifierFieldIDs, schema.Fields()...)
+	}
+
+	b.schemaList = append(b.schemaList, newSchema)
+	b.schemasById[newSchema.ID] = newSchema
+	b.updates = append(b.updates, NewAddSchemaUpdate(newSchema))
+	b.lastAddedSchemaID = &newSchemaID
+
+	return newSchemaID, nil
 }
 
-func (b *MetadataBuilder) RemoveProperties(_ []string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) RemoveProperties(keys []string) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if len(keys) == 0 {
+		return b
+	}
+
+	b.updates = append(b.updates, NewRemovePropertiesUpdate(keys))
+	for _, key := range keys {
+		delete(b.props, key)
+	}
+
+	return b
 }
 
-func (b *MetadataBuilder) AddVersion(_ *Version) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetCurrentVersionID(newVersionID int64) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if newVersionID == LastAddedID {
+		if b.lastAddedVersionID == nil {
+			b.err = errors.New("cannot set current version to last added, no version has been added")
+			return b
+		}
+		newVersionID = *b.lastAddedVersionID
+	}
+
+	if newVersionID == b.currentVersionID {
+		return b
+	}
+
+	_, ok := b.versionsById[newVersionID]
+	if !ok {
+		b.err = fmt.Errorf("cannot set current version to unknown version with id %d", newVersionID)
+		return b
+	}
+
+	if b.lastAddedVersionID != nil && *b.lastAddedVersionID == newVersionID {
+		b.updates = append(b.updates, NewSetCurrentVersionUpdate(LastAddedID))
+	} else {
+		b.updates = append(b.updates, NewSetCurrentVersionUpdate(newVersionID))
+	}
+	b.currentVersionID = newVersionID
+
+	// Set the current history entry.
+	// If the version was added in the current set of changes, use its timestamp,
+	// otherwise use system time.
+	updateTimestampMS := time.Now().UnixMilli()
+	for _, update := range b.updates {
+		if v, ok := update.(*addViewVersionUpdate); ok && v.Version.VersionID == newVersionID {
+			updateTimestampMS = v.Version.TimestampMS
+			break
+		}
+	}
+
+	b.versionHistoryEntry = &VersionLogEntry{
+		VersionID:   newVersionID,
+		TimestampMS: updateTimestampMS,
+	}
+
+	return b
 }
 
-func (b *MetadataBuilder) SetCurrentVersionID(_ int64) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetFormatVersion(formatVersion int) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if formatVersion < b.formatVersion {
+		b.err = fmt.Errorf("downgrading format version from %d to %d is not allowed",
+			b.formatVersion, formatVersion)
+		return b
+	}
+
+	if formatVersion > SupportedViewFormatVersion {
+		b.err = fmt.Errorf("unsupported format version %d", formatVersion)
+		return b
+	}
+
+	if formatVersion == b.formatVersion {
+		return b
+	}
+
+	b.updates = append(b.updates, NewUpgradeFormatVersionUpdate(formatVersion))
+	b.formatVersion = formatVersion
+
+	return b
+}
+
+func (b *MetadataBuilder) SetLoc(loc string) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if b.loc == loc {
+		return b
+	}
+
+	b.updates = append(b.updates, NewSetLocationUpdate(loc))
+	b.loc = loc
+
+	return b
+}
+
+func (b *MetadataBuilder) SetProperties(props iceberg.Properties) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if len(props) == 0 {
+		return b
+	}
+
+	b.updates = append(b.updates, NewSetPropertiesUpdate(props))
+	if b.props == nil {
+		b.props = props
+	} else {
+		maps.Copy(b.props, props)
+	}
+
+	return b
+}
+
+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
+	}
+
+	// Noop - same UUID does not generate a change.
+	if b.uuid == newUUID {
+		return b
+	}
+
+	if b.uuid != uuid.Nil {
+		b.err = errors.New("cannot reassign uuid")
+		return b
+	}
+
+	b.updates = append(b.updates, NewAssignUUIDUpdate(newUUID))
+	b.uuid = newUUID
+
+	return b
+}
+
+func (b *MetadataBuilder) Err() error {
+	return b.err
+}
+
+func (b *MetadataBuilder) buildMetadata(withUpdates bool) (*metadata, error) {
+	uuid_ := b.uuid
+	if uuid_ == uuid.Nil {
+		uuid_ = uuid.New()
+	}
+
+	md := &metadata{
+		FormatVersionValue:    b.formatVersion,
+		UUID:                  uuid_,
+		Loc:                   b.loc,
+		CurrentVersionIDValue: b.currentVersionID,
+		VersionList:           b.versionList,
+		VersionLogList:        b.versionLog,
+		SchemaList:            b.schemaList,
+		Props:                 b.props,
+	}
+	if withUpdates {
+		md.updates = b.updates
+	}
+	md.init()
+
+	return md, nil
+}
+
+// build builds the view metadata updates from the builder
+// if withUpdates is set to false, updates will not be included
+// in the returned Metadata, thus its Updates() will return an empty slice
+func (b *MetadataBuilder) build(withUpdates bool) (Metadata, error) {

Review Comment:
   Sounds good to me, will implement it 👍 . Just a heads up going to be unable to work on this tomorrow but will resume work on Friday. 



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#issuecomment-3608057284

   @twuebi ack, now overriding `VersionID` to `1` in `NewMetadataWithUUID` (called by `NewMetadata`). Updated test for it as well


-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#issuecomment-3602764130

   > nevermind, I just realized we're doing the schemaId reassignment with:
   > 
   > ```go
   > func (b *MetadataBuilder) addSchema(schema *iceberg.Schema) (int, error) {
   > 	newSchemaID := b.reuseOrCreateNewSchemaID(schema)
   > ```
   
   Yes that's correct.  
   
   On the note about the REST impl, that's a good catch. I'll update it to override the version ID to 1 in the `CreateView` method. Does that sound good? 


-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2542843800


##########
view/metadata_builder.go:
##########
@@ -1,57 +1,468 @@
-// 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 view
 
 import (
+	"errors"
+	"fmt"
+	"maps"
+	"slices"
+	"strings"
+	"time"
+
 	"github.com/apache/iceberg-go"
+
+	"github.com/google/uuid"
 )
 
-// MetadataBuilder is a struct used for building and updating Iceberg view metadata.
-type MetadataBuilder struct{}
+type MetadataBuilder struct {
+	base    Metadata
+	updates Updates
+
+	// common fields
+	formatVersion    int
+	uuid             uuid.UUID
+	loc              string
+	schemaList       []*iceberg.Schema
+	versionList      []*Version
+	currentVersionID int64
+	versionLog       []VersionLogEntry
+	props            iceberg.Properties
+
+	// lookup maps
+	versionsById map[int64]*Version
+	schemasById  map[int]*iceberg.Schema
+
+	// update tracking
+	versionHistoryEntry *VersionLogEntry
+	lastAddedVersionID  *int64
+	lastAddedSchemaID   *int
+
+	// error tracking for build chaining
+	// if set, subsequent operations become a noop
+	err error
+}
+
+func NewMetadataBuilder() (*MetadataBuilder, error) {
+	return &MetadataBuilder{
+		formatVersion: DefaultViewFormatVersion,
+		updates:       make([]Update, 0),
+		schemaList:    make([]*iceberg.Schema, 0),
+		versionList:   make([]*Version, 0),
+		versionLog:    make([]VersionLogEntry, 0),
+		props:         make(iceberg.Properties),
+		versionsById:  make(map[int64]*Version),
+		schemasById:   make(map[int]*iceberg.Schema),
+	}, nil
+}
+
+func MetadataBuilderFromBase(metadata Metadata) (*MetadataBuilder, error) {
+	b := &MetadataBuilder{}
+	b.base = metadata
+
+	// Copy fields
+	b.formatVersion = metadata.FormatVersion()
+	b.uuid = metadata.ViewUUID()
+	b.loc = metadata.Location()
+	b.schemaList = slices.Clone(metadata.Schemas())
+	b.currentVersionID = metadata.CurrentVersionID()
+	b.versionList = cloneSlice(metadata.Versions())
+	b.versionLog = slices.Clone(metadata.VersionLog())
+	b.props = maps.Clone(metadata.Properties())
+
+	// Build lookup maps
+	b.versionsById = indexBy(b.versionList, func(vl *Version) int64 { return vl.VersionID })
+	b.schemasById = indexBy(b.schemaList, func(sm *iceberg.Schema) int { return sm.ID })
+
+	return b, nil
+}
+
+func (b *MetadataBuilder) HasChanges() bool { return len(b.updates) > 0 }
 
-func (b *MetadataBuilder) AssignUUID(_ string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetCurrentVersion(version *Version, schema *iceberg.Schema) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	newSchemaID, err := b.addSchema(schema)
+	if b.setErr(err) {
+		return b
+	}
+
+	newVersion := version.Clone()
+	newVersion.SchemaID = newSchemaID
+	newVersionID, err := b.addVersion(newVersion)
+	if b.setErr(err) {
+		return b
+	}
+
+	return b.SetCurrentVersionID(newVersionID)
 }
 
-func (b *MetadataBuilder) UpgradeFormatVersion(_ int) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) AddVersion(newVersion *Version) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	_, err := b.addVersion(newVersion)
+	if b.setErr(err) {
+		return b
+	}
+	return b
 }
 
-func (b *MetadataBuilder) AddSchema(_ *iceberg.Schema) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) addVersion(newVersion *Version) (int64, error) {
+	newVersionID := b.reuseOrCreateNewVersionID(newVersion)
+	version := newVersion.Clone()
+	if newVersionID != version.VersionID {
+		version.VersionID = newVersionID
+	}
+
+	// Check if this version was added in an update already
+	if _, ok := b.versionsById[newVersionID]; ok {
+		for _, upd := range b.updates {
+			if vvu, ok := upd.(*addViewVersionUpdate); ok && vvu.Version.VersionID == newVersionID {
+				b.lastAddedVersionID = &newVersionID
+				return newVersionID, nil
+			}
+		}
+	}
+
+	// If the SchemaID of the version is unset (nil), we attach the lastAddedSchemaID to it.
+	// If we have no lastAddedSchemaID, we fail
+	if version.SchemaID == LastAddedID {
+		if b.lastAddedSchemaID == nil {
+			return 0, errors.New("cannot set last added schema: no schema has been added")
+		}
+		version.SchemaID = *b.lastAddedSchemaID
+	}
+
+	if _, ok := b.schemasById[version.SchemaID]; !ok {
+		return 0, fmt.Errorf("cannot add version with unknown schema: %d", version.SchemaID)
+	}
+
+	if len(version.Representations) == 0 {
+		return 0, fmt.Errorf("cannot add version with no representations")
+	}
+	dialects := make(map[string]struct{})
+	for _, repr := range version.Representations {
+		normalizedDialect := strings.ToLower(repr.Dialect)
+		if _, ok := dialects[normalizedDialect]; ok {
+			return 0, fmt.Errorf("invalid view version: cannot add multiple queries for dialect %s", normalizedDialect)
+		}
+		dialects[normalizedDialect] = struct{}{}
+	}
+
+	b.versionList = append(b.versionList, version)
+	b.versionsById[version.VersionID] = version
+
+	if b.lastAddedSchemaID != nil && version.SchemaID == *b.lastAddedSchemaID {
+		updateVersion := version.Clone()
+		updateVersion.SchemaID = LastAddedID
+		b.updates = append(b.updates, NewAddViewVersionUpdate(updateVersion))
+	} else {
+		b.updates = append(b.updates, NewAddViewVersionUpdate(version))
+	}
+
+	b.lastAddedVersionID = &newVersionID
+	return newVersionID, nil
 }
 
-func (b *MetadataBuilder) SetLocation(_ string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) AddSchema(schema *iceberg.Schema) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	_, err := b.addSchema(schema)
+	if b.setErr(err) {
+		return b
+	}
+	return b
 }
 
-func (b *MetadataBuilder) SetProperties(_ map[string]string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) addSchema(schema *iceberg.Schema) (int, error) {
+	newSchemaID := b.reuseOrCreateNewSchemaID(schema)
+
+	if _, ok := b.schemasById[newSchemaID]; ok {
+		return newSchemaID, nil
+	}
+
+	newSchema := schema
+	// Build a fresh schema if we reset the ID
+	if schema.ID != newSchemaID {
+		newSchema = iceberg.NewSchemaWithIdentifiers(newSchemaID, schema.IdentifierFieldIDs, schema.Fields()...)
+	}
+
+	b.schemaList = append(b.schemaList, newSchema)
+	b.schemasById[newSchema.ID] = newSchema
+	b.updates = append(b.updates, NewAddSchemaUpdate(newSchema))
+	b.lastAddedSchemaID = &newSchemaID
+
+	return newSchemaID, nil
 }
 
-func (b *MetadataBuilder) RemoveProperties(_ []string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) RemoveProperties(keys []string) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if len(keys) == 0 {
+		return b
+	}
+
+	b.updates = append(b.updates, NewRemovePropertiesUpdate(keys))
+	for _, key := range keys {
+		delete(b.props, key)
+	}
+
+	return b
 }
 
-func (b *MetadataBuilder) AddVersion(_ *Version) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetCurrentVersionID(newVersionID int64) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if newVersionID == LastAddedID {
+		if b.lastAddedVersionID == nil {
+			b.err = errors.New("cannot set current version to last added, no version has been added")
+			return b
+		}
+		newVersionID = *b.lastAddedVersionID
+	}
+
+	if newVersionID == b.currentVersionID {
+		return b
+	}
+
+	_, ok := b.versionsById[newVersionID]
+	if !ok {
+		b.err = fmt.Errorf("cannot set current version to unknown version with id %d", newVersionID)
+		return b
+	}
+
+	if b.lastAddedVersionID != nil && *b.lastAddedVersionID == newVersionID {
+		b.updates = append(b.updates, NewSetCurrentVersionUpdate(LastAddedID))
+	} else {
+		b.updates = append(b.updates, NewSetCurrentVersionUpdate(newVersionID))
+	}
+	b.currentVersionID = newVersionID
+
+	// Set the current history entry.
+	// If the version was added in the current set of changes, use its timestamp,
+	// otherwise use system time.
+	updateTimestampMS := time.Now().UnixMilli()
+	for _, update := range b.updates {
+		if v, ok := update.(*addViewVersionUpdate); ok && v.Version.VersionID == newVersionID {
+			updateTimestampMS = v.Version.TimestampMS
+			break
+		}
+	}
+
+	b.versionHistoryEntry = &VersionLogEntry{
+		VersionID:   newVersionID,
+		TimestampMS: updateTimestampMS,
+	}
+
+	return b
 }
 
-func (b *MetadataBuilder) SetCurrentVersionID(_ int64) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetFormatVersion(formatVersion int) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if formatVersion < b.formatVersion {
+		b.err = fmt.Errorf("downgrading format version from %d to %d is not allowed",
+			b.formatVersion, formatVersion)
+		return b
+	}
+
+	if formatVersion > SupportedViewFormatVersion {
+		b.err = fmt.Errorf("unsupported format version %d", formatVersion)
+		return b
+	}
+
+	if formatVersion == b.formatVersion {
+		return b
+	}
+
+	b.updates = append(b.updates, NewUpgradeFormatVersionUpdate(formatVersion))
+	b.formatVersion = formatVersion
+
+	return b
+}
+
+func (b *MetadataBuilder) SetLoc(loc string) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if b.loc == loc {
+		return b
+	}
+
+	b.updates = append(b.updates, NewSetLocationUpdate(loc))
+	b.loc = loc
+
+	return b
+}
+
+func (b *MetadataBuilder) SetProperties(props iceberg.Properties) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if len(props) == 0 {
+		return b
+	}
+
+	b.updates = append(b.updates, NewSetPropertiesUpdate(props))
+	if b.props == nil {
+		b.props = props
+	} else {
+		maps.Copy(b.props, props)
+	}
+
+	return b
+}
+
+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
+	}
+
+	// Noop - same UUID does not generate a change.
+	if b.uuid == newUUID {
+		return b
+	}
+
+	if b.uuid != uuid.Nil {
+		b.err = errors.New("cannot reassign uuid")
+		return b
+	}
+
+	b.updates = append(b.updates, NewAssignUUIDUpdate(newUUID))
+	b.uuid = newUUID
+
+	return b
+}
+
+func (b *MetadataBuilder) Err() error {
+	return b.err
+}
+
+func (b *MetadataBuilder) buildMetadata(withUpdates bool) (*metadata, error) {
+	uuid_ := b.uuid
+	if uuid_ == uuid.Nil {
+		uuid_ = uuid.New()
+	}
+
+	md := &metadata{
+		FormatVersionValue:    b.formatVersion,
+		UUID:                  uuid_,
+		Loc:                   b.loc,
+		CurrentVersionIDValue: b.currentVersionID,
+		VersionList:           b.versionList,
+		VersionLogList:        b.versionLog,
+		SchemaList:            b.schemaList,
+		Props:                 b.props,
+	}
+	if withUpdates {
+		md.updates = b.updates
+	}
+	md.init()
+
+	return md, nil
+}
+
+// build builds the view metadata updates from the builder
+// if withUpdates is set to false, updates will not be included
+// in the returned Metadata, thus its Updates() will return an empty slice
+func (b *MetadataBuilder) build(withUpdates bool) (Metadata, error) {

Review Comment:
   The reason I omitted this is because AFAICT it was a configurable property as to whether this was permitted or not. I made the naive assumption that ultimately the catalog could enforce this depending on whether it's permitted or not but I'm not sure.
   
   ```java
   !PropertyUtil.propertyAsBoolean(
                 properties,
                 ViewProperties.REPLACE_DROP_DIALECT_ALLOWED,
                 ViewProperties.REPLACE_DROP_DIALECT_ALLOWED_DEFAULT)) {
           checkIfDialectIsDropped(previousViewVersion, versionsById.get(currentVersionId))
   ```



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["twuebi (via GitHub)" <gi...@apache.org>]

twuebi commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2542032670


##########
view/metadata.go:
##########
@@ -18,135 +18,523 @@
 package view
 
 import (
-	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"maps"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
 	"time"
 
 	"github.com/apache/iceberg-go"
-	"github.com/apache/iceberg-go/internal"
-	"github.com/apache/iceberg-go/io"
+	iceinternal "github.com/apache/iceberg-go/internal"
+	"github.com/apache/iceberg-go/table"
+
 	"github.com/google/uuid"
 )
 
-// LoadMetadata reads and parses a view metadata file from the specified location.
+var (
+	ErrInvalidViewMetadata              = errors.New("invalid view metadata")
+	ErrInvalidViewMetadataFormatVersion = errors.New("invalid or missing format-version in view metadata")
+)
+
+const (
+	// LastAddedID is used in place of ID fields (e.g. schema, version) to indicate that
+	// the last added instance of that type should be used.
+	LastAddedID                = -1
+	SupportedViewFormatVersion = 1
+	DefaultViewFormatVersion   = SupportedViewFormatVersion
+)
+
+const (
+	initialSchemaID = 0
+
+	viewEngineProperty = "engine-name"
+	defaultViewEngine  = "iceberg-go"
+)
+
+// Metadata for an iceberg view as specified in the Iceberg spec
+// https://iceberg.apache.org/view-spec/
+type Metadata interface {
+	// FormatVersion indicates the version of this metadata, 1 for V1
+	FormatVersion() int
+	// ViewUUID returns a UUID that identifies the view, generated when the
+	// view is created. Implementations must throw an exception if a view's
+	// UUID does not match the expected UUID after refreshing metadata.
+	ViewUUID() uuid.UUID
+	// Location is the table's base location. This is used by writers to determine
+	// where to store data files, manifest files, and table metadata files.
+	Location() string
+	// Schemas returns the list of view schemas
+	Schemas() []*iceberg.Schema
+	// CurrentVersionID returns the ID of the current version of the view (version-id)
+	CurrentVersionID() int64
+	// CurrentVersion returns the current version of the view
+	CurrentVersion() *Version
+	// CurrentSchemaID returns the ID of the current schema
+	CurrentSchemaID() int
+	// CurrentSchema returns the current schema of the view
+	CurrentSchema() *iceberg.Schema
+	// SchemasByID returns a map of schema IDs to schemas
+	SchemasByID() map[int]*iceberg.Schema
+	// Versions returns the list of view versions
+	Versions() []*Version
+	// VersionLog returns a list of version log entries
+	// with the timestamp and version-id for every change to current-version-id
+	VersionLog() []VersionLogEntry
+	// Properties is a string to string map of view properties.
+	Properties() iceberg.Properties
+
+	// Updates returns the list of metadata updates used to build this metadata
+	Updates() Updates
+
+	Equals(Metadata) bool
+}
+
+// VersionSummary is string to string map of summary metadata about a view's version
+type VersionSummary map[string]string
+
+// Representation is a struct containing information about a view's representation
+// https://iceberg.apache.org/view-spec/#sql-representation
+type Representation struct {
+	// Must be sql
+	Type string `json:"type"`
+	// A SQL SELECT statement
+	Sql string `json:"sql"`
+	// The dialect of the sql SELECT statement (e.g., "trino" or "spark")
+	Dialect string `json:"dialect"`
+}
+type Representations []Representation
+
+func NewRepresentation(sql string, dialect string) Representation {
+	return Representation{
+		Type:    "sql",
+		Sql:     sql,
+		Dialect: dialect,
+	}
+}
+
+type Version struct {
+	// ID for the version
+	VersionID int64 `json:"version-id"`
+	// ID of the schema for the view version
+	SchemaID int `json:"schema-id"`
+	// Timestamp when the version was created (ms from epoch)
+	TimestampMS int64 `json:"timestamp-ms"`
+	// A string to string map of summary metadata about the version
+	Summary VersionSummary `json:"summary"`
+	// A list of representations for the view definition
+	Representations []Representation `json:"representations"`
+	// The default view namespace (as a list of strings)
+	DefaultNamespace table.Identifier `json:"default-namespace"`
+	// An (optional) default catalog name for querying the view
+	DefaultCatalog string `json:"default-catalog,omitempty"`
+}
+
+type VersionOpt func(*Version)
+
+func WithVersionSummary(summary VersionSummary) VersionOpt {
+	return func(v *Version) {
+		if v.Summary == nil {
+			v.Summary = summary
+		} else {
+			maps.Copy(v.Summary, summary)
+		}
+	}
+}
+
+func WithDefaultViewCatalog(catalogName string) func(*Version) {
+	return func(version *Version) {
+		version.DefaultCatalog = catalogName
+	}
+}
+
+func WithTimestampMS(timestampMS int64) VersionOpt {
+	return func(version *Version) {
+		version.TimestampMS = timestampMS
+	}
+}
+
+// NewVersion creates a Version instance from the provided parameters.
+// If building updates using the MetadataBuilder, and one desires to use the last
+// added schema ID, one should use the LastAddedID constant as the provided schemaID
 //
-// Returns the loaded Metadata or an error if the file cannot be read or parsed.
-func LoadMetadata(ctx context.Context,
-	props iceberg.Properties,
-	metadataLocation string,
-	name string,
-	namespace string,
-) (_ Metadata, err error) {
-	fs, err := io.LoadFS(ctx, props, metadataLocation)
-	if err != nil {
-		return nil, fmt.Errorf("error loading view metadata: %w", err)
+// Note that NewVersion automatically seeds TimestampMS to time.Now().UnixMilli(),
+// and one should use the option WithTimestampMS to override this behavior.
+func NewVersion(id int64, schemaID int, representations []Representation, defaultNS table.Identifier, opts ...VersionOpt) (*Version, error) {
+	if id < 1 {
+		return nil, errors.New("id should be greater than 0")
 	}
 
-	inputFile, err := fs.Open(metadataLocation)
-	if err != nil {
-		return nil, fmt.Errorf("%s.%s: error encountered loading view metadata: %w", namespace, name, err)
+	if len(representations) == 0 {
+		return nil, errors.New("invalid view version: must have at least one representation")
 	}
-	defer internal.CheckedClose(inputFile, &err)
 
-	var m metadata
-	if err := json.NewDecoder(inputFile).Decode(&m); err != nil {
-		return nil, fmt.Errorf("error encountered decoding view metadata: %w", err)
+	version := &Version{
+		VersionID:        id,
+		SchemaID:         schemaID,
+		Representations:  representations,
+		Summary:          VersionSummary{viewEngineProperty: defaultViewEngine},
+		TimestampMS:      time.Now().UnixMilli(),
+		DefaultNamespace: defaultNS,
 	}
 
-	return &m, nil
+	for _, opt := range opts {
+		opt(version)
+	}
+
+	return version, nil
 }
 
-// NewMetadata returns a view metadata for a given version.
-func NewMetadata(schema *iceberg.Schema,
-	version Version,
-	loc string,
-	props iceberg.Properties,
-) (Metadata, error) {
-	timestampMs := time.Now().UnixMilli()
+// NewVersionFromSQL creates a new Version with a single representation
+// using the provided SQL and dialect "default".
+func NewVersionFromSQL(id int64, schemaID int, sql string, defaultNS table.Identifier, opts ...VersionOpt) (*Version, error) {
+	return NewVersion(id, schemaID, Representations{NewRepresentation(sql, "default")}, defaultNS, opts...)
+}
 
-	if version.SchemaID != schema.ID {
-		return nil, fmt.Errorf("%w: version.SchemaID does not match schema.ID", iceberg.ErrInvalidArgument)
+// Equals checks whether the other Version would behave the same
+// while ignoring the view version id, and the creation timestamp
+func (v *Version) Equals(other *Version) bool {
+	return v.SchemaID == other.SchemaID &&
+		v.DefaultCatalog == other.DefaultCatalog &&
+		slices.Equal(v.DefaultNamespace, other.DefaultNamespace) &&
+		maps.Equal(v.Summary, other.Summary) &&
+		slices.Equal(v.Representations, other.Representations)
+}
+
+func (v *Version) Clone() *Version {
+	if v == nil {
+		return nil
 	}
 
-	return &metadata{
-		UUID:             uuid.NewString(),
-		FmtVersion:       1,
-		Loc:              loc,
-		SchemaList:       []*iceberg.Schema{schema},
-		CurrentVersionId: version.VersionID,
-		VersionList:      []Version{version},
-		VersionLogList: []VersionLogEntry{
-			{
-				TimestampMs: timestampMs,
-				VersionID:   version.VersionID,
-			},
-		},
-		Props: props,
-	}, nil
+	cloned := *v
+	cloned.Summary = maps.Clone(v.Summary)
+	cloned.Representations = slices.Clone(v.Representations)
+	cloned.DefaultNamespace = slices.Clone(v.DefaultNamespace)
+	return &cloned
 }
 
-// CreateMetadata creates a new view metadata file and writes it to storage.
-//
-// Returns the full path to the created metadata file, or an error if creation fails.
-//
-// Note: This function only supports creating new views with format version 1.
-// It does not support updating existing view metadata.
-func CreateMetadata(
-	ctx context.Context,
-	catalogName string,
-	nsIdent []string,
-	schema *iceberg.Schema,
-	viewSQL string,
-	loc string,
-	props iceberg.Properties,
-) (metadataLocation string, err error) {
-	versionId := int64(1)
-	timestampMs := time.Now().UnixMilli()
-
-	viewVersion := Version{
-		VersionID:   versionId,
-		TimestampMs: timestampMs,
-		SchemaID:    schema.ID,
-		Summary:     map[string]string{"sql": viewSQL},
-		Representations: []SQLRepresentation{
-			{Type: "sql", SQL: viewSQL, Dialect: "default"},
-		},
-		DefaultCatalog:   &catalogName,
-		DefaultNamespace: nsIdent,
-	}
-
-	metadataLocation = loc + "/metadata/view-" + uuid.New().String() + ".metadata.json"
-	viewMetadata, err := NewMetadata(schema, viewVersion, loc, props)
+type VersionLogEntry struct {
+	// Timestamp when the view's current-version-id was updated (ms from epoch)
+	TimestampMS int64 `json:"timestamp-ms"`
+	// ID that current-version-id was set to
+	VersionID int64 `json:"version-id"`
+}
+
+// ParseMetadata parses json metadata 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 "", fmt.Errorf("failed to create view metadata: %w", err)
+		return nil, err
 	}
 
-	viewMetadataBytes, err := json.Marshal(viewMetadata)
-	if err != nil {
-		return "", fmt.Errorf("failed to marshal view metadata: %w", 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) {
+	ver := struct {
+		FormatVersion int `json:"format-version"`
+	}{}
+	if err := json.Unmarshal(b, &ver); err != nil {
+		return nil, err
 	}
 
-	fs, err := io.LoadFS(ctx, props, metadataLocation)
-	if err != nil {
-		return "", fmt.Errorf("failed to load filesystem for view metadata: %w", err)
+	var ret Metadata
+	ret = &metadata{}
+	return ret, json.Unmarshal(b, ret)
+}
+
+// indexBy indexes a slice into a map, using a provided extractKey function
+// The extractKey function will be called on each item in the slice and assign the
+// item as a value for that key in the resultant map.
+func indexBy[T any, K comparable](s []T, extractKey func(T) K) map[K]T {
+	index := make(map[K]T)
+	for _, v := range s {
+		index[extractKey(v)] = v
+	}
+	return index
+}
+
+// Cloner is an interface which implements a Clone method for deep copying itself
+type cloner[T any] interface {
+	// Clone returns a deep copy of the underlying object
+	Clone() T
+}
+
+// cloneSlice returns a deep-clone of a Slice of elements implementing Cloner
+func cloneSlice[T cloner[T]](val []T) []T {
+	cloned := make([]T, len(val))
+	for i, elem := range val {
+		cloned[i] = elem.Clone()
+	}
+	return cloned
+}
+
+// https://iceberg.apache.org/view-spec/
+type metadata struct {
+	FormatVersionValue    int                `json:"format-version"`
+	UUID                  uuid.UUID          `json:"view-uuid"`
+	Loc                   string             `json:"location"`
+	CurrentVersionIDValue int64              `json:"current-version-id"`
+	VersionList           []*Version         `json:"versions"`
+	SchemaList            []*iceberg.Schema  `json:"schemas"`
+	VersionLogList        []VersionLogEntry  `json:"version-log"`
+	Props                 iceberg.Properties `json:"properties,omitempty"`
+
+	// updates from builder if applicable
+	updates Updates
+
+	// cached lookup helpers, must be initialized in init()
+	lazyVersionsByID func() map[int64]*Version
+	lazySchemasByID  func() map[int]*iceberg.Schema
+}
+
+func (m *metadata) Equals(other Metadata) bool {
+	if other == nil {
+		return false
+	}
+
+	if m == other {
+		return true
+	}
+
+	return m.UUID == other.ViewUUID() &&
+		m.Loc == other.Location() &&
+		m.FormatVersionValue == other.FormatVersion() &&
+		iceinternal.SliceEqualHelper(m.Schemas(), other.Schemas()) &&
+		iceinternal.SliceEqualHelper(m.Versions(), other.Versions()) &&
+		m.CurrentVersionIDValue == other.CurrentVersionID() &&
+		slices.Equal(m.VersionLogList, other.VersionLog())
+}
+
+func (m *metadata) Updates() Updates { return m.updates }
+
+func (m *metadata) FormatVersion() int         { return m.FormatVersionValue }
+func (m *metadata) ViewUUID() uuid.UUID        { return m.UUID }
+func (m *metadata) Location() string           { return m.Loc }
+func (m *metadata) Versions() []*Version       { return m.VersionList }
+func (m *metadata) Schemas() []*iceberg.Schema { return m.SchemaList }
+func (m *metadata) SchemasByID() map[int]*iceberg.Schema {
+	return maps.Clone(m.lazySchemasByID())
+}
+
+func (m *metadata) CurrentVersionID() int64 {
+	return m.CurrentVersionIDValue
+}
+
+func (m *metadata) CurrentVersion() *Version {
+	version, ok := m.lazyVersionsByID()[m.CurrentVersionIDValue]
+	if !ok {
+		panic("current version not found")
 	}
+	return version
+}
+
+func (m *metadata) CurrentSchemaID() int {
+	return m.CurrentVersion().SchemaID
+}
 
-	wfs, ok := fs.(io.WriteFileIO)
+func (m *metadata) CurrentSchema() *iceberg.Schema {
+	schema, ok := m.lazySchemasByID()[m.CurrentSchemaID()]
 	if !ok {
-		return "", errors.New("filesystem IO does not support writing")
+		panic("current schema not found")
+	}
+	return schema
+}
+
+func (m *metadata) VersionLog() []VersionLogEntry {
+	return m.VersionLogList
+}
+
+func (m *metadata) Properties() iceberg.Properties {
+	return m.Props
+}
+
+func (m *metadata) validate() error {
+	if m.Loc == "" {
+		return fmt.Errorf("%w: location is required", ErrInvalidViewMetadata)
+	}
+
+	if m.UUID == uuid.Nil {
+		return fmt.Errorf("%w: view-uuid is required", ErrInvalidViewMetadata)
+	}

Review Comment:
   Maybe a pedantic - 00000....00000 is a valid uuid for iceberg, I don't think we should reject this here



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["twuebi (via GitHub)" <gi...@apache.org>]

twuebi commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2543689849


##########
view/metadata.go:
##########
@@ -18,135 +18,523 @@
 package view
 
 import (
-	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"maps"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
 	"time"
 
 	"github.com/apache/iceberg-go"
-	"github.com/apache/iceberg-go/internal"
-	"github.com/apache/iceberg-go/io"
+	iceinternal "github.com/apache/iceberg-go/internal"
+	"github.com/apache/iceberg-go/table"
+
 	"github.com/google/uuid"
 )
 
-// LoadMetadata reads and parses a view metadata file from the specified location.
+var (
+	ErrInvalidViewMetadata              = errors.New("invalid view metadata")
+	ErrInvalidViewMetadataFormatVersion = errors.New("invalid or missing format-version in view metadata")
+)
+
+const (
+	// LastAddedID is used in place of ID fields (e.g. schema, version) to indicate that
+	// the last added instance of that type should be used.
+	LastAddedID                = -1
+	SupportedViewFormatVersion = 1
+	DefaultViewFormatVersion   = SupportedViewFormatVersion
+)
+
+const (
+	initialSchemaID = 0
+
+	viewEngineProperty = "engine-name"
+	defaultViewEngine  = "iceberg-go"
+)
+
+// Metadata for an iceberg view as specified in the Iceberg spec
+// https://iceberg.apache.org/view-spec/
+type Metadata interface {
+	// FormatVersion indicates the version of this metadata, 1 for V1
+	FormatVersion() int
+	// ViewUUID returns a UUID that identifies the view, generated when the
+	// view is created. Implementations must throw an exception if a view's
+	// UUID does not match the expected UUID after refreshing metadata.
+	ViewUUID() uuid.UUID
+	// Location is the table's base location. This is used by writers to determine
+	// where to store data files, manifest files, and table metadata files.
+	Location() string
+	// Schemas returns the list of view schemas
+	Schemas() []*iceberg.Schema
+	// CurrentVersionID returns the ID of the current version of the view (version-id)
+	CurrentVersionID() int64
+	// CurrentVersion returns the current version of the view
+	CurrentVersion() *Version
+	// CurrentSchemaID returns the ID of the current schema
+	CurrentSchemaID() int
+	// CurrentSchema returns the current schema of the view
+	CurrentSchema() *iceberg.Schema
+	// SchemasByID returns a map of schema IDs to schemas
+	SchemasByID() map[int]*iceberg.Schema
+	// Versions returns the list of view versions
+	Versions() []*Version
+	// VersionLog returns a list of version log entries
+	// with the timestamp and version-id for every change to current-version-id
+	VersionLog() []VersionLogEntry
+	// Properties is a string to string map of view properties.
+	Properties() iceberg.Properties
+
+	// Updates returns the list of metadata updates used to build this metadata
+	Updates() Updates

Review Comment:
   I borrowed the Result struct from Rust: https://github.com/lakekeeper/iceberg-rust/blob/446665652eedc9cd23b9a3ef1ed168e6e5bc36e6/crates/iceberg/src/spec/table_metadata_builder.rs#L61-L68



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2543874008


##########
view/metadata.go:
##########
@@ -18,135 +18,523 @@
 package view
 
 import (
-	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"maps"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
 	"time"
 
 	"github.com/apache/iceberg-go"
-	"github.com/apache/iceberg-go/internal"
-	"github.com/apache/iceberg-go/io"
+	iceinternal "github.com/apache/iceberg-go/internal"
+	"github.com/apache/iceberg-go/table"
+
 	"github.com/google/uuid"
 )
 
-// LoadMetadata reads and parses a view metadata file from the specified location.
+var (
+	ErrInvalidViewMetadata              = errors.New("invalid view metadata")
+	ErrInvalidViewMetadataFormatVersion = errors.New("invalid or missing format-version in view metadata")
+)
+
+const (
+	// LastAddedID is used in place of ID fields (e.g. schema, version) to indicate that
+	// the last added instance of that type should be used.
+	LastAddedID                = -1
+	SupportedViewFormatVersion = 1
+	DefaultViewFormatVersion   = SupportedViewFormatVersion
+)
+
+const (
+	initialSchemaID = 0
+
+	viewEngineProperty = "engine-name"
+	defaultViewEngine  = "iceberg-go"
+)
+
+// Metadata for an iceberg view as specified in the Iceberg spec
+// https://iceberg.apache.org/view-spec/
+type Metadata interface {
+	// FormatVersion indicates the version of this metadata, 1 for V1
+	FormatVersion() int
+	// ViewUUID returns a UUID that identifies the view, generated when the
+	// view is created. Implementations must throw an exception if a view's
+	// UUID does not match the expected UUID after refreshing metadata.
+	ViewUUID() uuid.UUID
+	// Location is the table's base location. This is used by writers to determine
+	// where to store data files, manifest files, and table metadata files.
+	Location() string
+	// Schemas returns the list of view schemas
+	Schemas() []*iceberg.Schema
+	// CurrentVersionID returns the ID of the current version of the view (version-id)
+	CurrentVersionID() int64
+	// CurrentVersion returns the current version of the view
+	CurrentVersion() *Version
+	// CurrentSchemaID returns the ID of the current schema
+	CurrentSchemaID() int
+	// CurrentSchema returns the current schema of the view
+	CurrentSchema() *iceberg.Schema
+	// SchemasByID returns a map of schema IDs to schemas
+	SchemasByID() map[int]*iceberg.Schema
+	// Versions returns the list of view versions
+	Versions() []*Version
+	// VersionLog returns a list of version log entries
+	// with the timestamp and version-id for every change to current-version-id
+	VersionLog() []VersionLogEntry
+	// Properties is a string to string map of view properties.
+	Properties() iceberg.Properties
+
+	// Updates returns the list of metadata updates used to build this metadata
+	Updates() Updates

Review Comment:
   Sounds good, I'll go with this approach then. 



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#issuecomment-3603144325

   > Sure, I'm wondering if there's a good reason not to do it here in the view constructor, as in if there's any harm in it.
   
   Meaning in `NewViewVersion`? I was thinking about that. I guess if users in general should be building updates using the `MetadataBuilder`, it doesn't matter if we default the ID in that constructor to `1` since we reassign in the builder.
   
   How about I default it to `1` there and add a variadic version opt `WithViewVersionID` in case there are reasons to set manually? 


-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#issuecomment-3617767945

   > Thanks for all the work here! and for the reviews from @twuebi and @alessandro-nori
   
   Thank you all!


-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#issuecomment-3607582611

   @twuebi for now just enforced in the REST handler. LMK if you want me to take the approach I described above of dropping the positional param for Version ID and making it default 1 with variadic opt. 
   
   @zeroshade any other blockers here I should look at in the meantime? 


-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["zeroshade (via GitHub)" <gi...@apache.org>]

zeroshade merged PR #629:
URL: https://github.com/apache/iceberg-go/pull/629


-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["twuebi (via GitHub)" <gi...@apache.org>]

twuebi commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2542032670


##########
view/metadata.go:
##########
@@ -18,135 +18,523 @@
 package view
 
 import (
-	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"maps"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
 	"time"
 
 	"github.com/apache/iceberg-go"
-	"github.com/apache/iceberg-go/internal"
-	"github.com/apache/iceberg-go/io"
+	iceinternal "github.com/apache/iceberg-go/internal"
+	"github.com/apache/iceberg-go/table"
+
 	"github.com/google/uuid"
 )
 
-// LoadMetadata reads and parses a view metadata file from the specified location.
+var (
+	ErrInvalidViewMetadata              = errors.New("invalid view metadata")
+	ErrInvalidViewMetadataFormatVersion = errors.New("invalid or missing format-version in view metadata")
+)
+
+const (
+	// LastAddedID is used in place of ID fields (e.g. schema, version) to indicate that
+	// the last added instance of that type should be used.
+	LastAddedID                = -1
+	SupportedViewFormatVersion = 1
+	DefaultViewFormatVersion   = SupportedViewFormatVersion
+)
+
+const (
+	initialSchemaID = 0
+
+	viewEngineProperty = "engine-name"
+	defaultViewEngine  = "iceberg-go"
+)
+
+// Metadata for an iceberg view as specified in the Iceberg spec
+// https://iceberg.apache.org/view-spec/
+type Metadata interface {
+	// FormatVersion indicates the version of this metadata, 1 for V1
+	FormatVersion() int
+	// ViewUUID returns a UUID that identifies the view, generated when the
+	// view is created. Implementations must throw an exception if a view's
+	// UUID does not match the expected UUID after refreshing metadata.
+	ViewUUID() uuid.UUID
+	// Location is the table's base location. This is used by writers to determine
+	// where to store data files, manifest files, and table metadata files.
+	Location() string
+	// Schemas returns the list of view schemas
+	Schemas() []*iceberg.Schema
+	// CurrentVersionID returns the ID of the current version of the view (version-id)
+	CurrentVersionID() int64
+	// CurrentVersion returns the current version of the view
+	CurrentVersion() *Version
+	// CurrentSchemaID returns the ID of the current schema
+	CurrentSchemaID() int
+	// CurrentSchema returns the current schema of the view
+	CurrentSchema() *iceberg.Schema
+	// SchemasByID returns a map of schema IDs to schemas
+	SchemasByID() map[int]*iceberg.Schema
+	// Versions returns the list of view versions
+	Versions() []*Version
+	// VersionLog returns a list of version log entries
+	// with the timestamp and version-id for every change to current-version-id
+	VersionLog() []VersionLogEntry
+	// Properties is a string to string map of view properties.
+	Properties() iceberg.Properties
+
+	// Updates returns the list of metadata updates used to build this metadata
+	Updates() Updates
+
+	Equals(Metadata) bool
+}
+
+// VersionSummary is string to string map of summary metadata about a view's version
+type VersionSummary map[string]string
+
+// Representation is a struct containing information about a view's representation
+// https://iceberg.apache.org/view-spec/#sql-representation
+type Representation struct {
+	// Must be sql
+	Type string `json:"type"`
+	// A SQL SELECT statement
+	Sql string `json:"sql"`
+	// The dialect of the sql SELECT statement (e.g., "trino" or "spark")
+	Dialect string `json:"dialect"`
+}
+type Representations []Representation
+
+func NewRepresentation(sql string, dialect string) Representation {
+	return Representation{
+		Type:    "sql",
+		Sql:     sql,
+		Dialect: dialect,
+	}
+}
+
+type Version struct {
+	// ID for the version
+	VersionID int64 `json:"version-id"`
+	// ID of the schema for the view version
+	SchemaID int `json:"schema-id"`
+	// Timestamp when the version was created (ms from epoch)
+	TimestampMS int64 `json:"timestamp-ms"`
+	// A string to string map of summary metadata about the version
+	Summary VersionSummary `json:"summary"`
+	// A list of representations for the view definition
+	Representations []Representation `json:"representations"`
+	// The default view namespace (as a list of strings)
+	DefaultNamespace table.Identifier `json:"default-namespace"`
+	// An (optional) default catalog name for querying the view
+	DefaultCatalog string `json:"default-catalog,omitempty"`
+}
+
+type VersionOpt func(*Version)
+
+func WithVersionSummary(summary VersionSummary) VersionOpt {
+	return func(v *Version) {
+		if v.Summary == nil {
+			v.Summary = summary
+		} else {
+			maps.Copy(v.Summary, summary)
+		}
+	}
+}
+
+func WithDefaultViewCatalog(catalogName string) func(*Version) {
+	return func(version *Version) {
+		version.DefaultCatalog = catalogName
+	}
+}
+
+func WithTimestampMS(timestampMS int64) VersionOpt {
+	return func(version *Version) {
+		version.TimestampMS = timestampMS
+	}
+}
+
+// NewVersion creates a Version instance from the provided parameters.
+// If building updates using the MetadataBuilder, and one desires to use the last
+// added schema ID, one should use the LastAddedID constant as the provided schemaID
 //
-// Returns the loaded Metadata or an error if the file cannot be read or parsed.
-func LoadMetadata(ctx context.Context,
-	props iceberg.Properties,
-	metadataLocation string,
-	name string,
-	namespace string,
-) (_ Metadata, err error) {
-	fs, err := io.LoadFS(ctx, props, metadataLocation)
-	if err != nil {
-		return nil, fmt.Errorf("error loading view metadata: %w", err)
+// Note that NewVersion automatically seeds TimestampMS to time.Now().UnixMilli(),
+// and one should use the option WithTimestampMS to override this behavior.
+func NewVersion(id int64, schemaID int, representations []Representation, defaultNS table.Identifier, opts ...VersionOpt) (*Version, error) {
+	if id < 1 {
+		return nil, errors.New("id should be greater than 0")
 	}
 
-	inputFile, err := fs.Open(metadataLocation)
-	if err != nil {
-		return nil, fmt.Errorf("%s.%s: error encountered loading view metadata: %w", namespace, name, err)
+	if len(representations) == 0 {
+		return nil, errors.New("invalid view version: must have at least one representation")
 	}
-	defer internal.CheckedClose(inputFile, &err)
 
-	var m metadata
-	if err := json.NewDecoder(inputFile).Decode(&m); err != nil {
-		return nil, fmt.Errorf("error encountered decoding view metadata: %w", err)
+	version := &Version{
+		VersionID:        id,
+		SchemaID:         schemaID,
+		Representations:  representations,
+		Summary:          VersionSummary{viewEngineProperty: defaultViewEngine},
+		TimestampMS:      time.Now().UnixMilli(),
+		DefaultNamespace: defaultNS,
 	}
 
-	return &m, nil
+	for _, opt := range opts {
+		opt(version)
+	}
+
+	return version, nil
 }
 
-// NewMetadata returns a view metadata for a given version.
-func NewMetadata(schema *iceberg.Schema,
-	version Version,
-	loc string,
-	props iceberg.Properties,
-) (Metadata, error) {
-	timestampMs := time.Now().UnixMilli()
+// NewVersionFromSQL creates a new Version with a single representation
+// using the provided SQL and dialect "default".
+func NewVersionFromSQL(id int64, schemaID int, sql string, defaultNS table.Identifier, opts ...VersionOpt) (*Version, error) {
+	return NewVersion(id, schemaID, Representations{NewRepresentation(sql, "default")}, defaultNS, opts...)
+}
 
-	if version.SchemaID != schema.ID {
-		return nil, fmt.Errorf("%w: version.SchemaID does not match schema.ID", iceberg.ErrInvalidArgument)
+// Equals checks whether the other Version would behave the same
+// while ignoring the view version id, and the creation timestamp
+func (v *Version) Equals(other *Version) bool {
+	return v.SchemaID == other.SchemaID &&
+		v.DefaultCatalog == other.DefaultCatalog &&
+		slices.Equal(v.DefaultNamespace, other.DefaultNamespace) &&
+		maps.Equal(v.Summary, other.Summary) &&
+		slices.Equal(v.Representations, other.Representations)
+}
+
+func (v *Version) Clone() *Version {
+	if v == nil {
+		return nil
 	}
 
-	return &metadata{
-		UUID:             uuid.NewString(),
-		FmtVersion:       1,
-		Loc:              loc,
-		SchemaList:       []*iceberg.Schema{schema},
-		CurrentVersionId: version.VersionID,
-		VersionList:      []Version{version},
-		VersionLogList: []VersionLogEntry{
-			{
-				TimestampMs: timestampMs,
-				VersionID:   version.VersionID,
-			},
-		},
-		Props: props,
-	}, nil
+	cloned := *v
+	cloned.Summary = maps.Clone(v.Summary)
+	cloned.Representations = slices.Clone(v.Representations)
+	cloned.DefaultNamespace = slices.Clone(v.DefaultNamespace)
+	return &cloned
 }
 
-// CreateMetadata creates a new view metadata file and writes it to storage.
-//
-// Returns the full path to the created metadata file, or an error if creation fails.
-//
-// Note: This function only supports creating new views with format version 1.
-// It does not support updating existing view metadata.
-func CreateMetadata(
-	ctx context.Context,
-	catalogName string,
-	nsIdent []string,
-	schema *iceberg.Schema,
-	viewSQL string,
-	loc string,
-	props iceberg.Properties,
-) (metadataLocation string, err error) {
-	versionId := int64(1)
-	timestampMs := time.Now().UnixMilli()
-
-	viewVersion := Version{
-		VersionID:   versionId,
-		TimestampMs: timestampMs,
-		SchemaID:    schema.ID,
-		Summary:     map[string]string{"sql": viewSQL},
-		Representations: []SQLRepresentation{
-			{Type: "sql", SQL: viewSQL, Dialect: "default"},
-		},
-		DefaultCatalog:   &catalogName,
-		DefaultNamespace: nsIdent,
-	}
-
-	metadataLocation = loc + "/metadata/view-" + uuid.New().String() + ".metadata.json"
-	viewMetadata, err := NewMetadata(schema, viewVersion, loc, props)
+type VersionLogEntry struct {
+	// Timestamp when the view's current-version-id was updated (ms from epoch)
+	TimestampMS int64 `json:"timestamp-ms"`
+	// ID that current-version-id was set to
+	VersionID int64 `json:"version-id"`
+}
+
+// ParseMetadata parses json metadata 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 "", fmt.Errorf("failed to create view metadata: %w", err)
+		return nil, err
 	}
 
-	viewMetadataBytes, err := json.Marshal(viewMetadata)
-	if err != nil {
-		return "", fmt.Errorf("failed to marshal view metadata: %w", 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) {
+	ver := struct {
+		FormatVersion int `json:"format-version"`
+	}{}
+	if err := json.Unmarshal(b, &ver); err != nil {
+		return nil, err
 	}
 
-	fs, err := io.LoadFS(ctx, props, metadataLocation)
-	if err != nil {
-		return "", fmt.Errorf("failed to load filesystem for view metadata: %w", err)
+	var ret Metadata
+	ret = &metadata{}
+	return ret, json.Unmarshal(b, ret)
+}
+
+// indexBy indexes a slice into a map, using a provided extractKey function
+// The extractKey function will be called on each item in the slice and assign the
+// item as a value for that key in the resultant map.
+func indexBy[T any, K comparable](s []T, extractKey func(T) K) map[K]T {
+	index := make(map[K]T)
+	for _, v := range s {
+		index[extractKey(v)] = v
+	}
+	return index
+}
+
+// Cloner is an interface which implements a Clone method for deep copying itself
+type cloner[T any] interface {
+	// Clone returns a deep copy of the underlying object
+	Clone() T
+}
+
+// cloneSlice returns a deep-clone of a Slice of elements implementing Cloner
+func cloneSlice[T cloner[T]](val []T) []T {
+	cloned := make([]T, len(val))
+	for i, elem := range val {
+		cloned[i] = elem.Clone()
+	}
+	return cloned
+}
+
+// https://iceberg.apache.org/view-spec/
+type metadata struct {
+	FormatVersionValue    int                `json:"format-version"`
+	UUID                  uuid.UUID          `json:"view-uuid"`
+	Loc                   string             `json:"location"`
+	CurrentVersionIDValue int64              `json:"current-version-id"`
+	VersionList           []*Version         `json:"versions"`
+	SchemaList            []*iceberg.Schema  `json:"schemas"`
+	VersionLogList        []VersionLogEntry  `json:"version-log"`
+	Props                 iceberg.Properties `json:"properties,omitempty"`
+
+	// updates from builder if applicable
+	updates Updates
+
+	// cached lookup helpers, must be initialized in init()
+	lazyVersionsByID func() map[int64]*Version
+	lazySchemasByID  func() map[int]*iceberg.Schema
+}
+
+func (m *metadata) Equals(other Metadata) bool {
+	if other == nil {
+		return false
+	}
+
+	if m == other {
+		return true
+	}
+
+	return m.UUID == other.ViewUUID() &&
+		m.Loc == other.Location() &&
+		m.FormatVersionValue == other.FormatVersion() &&
+		iceinternal.SliceEqualHelper(m.Schemas(), other.Schemas()) &&
+		iceinternal.SliceEqualHelper(m.Versions(), other.Versions()) &&
+		m.CurrentVersionIDValue == other.CurrentVersionID() &&
+		slices.Equal(m.VersionLogList, other.VersionLog())
+}
+
+func (m *metadata) Updates() Updates { return m.updates }
+
+func (m *metadata) FormatVersion() int         { return m.FormatVersionValue }
+func (m *metadata) ViewUUID() uuid.UUID        { return m.UUID }
+func (m *metadata) Location() string           { return m.Loc }
+func (m *metadata) Versions() []*Version       { return m.VersionList }
+func (m *metadata) Schemas() []*iceberg.Schema { return m.SchemaList }
+func (m *metadata) SchemasByID() map[int]*iceberg.Schema {
+	return maps.Clone(m.lazySchemasByID())
+}
+
+func (m *metadata) CurrentVersionID() int64 {
+	return m.CurrentVersionIDValue
+}
+
+func (m *metadata) CurrentVersion() *Version {
+	version, ok := m.lazyVersionsByID()[m.CurrentVersionIDValue]
+	if !ok {
+		panic("current version not found")
 	}
+	return version
+}
+
+func (m *metadata) CurrentSchemaID() int {
+	return m.CurrentVersion().SchemaID
+}
 
-	wfs, ok := fs.(io.WriteFileIO)
+func (m *metadata) CurrentSchema() *iceberg.Schema {
+	schema, ok := m.lazySchemasByID()[m.CurrentSchemaID()]
 	if !ok {
-		return "", errors.New("filesystem IO does not support writing")
+		panic("current schema not found")
+	}
+	return schema
+}
+
+func (m *metadata) VersionLog() []VersionLogEntry {
+	return m.VersionLogList
+}
+
+func (m *metadata) Properties() iceberg.Properties {
+	return m.Props
+}
+
+func (m *metadata) validate() error {
+	if m.Loc == "" {
+		return fmt.Errorf("%w: location is required", ErrInvalidViewMetadata)
+	}
+
+	if m.UUID == uuid.Nil {
+		return fmt.Errorf("%w: view-uuid is required", ErrInvalidViewMetadata)
+	}

Review Comment:
   Maybe a pedantic - 00000....00000 is a valid for iceberg, I don't think we should reject this here



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["twuebi (via GitHub)" <gi...@apache.org>]

twuebi commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2542058280


##########
view/metadata_builder.go:
##########
@@ -1,57 +1,468 @@
-// 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 view
 
 import (
+	"errors"
+	"fmt"
+	"maps"
+	"slices"
+	"strings"
+	"time"
+
 	"github.com/apache/iceberg-go"
+
+	"github.com/google/uuid"
 )
 
-// MetadataBuilder is a struct used for building and updating Iceberg view metadata.
-type MetadataBuilder struct{}
+type MetadataBuilder struct {
+	base    Metadata
+	updates Updates
+
+	// common fields
+	formatVersion    int
+	uuid             uuid.UUID
+	loc              string
+	schemaList       []*iceberg.Schema
+	versionList      []*Version
+	currentVersionID int64
+	versionLog       []VersionLogEntry
+	props            iceberg.Properties
+
+	// lookup maps
+	versionsById map[int64]*Version
+	schemasById  map[int]*iceberg.Schema
+
+	// update tracking
+	versionHistoryEntry *VersionLogEntry
+	lastAddedVersionID  *int64
+	lastAddedSchemaID   *int
+
+	// error tracking for build chaining
+	// if set, subsequent operations become a noop
+	err error
+}
+
+func NewMetadataBuilder() (*MetadataBuilder, error) {
+	return &MetadataBuilder{
+		formatVersion: DefaultViewFormatVersion,
+		updates:       make([]Update, 0),
+		schemaList:    make([]*iceberg.Schema, 0),
+		versionList:   make([]*Version, 0),
+		versionLog:    make([]VersionLogEntry, 0),
+		props:         make(iceberg.Properties),
+		versionsById:  make(map[int64]*Version),
+		schemasById:   make(map[int]*iceberg.Schema),
+	}, nil
+}
+
+func MetadataBuilderFromBase(metadata Metadata) (*MetadataBuilder, error) {
+	b := &MetadataBuilder{}
+	b.base = metadata
+
+	// Copy fields
+	b.formatVersion = metadata.FormatVersion()
+	b.uuid = metadata.ViewUUID()
+	b.loc = metadata.Location()
+	b.schemaList = slices.Clone(metadata.Schemas())
+	b.currentVersionID = metadata.CurrentVersionID()
+	b.versionList = cloneSlice(metadata.Versions())
+	b.versionLog = slices.Clone(metadata.VersionLog())
+	b.props = maps.Clone(metadata.Properties())
+
+	// Build lookup maps
+	b.versionsById = indexBy(b.versionList, func(vl *Version) int64 { return vl.VersionID })
+	b.schemasById = indexBy(b.schemaList, func(sm *iceberg.Schema) int { return sm.ID })
+
+	return b, nil
+}
+
+func (b *MetadataBuilder) HasChanges() bool { return len(b.updates) > 0 }
 
-func (b *MetadataBuilder) AssignUUID(_ string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetCurrentVersion(version *Version, schema *iceberg.Schema) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	newSchemaID, err := b.addSchema(schema)
+	if b.setErr(err) {
+		return b
+	}
+
+	newVersion := version.Clone()
+	newVersion.SchemaID = newSchemaID
+	newVersionID, err := b.addVersion(newVersion)
+	if b.setErr(err) {
+		return b
+	}
+
+	return b.SetCurrentVersionID(newVersionID)
 }
 
-func (b *MetadataBuilder) UpgradeFormatVersion(_ int) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) AddVersion(newVersion *Version) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	_, err := b.addVersion(newVersion)
+	if b.setErr(err) {
+		return b
+	}
+	return b
 }
 
-func (b *MetadataBuilder) AddSchema(_ *iceberg.Schema) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) addVersion(newVersion *Version) (int64, error) {
+	newVersionID := b.reuseOrCreateNewVersionID(newVersion)
+	version := newVersion.Clone()
+	if newVersionID != version.VersionID {
+		version.VersionID = newVersionID
+	}
+
+	// Check if this version was added in an update already
+	if _, ok := b.versionsById[newVersionID]; ok {
+		for _, upd := range b.updates {
+			if vvu, ok := upd.(*addViewVersionUpdate); ok && vvu.Version.VersionID == newVersionID {
+				b.lastAddedVersionID = &newVersionID
+				return newVersionID, nil
+			}
+		}
+	}
+
+	// If the SchemaID of the version is unset (nil), we attach the lastAddedSchemaID to it.
+	// If we have no lastAddedSchemaID, we fail
+	if version.SchemaID == LastAddedID {
+		if b.lastAddedSchemaID == nil {
+			return 0, errors.New("cannot set last added schema: no schema has been added")
+		}
+		version.SchemaID = *b.lastAddedSchemaID
+	}
+
+	if _, ok := b.schemasById[version.SchemaID]; !ok {
+		return 0, fmt.Errorf("cannot add version with unknown schema: %d", version.SchemaID)
+	}
+
+	if len(version.Representations) == 0 {
+		return 0, fmt.Errorf("cannot add version with no representations")
+	}
+	dialects := make(map[string]struct{})
+	for _, repr := range version.Representations {
+		normalizedDialect := strings.ToLower(repr.Dialect)
+		if _, ok := dialects[normalizedDialect]; ok {
+			return 0, fmt.Errorf("invalid view version: cannot add multiple queries for dialect %s", normalizedDialect)
+		}
+		dialects[normalizedDialect] = struct{}{}
+	}
+
+	b.versionList = append(b.versionList, version)
+	b.versionsById[version.VersionID] = version
+
+	if b.lastAddedSchemaID != nil && version.SchemaID == *b.lastAddedSchemaID {
+		updateVersion := version.Clone()
+		updateVersion.SchemaID = LastAddedID
+		b.updates = append(b.updates, NewAddViewVersionUpdate(updateVersion))
+	} else {
+		b.updates = append(b.updates, NewAddViewVersionUpdate(version))
+	}
+
+	b.lastAddedVersionID = &newVersionID
+	return newVersionID, nil
 }
 
-func (b *MetadataBuilder) SetLocation(_ string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) AddSchema(schema *iceberg.Schema) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	_, err := b.addSchema(schema)
+	if b.setErr(err) {
+		return b
+	}
+	return b
 }
 
-func (b *MetadataBuilder) SetProperties(_ map[string]string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) addSchema(schema *iceberg.Schema) (int, error) {
+	newSchemaID := b.reuseOrCreateNewSchemaID(schema)
+
+	if _, ok := b.schemasById[newSchemaID]; ok {
+		return newSchemaID, nil
+	}
+
+	newSchema := schema
+	// Build a fresh schema if we reset the ID
+	if schema.ID != newSchemaID {
+		newSchema = iceberg.NewSchemaWithIdentifiers(newSchemaID, schema.IdentifierFieldIDs, schema.Fields()...)
+	}
+
+	b.schemaList = append(b.schemaList, newSchema)
+	b.schemasById[newSchema.ID] = newSchema
+	b.updates = append(b.updates, NewAddSchemaUpdate(newSchema))
+	b.lastAddedSchemaID = &newSchemaID
+
+	return newSchemaID, nil
 }
 
-func (b *MetadataBuilder) RemoveProperties(_ []string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) RemoveProperties(keys []string) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if len(keys) == 0 {
+		return b
+	}
+
+	b.updates = append(b.updates, NewRemovePropertiesUpdate(keys))
+	for _, key := range keys {
+		delete(b.props, key)
+	}
+
+	return b
 }
 
-func (b *MetadataBuilder) AddVersion(_ *Version) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetCurrentVersionID(newVersionID int64) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if newVersionID == LastAddedID {
+		if b.lastAddedVersionID == nil {
+			b.err = errors.New("cannot set current version to last added, no version has been added")
+			return b
+		}
+		newVersionID = *b.lastAddedVersionID
+	}
+
+	if newVersionID == b.currentVersionID {
+		return b
+	}
+
+	_, ok := b.versionsById[newVersionID]
+	if !ok {
+		b.err = fmt.Errorf("cannot set current version to unknown version with id %d", newVersionID)
+		return b
+	}
+
+	if b.lastAddedVersionID != nil && *b.lastAddedVersionID == newVersionID {
+		b.updates = append(b.updates, NewSetCurrentVersionUpdate(LastAddedID))
+	} else {
+		b.updates = append(b.updates, NewSetCurrentVersionUpdate(newVersionID))
+	}
+	b.currentVersionID = newVersionID
+
+	// Set the current history entry.
+	// If the version was added in the current set of changes, use its timestamp,
+	// otherwise use system time.
+	updateTimestampMS := time.Now().UnixMilli()
+	for _, update := range b.updates {
+		if v, ok := update.(*addViewVersionUpdate); ok && v.Version.VersionID == newVersionID {
+			updateTimestampMS = v.Version.TimestampMS
+			break
+		}
+	}
+
+	b.versionHistoryEntry = &VersionLogEntry{
+		VersionID:   newVersionID,
+		TimestampMS: updateTimestampMS,
+	}
+
+	return b
 }
 
-func (b *MetadataBuilder) SetCurrentVersionID(_ int64) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetFormatVersion(formatVersion int) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if formatVersion < b.formatVersion {
+		b.err = fmt.Errorf("downgrading format version from %d to %d is not allowed",
+			b.formatVersion, formatVersion)
+		return b
+	}
+
+	if formatVersion > SupportedViewFormatVersion {
+		b.err = fmt.Errorf("unsupported format version %d", formatVersion)
+		return b
+	}
+
+	if formatVersion == b.formatVersion {
+		return b
+	}
+
+	b.updates = append(b.updates, NewUpgradeFormatVersionUpdate(formatVersion))
+	b.formatVersion = formatVersion
+
+	return b
+}
+
+func (b *MetadataBuilder) SetLoc(loc string) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if b.loc == loc {
+		return b
+	}
+
+	b.updates = append(b.updates, NewSetLocationUpdate(loc))
+	b.loc = loc
+
+	return b
+}
+
+func (b *MetadataBuilder) SetProperties(props iceberg.Properties) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if len(props) == 0 {
+		return b
+	}
+
+	b.updates = append(b.updates, NewSetPropertiesUpdate(props))
+	if b.props == nil {
+		b.props = props
+	} else {
+		maps.Copy(b.props, props)
+	}
+
+	return b
+}
+
+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
+	}
+
+	// Noop - same UUID does not generate a change.
+	if b.uuid == newUUID {
+		return b
+	}
+
+	if b.uuid != uuid.Nil {
+		b.err = errors.New("cannot reassign uuid")
+		return b
+	}
+
+	b.updates = append(b.updates, NewAssignUUIDUpdate(newUUID))
+	b.uuid = newUUID
+
+	return b
+}
+
+func (b *MetadataBuilder) Err() error {
+	return b.err
+}
+
+func (b *MetadataBuilder) buildMetadata(withUpdates bool) (*metadata, error) {
+	uuid_ := b.uuid
+	if uuid_ == uuid.Nil {
+		uuid_ = uuid.New()
+	}
+
+	md := &metadata{
+		FormatVersionValue:    b.formatVersion,
+		UUID:                  uuid_,
+		Loc:                   b.loc,
+		CurrentVersionIDValue: b.currentVersionID,
+		VersionList:           b.versionList,
+		VersionLogList:        b.versionLog,
+		SchemaList:            b.schemaList,
+		Props:                 b.props,
+	}
+	if withUpdates {
+		md.updates = b.updates
+	}
+	md.init()
+
+	return md, nil
+}
+
+// build builds the view metadata updates from the builder
+// if withUpdates is set to false, updates will not be included
+// in the returned Metadata, thus its Updates() will return an empty slice
+func (b *MetadataBuilder) build(withUpdates bool) (Metadata, error) {

Review Comment:
   I may have missed it, I think we're missing this validation:
   
   ```java
       private void checkIfDialectIsDropped(ViewVersion previous, ViewVersion current) {
         Set<String> baseDialects = sqlDialectsFor(previous);
         Set<String> updatedDialects = sqlDialectsFor(current);
   
         Preconditions.checkState(
             updatedDialects.containsAll(baseDialects),
             "Cannot replace view due to loss of view dialects (%s=false):\nPrevious dialects: %s\nNew dialects: %s",
             ViewProperties.REPLACE_DROP_DIALECT_ALLOWED,
             baseDialects,
             updatedDialects);
       }
   ```



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["twuebi (via GitHub)" <gi...@apache.org>]

twuebi commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2541988038


##########
view/metadata.go:
##########
@@ -18,135 +18,523 @@
 package view
 
 import (
-	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"maps"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
 	"time"
 
 	"github.com/apache/iceberg-go"
-	"github.com/apache/iceberg-go/internal"
-	"github.com/apache/iceberg-go/io"
+	iceinternal "github.com/apache/iceberg-go/internal"
+	"github.com/apache/iceberg-go/table"
+
 	"github.com/google/uuid"
 )
 
-// LoadMetadata reads and parses a view metadata file from the specified location.
+var (
+	ErrInvalidViewMetadata              = errors.New("invalid view metadata")
+	ErrInvalidViewMetadataFormatVersion = errors.New("invalid or missing format-version in view metadata")
+)
+
+const (
+	// LastAddedID is used in place of ID fields (e.g. schema, version) to indicate that
+	// the last added instance of that type should be used.
+	LastAddedID                = -1
+	SupportedViewFormatVersion = 1
+	DefaultViewFormatVersion   = SupportedViewFormatVersion
+)
+
+const (
+	initialSchemaID = 0
+
+	viewEngineProperty = "engine-name"
+	defaultViewEngine  = "iceberg-go"
+)
+
+// Metadata for an iceberg view as specified in the Iceberg spec
+// https://iceberg.apache.org/view-spec/
+type Metadata interface {
+	// FormatVersion indicates the version of this metadata, 1 for V1
+	FormatVersion() int
+	// ViewUUID returns a UUID that identifies the view, generated when the
+	// view is created. Implementations must throw an exception if a view's
+	// UUID does not match the expected UUID after refreshing metadata.
+	ViewUUID() uuid.UUID
+	// Location is the table's base location. This is used by writers to determine
+	// where to store data files, manifest files, and table metadata files.
+	Location() string
+	// Schemas returns the list of view schemas
+	Schemas() []*iceberg.Schema
+	// CurrentVersionID returns the ID of the current version of the view (version-id)
+	CurrentVersionID() int64
+	// CurrentVersion returns the current version of the view
+	CurrentVersion() *Version
+	// CurrentSchemaID returns the ID of the current schema
+	CurrentSchemaID() int
+	// CurrentSchema returns the current schema of the view
+	CurrentSchema() *iceberg.Schema
+	// SchemasByID returns a map of schema IDs to schemas
+	SchemasByID() map[int]*iceberg.Schema
+	// Versions returns the list of view versions
+	Versions() []*Version
+	// VersionLog returns a list of version log entries
+	// with the timestamp and version-id for every change to current-version-id
+	VersionLog() []VersionLogEntry
+	// Properties is a string to string map of view properties.
+	Properties() iceberg.Properties
+
+	// Updates returns the list of metadata updates used to build this metadata
+	Updates() Updates

Review Comment:
   I'd prefer if we'd return a result struct like:
   
   ```go
   type ViewMetadataBuildResult struct {
   	Metadata *metadata
   	Changes  []ViewUpdate
   }
   ```
   
   from builder.Build() instead, the updates are a property of the builder / result of the builder but not the metadata itself, reading metadata from disk or receiving it on the wire will have an empty list here, the result struct would separate the concerns and prevent this awkwardness.



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["twuebi (via GitHub)" <gi...@apache.org>]

twuebi commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2542743626


##########
view/metadata_builder.go:
##########
@@ -1,57 +1,468 @@
-// 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 view
 
 import (
+	"errors"
+	"fmt"
+	"maps"
+	"slices"
+	"strings"
+	"time"
+
 	"github.com/apache/iceberg-go"
+
+	"github.com/google/uuid"
 )
 
-// MetadataBuilder is a struct used for building and updating Iceberg view metadata.
-type MetadataBuilder struct{}
+type MetadataBuilder struct {
+	base    Metadata
+	updates Updates
+
+	// common fields
+	formatVersion    int
+	uuid             uuid.UUID
+	loc              string
+	schemaList       []*iceberg.Schema
+	versionList      []*Version
+	currentVersionID int64
+	versionLog       []VersionLogEntry
+	props            iceberg.Properties
+
+	// lookup maps
+	versionsById map[int64]*Version
+	schemasById  map[int]*iceberg.Schema
+
+	// update tracking
+	versionHistoryEntry *VersionLogEntry
+	lastAddedVersionID  *int64
+	lastAddedSchemaID   *int
+
+	// error tracking for build chaining
+	// if set, subsequent operations become a noop
+	err error
+}
+
+func NewMetadataBuilder() (*MetadataBuilder, error) {
+	return &MetadataBuilder{
+		formatVersion: DefaultViewFormatVersion,
+		updates:       make([]Update, 0),
+		schemaList:    make([]*iceberg.Schema, 0),
+		versionList:   make([]*Version, 0),
+		versionLog:    make([]VersionLogEntry, 0),
+		props:         make(iceberg.Properties),
+		versionsById:  make(map[int64]*Version),
+		schemasById:   make(map[int]*iceberg.Schema),
+	}, nil
+}
+
+func MetadataBuilderFromBase(metadata Metadata) (*MetadataBuilder, error) {
+	b := &MetadataBuilder{}
+	b.base = metadata
+
+	// Copy fields
+	b.formatVersion = metadata.FormatVersion()
+	b.uuid = metadata.ViewUUID()
+	b.loc = metadata.Location()
+	b.schemaList = slices.Clone(metadata.Schemas())
+	b.currentVersionID = metadata.CurrentVersionID()
+	b.versionList = cloneSlice(metadata.Versions())
+	b.versionLog = slices.Clone(metadata.VersionLog())
+	b.props = maps.Clone(metadata.Properties())
+
+	// Build lookup maps
+	b.versionsById = indexBy(b.versionList, func(vl *Version) int64 { return vl.VersionID })
+	b.schemasById = indexBy(b.schemaList, func(sm *iceberg.Schema) int { return sm.ID })
+
+	return b, nil
+}
+
+func (b *MetadataBuilder) HasChanges() bool { return len(b.updates) > 0 }
 
-func (b *MetadataBuilder) AssignUUID(_ string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetCurrentVersion(version *Version, schema *iceberg.Schema) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	newSchemaID, err := b.addSchema(schema)
+	if b.setErr(err) {
+		return b
+	}
+
+	newVersion := version.Clone()
+	newVersion.SchemaID = newSchemaID
+	newVersionID, err := b.addVersion(newVersion)
+	if b.setErr(err) {
+		return b
+	}
+
+	return b.SetCurrentVersionID(newVersionID)
 }
 
-func (b *MetadataBuilder) UpgradeFormatVersion(_ int) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) AddVersion(newVersion *Version) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	_, err := b.addVersion(newVersion)
+	if b.setErr(err) {
+		return b
+	}
+	return b
 }
 
-func (b *MetadataBuilder) AddSchema(_ *iceberg.Schema) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) addVersion(newVersion *Version) (int64, error) {
+	newVersionID := b.reuseOrCreateNewVersionID(newVersion)
+	version := newVersion.Clone()
+	if newVersionID != version.VersionID {
+		version.VersionID = newVersionID
+	}
+
+	// Check if this version was added in an update already
+	if _, ok := b.versionsById[newVersionID]; ok {
+		for _, upd := range b.updates {
+			if vvu, ok := upd.(*addViewVersionUpdate); ok && vvu.Version.VersionID == newVersionID {
+				b.lastAddedVersionID = &newVersionID
+				return newVersionID, nil
+			}
+		}
+	}
+
+	// If the SchemaID of the version is unset (nil), we attach the lastAddedSchemaID to it.
+	// If we have no lastAddedSchemaID, we fail
+	if version.SchemaID == LastAddedID {
+		if b.lastAddedSchemaID == nil {
+			return 0, errors.New("cannot set last added schema: no schema has been added")
+		}
+		version.SchemaID = *b.lastAddedSchemaID
+	}
+
+	if _, ok := b.schemasById[version.SchemaID]; !ok {
+		return 0, fmt.Errorf("cannot add version with unknown schema: %d", version.SchemaID)
+	}
+
+	if len(version.Representations) == 0 {
+		return 0, fmt.Errorf("cannot add version with no representations")
+	}
+	dialects := make(map[string]struct{})
+	for _, repr := range version.Representations {
+		normalizedDialect := strings.ToLower(repr.Dialect)
+		if _, ok := dialects[normalizedDialect]; ok {
+			return 0, fmt.Errorf("invalid view version: cannot add multiple queries for dialect %s", normalizedDialect)
+		}
+		dialects[normalizedDialect] = struct{}{}
+	}
+
+	b.versionList = append(b.versionList, version)
+	b.versionsById[version.VersionID] = version
+
+	if b.lastAddedSchemaID != nil && version.SchemaID == *b.lastAddedSchemaID {
+		updateVersion := version.Clone()
+		updateVersion.SchemaID = LastAddedID
+		b.updates = append(b.updates, NewAddViewVersionUpdate(updateVersion))
+	} else {
+		b.updates = append(b.updates, NewAddViewVersionUpdate(version))
+	}
+
+	b.lastAddedVersionID = &newVersionID
+	return newVersionID, nil
 }
 
-func (b *MetadataBuilder) SetLocation(_ string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) AddSchema(schema *iceberg.Schema) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	_, err := b.addSchema(schema)
+	if b.setErr(err) {
+		return b
+	}
+	return b
 }
 
-func (b *MetadataBuilder) SetProperties(_ map[string]string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) addSchema(schema *iceberg.Schema) (int, error) {
+	newSchemaID := b.reuseOrCreateNewSchemaID(schema)
+
+	if _, ok := b.schemasById[newSchemaID]; ok {
+		return newSchemaID, nil
+	}
+
+	newSchema := schema
+	// Build a fresh schema if we reset the ID
+	if schema.ID != newSchemaID {
+		newSchema = iceberg.NewSchemaWithIdentifiers(newSchemaID, schema.IdentifierFieldIDs, schema.Fields()...)
+	}
+
+	b.schemaList = append(b.schemaList, newSchema)
+	b.schemasById[newSchema.ID] = newSchema
+	b.updates = append(b.updates, NewAddSchemaUpdate(newSchema))
+	b.lastAddedSchemaID = &newSchemaID
+
+	return newSchemaID, nil
 }
 
-func (b *MetadataBuilder) RemoveProperties(_ []string) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) RemoveProperties(keys []string) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if len(keys) == 0 {
+		return b
+	}
+
+	b.updates = append(b.updates, NewRemovePropertiesUpdate(keys))
+	for _, key := range keys {
+		delete(b.props, key)
+	}
+
+	return b
 }
 
-func (b *MetadataBuilder) AddVersion(_ *Version) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetCurrentVersionID(newVersionID int64) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if newVersionID == LastAddedID {
+		if b.lastAddedVersionID == nil {
+			b.err = errors.New("cannot set current version to last added, no version has been added")
+			return b
+		}
+		newVersionID = *b.lastAddedVersionID
+	}
+
+	if newVersionID == b.currentVersionID {
+		return b
+	}
+
+	_, ok := b.versionsById[newVersionID]
+	if !ok {
+		b.err = fmt.Errorf("cannot set current version to unknown version with id %d", newVersionID)
+		return b
+	}
+
+	if b.lastAddedVersionID != nil && *b.lastAddedVersionID == newVersionID {
+		b.updates = append(b.updates, NewSetCurrentVersionUpdate(LastAddedID))
+	} else {
+		b.updates = append(b.updates, NewSetCurrentVersionUpdate(newVersionID))
+	}
+	b.currentVersionID = newVersionID
+
+	// Set the current history entry.
+	// If the version was added in the current set of changes, use its timestamp,
+	// otherwise use system time.
+	updateTimestampMS := time.Now().UnixMilli()
+	for _, update := range b.updates {
+		if v, ok := update.(*addViewVersionUpdate); ok && v.Version.VersionID == newVersionID {
+			updateTimestampMS = v.Version.TimestampMS
+			break
+		}
+	}
+
+	b.versionHistoryEntry = &VersionLogEntry{
+		VersionID:   newVersionID,
+		TimestampMS: updateTimestampMS,
+	}
+
+	return b
 }
 
-func (b *MetadataBuilder) SetCurrentVersionID(_ int64) error {
-	return iceberg.ErrNotImplemented
+func (b *MetadataBuilder) SetFormatVersion(formatVersion int) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if formatVersion < b.formatVersion {
+		b.err = fmt.Errorf("downgrading format version from %d to %d is not allowed",
+			b.formatVersion, formatVersion)
+		return b
+	}
+
+	if formatVersion > SupportedViewFormatVersion {
+		b.err = fmt.Errorf("unsupported format version %d", formatVersion)
+		return b
+	}
+
+	if formatVersion == b.formatVersion {
+		return b
+	}
+
+	b.updates = append(b.updates, NewUpgradeFormatVersionUpdate(formatVersion))
+	b.formatVersion = formatVersion
+
+	return b
+}
+
+func (b *MetadataBuilder) SetLoc(loc string) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if b.loc == loc {
+		return b
+	}
+
+	b.updates = append(b.updates, NewSetLocationUpdate(loc))
+	b.loc = loc
+
+	return b
+}
+
+func (b *MetadataBuilder) SetProperties(props iceberg.Properties) *MetadataBuilder {
+	if b.err != nil {
+		return b
+	}
+
+	if len(props) == 0 {
+		return b
+	}
+
+	b.updates = append(b.updates, NewSetPropertiesUpdate(props))
+	if b.props == nil {
+		b.props = props
+	} else {
+		maps.Copy(b.props, props)
+	}
+
+	return b
+}
+
+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
+	}
+
+	// Noop - same UUID does not generate a change.
+	if b.uuid == newUUID {
+		return b
+	}
+
+	if b.uuid != uuid.Nil {
+		b.err = errors.New("cannot reassign uuid")
+		return b
+	}
+
+	b.updates = append(b.updates, NewAssignUUIDUpdate(newUUID))
+	b.uuid = newUUID
+
+	return b
+}
+
+func (b *MetadataBuilder) Err() error {
+	return b.err
+}
+
+func (b *MetadataBuilder) buildMetadata(withUpdates bool) (*metadata, error) {
+	uuid_ := b.uuid
+	if uuid_ == uuid.Nil {
+		uuid_ = uuid.New()
+	}
+
+	md := &metadata{
+		FormatVersionValue:    b.formatVersion,
+		UUID:                  uuid_,
+		Loc:                   b.loc,
+		CurrentVersionIDValue: b.currentVersionID,
+		VersionList:           b.versionList,
+		VersionLogList:        b.versionLog,
+		SchemaList:            b.schemaList,
+		Props:                 b.props,
+	}
+	if withUpdates {
+		md.updates = b.updates
+	}
+	md.init()
+
+	return md, nil
+}
+
+// build builds the view metadata updates from the builder
+// if withUpdates is set to false, updates will not be included
+// in the returned Metadata, thus its Updates() will return an empty slice
+func (b *MetadataBuilder) build(withUpdates bool) (Metadata, error) {

Review Comment:
   We're also missing this action I think:
   
   ```java
   public ViewMetadata build() {
         ...
         // expire old versions, but keep at least the versions added in this builder and the current
         // version
         int numVersions =
             ImmutableSet.builder()
                 .addAll(
                     changes(MetadataUpdate.AddViewVersion.class)
                         .map(v -> v.viewVersion().versionId())
                         .collect(Collectors.toSet()))
                 .add(currentVersionId)
                 .build()
                 .size();
         int numVersionsToKeep = Math.max(numVersions, historySize);
         List<ViewVersion> retainedVersions;
         List<ViewHistoryEntry> retainedHistory;
         if (versions.size() > numVersionsToKeep) {
           retainedVersions =
               expireVersions(versionsById, numVersionsToKeep, versionsById.get(currentVersionId));
           Set<Integer> retainedVersionIds =
               retainedVersions.stream().map(ViewVersion::versionId).collect(Collectors.toSet());
           retainedHistory = updateHistory(history, retainedVersionIds);
         } else {
           retainedVersions = versions;
           retainedHistory = history;
         }
       ...
       }
   
       static List<ViewVersion> expireVersions(
           Map<Integer, ViewVersion> versionsById, int numVersionsToKeep, ViewVersion currentVersion) {
         // version ids are assigned sequentially. keep the latest versions by ID.
         List<Integer> ids = Lists.newArrayList(versionsById.keySet());
         ids.sort(Comparator.reverseOrder());
   
         List<ViewVersion> retainedVersions = Lists.newArrayList();
         // always retain the current version
         retainedVersions.add(currentVersion);
   
         for (int idToKeep : ids.subList(0, numVersionsToKeep)) {
           if (retainedVersions.size() == numVersionsToKeep) {
             break;
           }
   
           ViewVersion version = versionsById.get(idToKeep);
           if (currentVersion.versionId() != version.versionId()) {
             retainedVersions.add(version);
           }
         }
   
         return retainedVersions;
       }
   
   
       static List<ViewHistoryEntry> updateHistory(List<ViewHistoryEntry> history, Set<Integer> ids) {
         List<ViewHistoryEntry> retainedHistory = Lists.newArrayList();
         for (ViewHistoryEntry entry : history) {
           if (ids.contains(entry.versionId())) {
             retainedHistory.add(entry);
           } else {
             // clear history past any unknown version
             retainedHistory.clear();
           }
         }
   
         return retainedHistory;
       }
   ```



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2542958488


##########
view/metadata.go:
##########
@@ -18,135 +18,523 @@
 package view
 
 import (
-	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"maps"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
 	"time"
 
 	"github.com/apache/iceberg-go"
-	"github.com/apache/iceberg-go/internal"
-	"github.com/apache/iceberg-go/io"
+	iceinternal "github.com/apache/iceberg-go/internal"
+	"github.com/apache/iceberg-go/table"
+
 	"github.com/google/uuid"
 )
 
-// LoadMetadata reads and parses a view metadata file from the specified location.
+var (
+	ErrInvalidViewMetadata              = errors.New("invalid view metadata")
+	ErrInvalidViewMetadataFormatVersion = errors.New("invalid or missing format-version in view metadata")
+)
+
+const (
+	// LastAddedID is used in place of ID fields (e.g. schema, version) to indicate that
+	// the last added instance of that type should be used.
+	LastAddedID                = -1
+	SupportedViewFormatVersion = 1
+	DefaultViewFormatVersion   = SupportedViewFormatVersion
+)
+
+const (
+	initialSchemaID = 0
+
+	viewEngineProperty = "engine-name"
+	defaultViewEngine  = "iceberg-go"
+)
+
+// Metadata for an iceberg view as specified in the Iceberg spec
+// https://iceberg.apache.org/view-spec/
+type Metadata interface {
+	// FormatVersion indicates the version of this metadata, 1 for V1
+	FormatVersion() int
+	// ViewUUID returns a UUID that identifies the view, generated when the
+	// view is created. Implementations must throw an exception if a view's
+	// UUID does not match the expected UUID after refreshing metadata.
+	ViewUUID() uuid.UUID
+	// Location is the table's base location. This is used by writers to determine
+	// where to store data files, manifest files, and table metadata files.
+	Location() string
+	// Schemas returns the list of view schemas
+	Schemas() []*iceberg.Schema
+	// CurrentVersionID returns the ID of the current version of the view (version-id)
+	CurrentVersionID() int64
+	// CurrentVersion returns the current version of the view
+	CurrentVersion() *Version
+	// CurrentSchemaID returns the ID of the current schema
+	CurrentSchemaID() int
+	// CurrentSchema returns the current schema of the view
+	CurrentSchema() *iceberg.Schema
+	// SchemasByID returns a map of schema IDs to schemas
+	SchemasByID() map[int]*iceberg.Schema
+	// Versions returns the list of view versions
+	Versions() []*Version
+	// VersionLog returns a list of version log entries
+	// with the timestamp and version-id for every change to current-version-id
+	VersionLog() []VersionLogEntry
+	// Properties is a string to string map of view properties.
+	Properties() iceberg.Properties
+
+	// Updates returns the list of metadata updates used to build this metadata
+	Updates() Updates

Review Comment:
   I guess to play devil's advocate, I can see why the java implementation might choose to couple changes / updates and metadata. The only reason I added a method for `BuildWithoutUpdates` was to discard the updates reference to save memory if the caller doesn't need it. But it's not as though the updates is a big chunk of memory.
   
   I could also just get rid of the ugly switching on `withUpdates` to make this cleaner if you're ok with that as well



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2543029793


##########
view/metadata.go:
##########
@@ -18,135 +18,523 @@
 package view
 
 import (
-	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"maps"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
 	"time"
 
 	"github.com/apache/iceberg-go"
-	"github.com/apache/iceberg-go/internal"
-	"github.com/apache/iceberg-go/io"
+	iceinternal "github.com/apache/iceberg-go/internal"
+	"github.com/apache/iceberg-go/table"
+
 	"github.com/google/uuid"
 )
 
-// LoadMetadata reads and parses a view metadata file from the specified location.
+var (
+	ErrInvalidViewMetadata              = errors.New("invalid view metadata")
+	ErrInvalidViewMetadataFormatVersion = errors.New("invalid or missing format-version in view metadata")
+)
+
+const (
+	// LastAddedID is used in place of ID fields (e.g. schema, version) to indicate that
+	// the last added instance of that type should be used.
+	LastAddedID                = -1
+	SupportedViewFormatVersion = 1
+	DefaultViewFormatVersion   = SupportedViewFormatVersion
+)
+
+const (
+	initialSchemaID = 0
+
+	viewEngineProperty = "engine-name"
+	defaultViewEngine  = "iceberg-go"
+)
+
+// Metadata for an iceberg view as specified in the Iceberg spec
+// https://iceberg.apache.org/view-spec/
+type Metadata interface {
+	// FormatVersion indicates the version of this metadata, 1 for V1
+	FormatVersion() int
+	// ViewUUID returns a UUID that identifies the view, generated when the
+	// view is created. Implementations must throw an exception if a view's
+	// UUID does not match the expected UUID after refreshing metadata.
+	ViewUUID() uuid.UUID
+	// Location is the table's base location. This is used by writers to determine
+	// where to store data files, manifest files, and table metadata files.
+	Location() string
+	// Schemas returns the list of view schemas
+	Schemas() []*iceberg.Schema
+	// CurrentVersionID returns the ID of the current version of the view (version-id)
+	CurrentVersionID() int64
+	// CurrentVersion returns the current version of the view
+	CurrentVersion() *Version
+	// CurrentSchemaID returns the ID of the current schema
+	CurrentSchemaID() int
+	// CurrentSchema returns the current schema of the view
+	CurrentSchema() *iceberg.Schema
+	// SchemasByID returns a map of schema IDs to schemas
+	SchemasByID() map[int]*iceberg.Schema
+	// Versions returns the list of view versions
+	Versions() []*Version
+	// VersionLog returns a list of version log entries
+	// with the timestamp and version-id for every change to current-version-id
+	VersionLog() []VersionLogEntry
+	// Properties is a string to string map of view properties.
+	Properties() iceberg.Properties
+
+	// Updates returns the list of metadata updates used to build this metadata
+	Updates() Updates
+
+	Equals(Metadata) bool
+}
+
+// VersionSummary is string to string map of summary metadata about a view's version
+type VersionSummary map[string]string
+
+// Representation is a struct containing information about a view's representation
+// https://iceberg.apache.org/view-spec/#sql-representation
+type Representation struct {
+	// Must be sql
+	Type string `json:"type"`
+	// A SQL SELECT statement
+	Sql string `json:"sql"`
+	// The dialect of the sql SELECT statement (e.g., "trino" or "spark")
+	Dialect string `json:"dialect"`
+}
+type Representations []Representation
+
+func NewRepresentation(sql string, dialect string) Representation {
+	return Representation{
+		Type:    "sql",
+		Sql:     sql,
+		Dialect: dialect,
+	}
+}
+
+type Version struct {
+	// ID for the version
+	VersionID int64 `json:"version-id"`
+	// ID of the schema for the view version
+	SchemaID int `json:"schema-id"`
+	// Timestamp when the version was created (ms from epoch)
+	TimestampMS int64 `json:"timestamp-ms"`
+	// A string to string map of summary metadata about the version
+	Summary VersionSummary `json:"summary"`
+	// A list of representations for the view definition
+	Representations []Representation `json:"representations"`
+	// The default view namespace (as a list of strings)
+	DefaultNamespace table.Identifier `json:"default-namespace"`
+	// An (optional) default catalog name for querying the view
+	DefaultCatalog string `json:"default-catalog,omitempty"`
+}
+
+type VersionOpt func(*Version)
+
+func WithVersionSummary(summary VersionSummary) VersionOpt {
+	return func(v *Version) {
+		if v.Summary == nil {
+			v.Summary = summary
+		} else {
+			maps.Copy(v.Summary, summary)
+		}
+	}
+}
+
+func WithDefaultViewCatalog(catalogName string) func(*Version) {
+	return func(version *Version) {
+		version.DefaultCatalog = catalogName
+	}
+}
+
+func WithTimestampMS(timestampMS int64) VersionOpt {
+	return func(version *Version) {
+		version.TimestampMS = timestampMS
+	}
+}
+
+// NewVersion creates a Version instance from the provided parameters.
+// If building updates using the MetadataBuilder, and one desires to use the last
+// added schema ID, one should use the LastAddedID constant as the provided schemaID
 //
-// Returns the loaded Metadata or an error if the file cannot be read or parsed.
-func LoadMetadata(ctx context.Context,
-	props iceberg.Properties,
-	metadataLocation string,
-	name string,
-	namespace string,
-) (_ Metadata, err error) {
-	fs, err := io.LoadFS(ctx, props, metadataLocation)
-	if err != nil {
-		return nil, fmt.Errorf("error loading view metadata: %w", err)
+// Note that NewVersion automatically seeds TimestampMS to time.Now().UnixMilli(),
+// and one should use the option WithTimestampMS to override this behavior.
+func NewVersion(id int64, schemaID int, representations []Representation, defaultNS table.Identifier, opts ...VersionOpt) (*Version, error) {
+	if id < 1 {
+		return nil, errors.New("id should be greater than 0")
 	}
 
-	inputFile, err := fs.Open(metadataLocation)
-	if err != nil {
-		return nil, fmt.Errorf("%s.%s: error encountered loading view metadata: %w", namespace, name, err)
+	if len(representations) == 0 {
+		return nil, errors.New("invalid view version: must have at least one representation")
 	}
-	defer internal.CheckedClose(inputFile, &err)
 
-	var m metadata
-	if err := json.NewDecoder(inputFile).Decode(&m); err != nil {
-		return nil, fmt.Errorf("error encountered decoding view metadata: %w", err)
+	version := &Version{
+		VersionID:        id,
+		SchemaID:         schemaID,
+		Representations:  representations,
+		Summary:          VersionSummary{viewEngineProperty: defaultViewEngine},
+		TimestampMS:      time.Now().UnixMilli(),
+		DefaultNamespace: defaultNS,
 	}
 
-	return &m, nil
+	for _, opt := range opts {
+		opt(version)
+	}
+
+	return version, nil
 }
 
-// NewMetadata returns a view metadata for a given version.
-func NewMetadata(schema *iceberg.Schema,
-	version Version,
-	loc string,
-	props iceberg.Properties,
-) (Metadata, error) {
-	timestampMs := time.Now().UnixMilli()
+// NewVersionFromSQL creates a new Version with a single representation
+// using the provided SQL and dialect "default".
+func NewVersionFromSQL(id int64, schemaID int, sql string, defaultNS table.Identifier, opts ...VersionOpt) (*Version, error) {
+	return NewVersion(id, schemaID, Representations{NewRepresentation(sql, "default")}, defaultNS, opts...)
+}
 
-	if version.SchemaID != schema.ID {
-		return nil, fmt.Errorf("%w: version.SchemaID does not match schema.ID", iceberg.ErrInvalidArgument)
+// Equals checks whether the other Version would behave the same
+// while ignoring the view version id, and the creation timestamp
+func (v *Version) Equals(other *Version) bool {
+	return v.SchemaID == other.SchemaID &&
+		v.DefaultCatalog == other.DefaultCatalog &&
+		slices.Equal(v.DefaultNamespace, other.DefaultNamespace) &&
+		maps.Equal(v.Summary, other.Summary) &&
+		slices.Equal(v.Representations, other.Representations)
+}
+
+func (v *Version) Clone() *Version {
+	if v == nil {
+		return nil
 	}
 
-	return &metadata{
-		UUID:             uuid.NewString(),
-		FmtVersion:       1,
-		Loc:              loc,
-		SchemaList:       []*iceberg.Schema{schema},
-		CurrentVersionId: version.VersionID,
-		VersionList:      []Version{version},
-		VersionLogList: []VersionLogEntry{
-			{
-				TimestampMs: timestampMs,
-				VersionID:   version.VersionID,
-			},
-		},
-		Props: props,
-	}, nil
+	cloned := *v
+	cloned.Summary = maps.Clone(v.Summary)
+	cloned.Representations = slices.Clone(v.Representations)
+	cloned.DefaultNamespace = slices.Clone(v.DefaultNamespace)
+	return &cloned
 }
 
-// CreateMetadata creates a new view metadata file and writes it to storage.
-//
-// Returns the full path to the created metadata file, or an error if creation fails.
-//
-// Note: This function only supports creating new views with format version 1.
-// It does not support updating existing view metadata.
-func CreateMetadata(
-	ctx context.Context,
-	catalogName string,
-	nsIdent []string,
-	schema *iceberg.Schema,
-	viewSQL string,
-	loc string,
-	props iceberg.Properties,
-) (metadataLocation string, err error) {
-	versionId := int64(1)
-	timestampMs := time.Now().UnixMilli()
-
-	viewVersion := Version{
-		VersionID:   versionId,
-		TimestampMs: timestampMs,
-		SchemaID:    schema.ID,
-		Summary:     map[string]string{"sql": viewSQL},
-		Representations: []SQLRepresentation{
-			{Type: "sql", SQL: viewSQL, Dialect: "default"},
-		},
-		DefaultCatalog:   &catalogName,
-		DefaultNamespace: nsIdent,
-	}
-
-	metadataLocation = loc + "/metadata/view-" + uuid.New().String() + ".metadata.json"
-	viewMetadata, err := NewMetadata(schema, viewVersion, loc, props)
+type VersionLogEntry struct {
+	// Timestamp when the view's current-version-id was updated (ms from epoch)
+	TimestampMS int64 `json:"timestamp-ms"`
+	// ID that current-version-id was set to
+	VersionID int64 `json:"version-id"`
+}
+
+// ParseMetadata parses json metadata 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 "", fmt.Errorf("failed to create view metadata: %w", err)
+		return nil, err
 	}
 
-	viewMetadataBytes, err := json.Marshal(viewMetadata)
-	if err != nil {
-		return "", fmt.Errorf("failed to marshal view metadata: %w", 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) {
+	ver := struct {
+		FormatVersion int `json:"format-version"`
+	}{}
+	if err := json.Unmarshal(b, &ver); err != nil {
+		return nil, err
 	}
 
-	fs, err := io.LoadFS(ctx, props, metadataLocation)
-	if err != nil {
-		return "", fmt.Errorf("failed to load filesystem for view metadata: %w", err)
+	var ret Metadata
+	ret = &metadata{}
+	return ret, json.Unmarshal(b, ret)
+}
+
+// indexBy indexes a slice into a map, using a provided extractKey function
+// The extractKey function will be called on each item in the slice and assign the
+// item as a value for that key in the resultant map.
+func indexBy[T any, K comparable](s []T, extractKey func(T) K) map[K]T {
+	index := make(map[K]T)
+	for _, v := range s {
+		index[extractKey(v)] = v
+	}
+	return index
+}
+
+// Cloner is an interface which implements a Clone method for deep copying itself
+type cloner[T any] interface {
+	// Clone returns a deep copy of the underlying object
+	Clone() T
+}
+
+// cloneSlice returns a deep-clone of a Slice of elements implementing Cloner
+func cloneSlice[T cloner[T]](val []T) []T {
+	cloned := make([]T, len(val))
+	for i, elem := range val {
+		cloned[i] = elem.Clone()
+	}
+	return cloned
+}
+
+// https://iceberg.apache.org/view-spec/
+type metadata struct {
+	FormatVersionValue    int                `json:"format-version"`
+	UUID                  uuid.UUID          `json:"view-uuid"`
+	Loc                   string             `json:"location"`
+	CurrentVersionIDValue int64              `json:"current-version-id"`
+	VersionList           []*Version         `json:"versions"`
+	SchemaList            []*iceberg.Schema  `json:"schemas"`
+	VersionLogList        []VersionLogEntry  `json:"version-log"`
+	Props                 iceberg.Properties `json:"properties,omitempty"`
+
+	// updates from builder if applicable
+	updates Updates
+
+	// cached lookup helpers, must be initialized in init()
+	lazyVersionsByID func() map[int64]*Version
+	lazySchemasByID  func() map[int]*iceberg.Schema
+}
+
+func (m *metadata) Equals(other Metadata) bool {
+	if other == nil {
+		return false
+	}
+
+	if m == other {
+		return true
+	}
+
+	return m.UUID == other.ViewUUID() &&
+		m.Loc == other.Location() &&
+		m.FormatVersionValue == other.FormatVersion() &&
+		iceinternal.SliceEqualHelper(m.Schemas(), other.Schemas()) &&
+		iceinternal.SliceEqualHelper(m.Versions(), other.Versions()) &&
+		m.CurrentVersionIDValue == other.CurrentVersionID() &&
+		slices.Equal(m.VersionLogList, other.VersionLog())
+}
+
+func (m *metadata) Updates() Updates { return m.updates }
+
+func (m *metadata) FormatVersion() int         { return m.FormatVersionValue }
+func (m *metadata) ViewUUID() uuid.UUID        { return m.UUID }
+func (m *metadata) Location() string           { return m.Loc }
+func (m *metadata) Versions() []*Version       { return m.VersionList }
+func (m *metadata) Schemas() []*iceberg.Schema { return m.SchemaList }
+func (m *metadata) SchemasByID() map[int]*iceberg.Schema {
+	return maps.Clone(m.lazySchemasByID())
+}
+
+func (m *metadata) CurrentVersionID() int64 {
+	return m.CurrentVersionIDValue
+}
+
+func (m *metadata) CurrentVersion() *Version {
+	version, ok := m.lazyVersionsByID()[m.CurrentVersionIDValue]
+	if !ok {
+		panic("current version not found")
 	}

Review Comment:
   To address your question around user of the library provoking, the answer should be no they should not be able to provoke it. `metadata` is private and it's initialization should be abstracted by the `view` package. Thus, a user should not be able to initialize an invalid `metadata` instance unless there is a bug in `iceberg-go`



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["jnagel12 (via GitHub)" <gi...@apache.org>]

jnagel12 commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2539763461


##########
view/view.go:
##########
@@ -18,234 +18,156 @@
 package view
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
-	"iter"
 	"slices"
-	"strings"
 
 	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/internal"
+	"github.com/apache/iceberg-go/io"
+	"github.com/apache/iceberg-go/table"
+	"github.com/google/uuid"
 )
 
-var (
-	ErrInvalidViewMetadata              = errors.New("invalid view metadata")
-	ErrInvalidViewMetadataFormatVersion = errors.New("invalid or missing format-version in view metadata")
-)
-
-// Metadata defines the format for view metadata,
-// similar to how Iceberg supports a common table format for tables
-type Metadata interface {

Review Comment:
   moved to `metadata.go`



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(view): Implement view metadata builder and UpdateView method in REST catalog [iceberg-go]

["twuebi (via GitHub)" <gi...@apache.org>]

twuebi commented on code in PR #629:
URL: https://github.com/apache/iceberg-go/pull/629#discussion_r2547132115


##########
view/requirements.go:
##########


Review Comment:
   Let's add some tests, maybe something like:
   
   ```go
   // 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 view_test
   
   import (
   	"encoding/json"
   	"testing"
   
   	"github.com/apache/iceberg-go/view"
   	"github.com/google/uuid"
   	"github.com/stretchr/testify/assert"
   )
   
   func TestParseRequirementList(t *testing.T) {
   	t.Run("should parse a list of requirements", func(t *testing.T) {
   		jsonData := []byte(`[
   			{"type": "assert-view-uuid", "uuid": "550e8400-e29b-41d4-a716-446655440000"}
   		]`)
   
   		expected := view.Requirements{
   			view.AssertViewUUID(uuid.MustParse("550e8400-e29b-41d4-a716-446655440000")),
   		}
   
   		var actual view.Requirements
   		err := json.Unmarshal(jsonData, &actual)
   
   		assert.NoError(t, err)
   		assert.Equal(t, expected, actual)
   	})
   
   	t.Run("should handle an empty list", func(t *testing.T) {
   		jsonData := []byte(`[]`)
   		var actual view.Requirements
   		err := json.Unmarshal(jsonData, &actual)
   		assert.NoError(t, err)
   		assert.Empty(t, actual)
   	})
   
   	t.Run("should return an error for an unknown requirement type in the list", func(t *testing.T) {
   		jsonData := []byte(`[
   			{"type": "assert-view-uuid", "uuid": "550e8400-e29b-41d4-a716-446655440000"},
   			{"type": "assert-foo-bar"}
   		]`)
   
   		var actual view.Requirements
   		err := json.Unmarshal(jsonData, &actual)
   		assert.Error(t, err)
   		assert.Contains(t, err.Error(), "unknown requirement type: assert-foo-bar")
   	})
   
   	t.Run("should return an error for invalid json", func(t *testing.T) {
   		jsonData := []byte(`[{"type": "assert-view-uuid", "uuid": "550e8400-e29b-41d4-a716-446655440000"},]`) // trailing comma
   		var actual view.Requirements
   		err := json.Unmarshal(jsonData, &actual)
   		assert.Error(t, err)
   	})
   }
   
   ```



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org