laskoviymishka commented on code in PR #2290: URL: https://github.com/apache/iceberg-go/pull/2290#discussion_r4368745368
########## catalog/hadoop/tables.go: ########## @@ -0,0 +1,465 @@ +// 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 hadoop + +import ( + "context" + "errors" + "fmt" + "io/fs" + "iter" + "log" + "strings" + "time" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/catalog" + "github.com/apache/iceberg-go/catalog/internal" + "github.com/apache/iceberg-go/io" + "github.com/apache/iceberg-go/table" + "github.com/apache/iceberg-go/table/maintenance" + "github.com/google/uuid" +) + +func (c *Catalog) CreateTable(ctx context.Context, ident table.Identifier, sc *iceberg.Schema, opts ...catalog.CreateTableOpt) (*table.Table, error) { + var cfg catalog.CreateTableCfg + for _, opt := range opts { + opt(&cfg) + } + + if err := validateTableIdentifier(ident); err != nil { + return nil, err + } + + // Resolve the reporter before any catalog mutation: a bad + // metrics-reporter-impl must fail here, not after the metadata file and + // version hint are written, which would report a failure for a table that + // was actually created (and make the retry hit ErrTableAlreadyExists). + reporter, err := c.reporter.Get(c.props) + if err != nil { + return nil, fmt.Errorf("failed to initialize metrics reporter: %w", err) + } + + ns := catalog.NamespaceFromIdent(ident) + nsPath := c.namespaceToPath(ns) + + info, err := c.filesystem.Stat(nsPath) + if errors.Is(err, fs.ErrNotExist) || (err == nil && !info.IsDir()) { + return nil, fmt.Errorf("%w: %s", catalog.ErrNoSuchNamespace, strings.Join(ns, ".")) + } + + if err != nil { + return nil, fmt.Errorf("hadoop catalog: failed to stat namespace directory: %w", err) + } + + loc := c.defaultTableLocation(ident) + if cfg.Location != "" && cfg.Location != loc { + return nil, errors.New("hadoop catalog: custom table locations are not supported") + } + + exists, err := isTableDir(c.filesystem, c.isLocal, loc) + if err != nil { + return nil, fmt.Errorf("hadoop catalog: failed to inspect table directory: %w", err) + } + if exists { + return nil, fmt.Errorf("%w: %s", catalog.ErrTableAlreadyExists, strings.Join(ident, ".")) + } + + metadata, err := table.NewMetadata(sc, cfg.PartitionSpec, cfg.SortOrder, loc, cfg.Properties) + if err != nil { + return nil, fmt.Errorf("hadoop catalog: failed to create table metadata: %w", err) + } + + metaDir := c.metadataDir(ident) + if err := c.filesystem.MkdirAll(metaDir); err != nil { + return nil, fmt.Errorf("hadoop catalog: failed to create metadata directory: %w", err) + } + + version := 1 + compression := table.MetadataCompressionDefault + if cfg.Properties != nil { + if v, ok := cfg.Properties[table.MetadataCompressionKey]; ok { + compression = v + } + } + + metaPath, err := c.metadataFilePathForCompression(ident, version, compression) + if err != nil { + return nil, fmt.Errorf("hadoop catalog: failed to determine metadata file path: %w", err) + } + + tempPath := joinPath(c.isLocal, metaDir, uuid.New().String()+".metadata.json") + + if err := internal.WriteTableMetadata(metadata, c.filesystem, tempPath, compression); err != nil { + _ = c.filesystem.Remove(tempPath) + + return nil, fmt.Errorf("hadoop catalog: failed to write table metadata: %w", err) + } + + if err := c.commitMetadataFile(ident, version, tempPath, metaPath, catalog.ErrTableAlreadyExists); err != nil { + return nil, err + } + + c.writeVersionHint(ident, version) + + tbl := table.New( + ident, + metadata, + metaPath, + io.LoadFSFunc(c.props, metaPath), + c, + table.WithMetricsReporter(reporter), + table.WithScanPlanningIOProperties(c.props), + ) + + return tbl, nil +} + +func (c *Catalog) loadTable(ctx context.Context, ident table.Identifier) (*table.Table, int, error) { + if err := validateTableIdentifier(ident); err != nil { + return nil, 0, err + } + + metaPath, version, err := c.findMetadataLocation(ident) + if err != nil { + return nil, 0, err + } + + reporter, err := c.reporter.Get(c.props) + if err != nil { + return nil, 0, fmt.Errorf("failed to initialize metrics reporter: %w", err) + } + + tbl, err := table.NewFromLocation( + ctx, ident, metaPath, io.LoadFSFunc(c.props, metaPath), c, + table.WithMetricsReporter(reporter), + table.WithScanPlanningIOProperties(c.props), + ) + if err != nil { + return nil, 0, err + } + + return tbl, version, nil +} + +func (c *Catalog) LoadTable(ctx context.Context, ident table.Identifier) (*table.Table, error) { + tbl, _, err := c.loadTable(ctx, ident) + + return tbl, err +} + +func (c *Catalog) CheckTableExists(_ context.Context, ident table.Identifier) (bool, error) { + if err := validateTableIdentifier(ident); err != nil { + return false, err + } + + // Direct existence checks surface metadata directory read errors; listing + // paths use isTableDirForListing to skip unreadable entries. + exists, err := isTableDir(c.filesystem, c.isLocal, c.tableToPath(ident)) + if err != nil { + return false, err + } + + return exists, nil +} + +func (c *Catalog) CommitTable(ctx context.Context, ident table.Identifier, reqs []table.Requirement, updates []table.Update) (table.Metadata, string, error) { + if err := validateTableIdentifier(ident); err != nil { + return nil, "", err + } + + // Step 1: Load current table (nil for create-via-commit). + current, currentVersion, err := c.loadTable(ctx, ident) + if err != nil && !errors.Is(err, catalog.ErrNoSuchTable) { + return nil, "", err + } + + // Step 2: Validate requirements against current metadata. + var currentMetadata table.Metadata + if current != nil { + currentMetadata = current.Metadata() + } + if err := internal.ValidateRequirements(ident, currentMetadata, reqs); err != nil { + return nil, "", err + } + + // Step 3: Apply updates to produce new metadata. + var baseMeta table.Metadata + var currentMetadataLoc string + + if current != nil { + baseMeta = current.Metadata() + currentMetadataLoc = current.MetadataLocation() + } else { + baseMeta, err = table.NewMetadata(iceberg.NewSchema(0), nil, table.UnsortedSortOrder, "", nil) + if err != nil { + return nil, "", fmt.Errorf("hadoop catalog: failed to create base metadata: %w", err) + } + } + + updated, err := internal.UpdateTableMetadata(baseMeta, updates, currentMetadataLoc) + if err != nil { + return nil, "", fmt.Errorf("hadoop catalog: failed to apply updates: %w", err) + } + + // Step 4: Validate table location has not changed. + if current != nil && updated.Location() != current.Location() { + return nil, "", errors.New("hadoop catalog: table location cannot be changed") + } + + // Step 5: Reject write.metadata.location property. + if v := updated.Properties().Get(table.WriteMetadataLocationKey, ""); v != "" { + return nil, "", fmt.Errorf("hadoop catalog: %s property is not supported", table.WriteMetadataLocationKey) + } + + // Step 6: Determine next version number. + newVersion := currentVersion + 1 + + // Step 7: Create metadata directory if needed (create-via-commit). + metaDir := c.metadataDir(ident) + if err := c.filesystem.MkdirAll(metaDir); err != nil { + return nil, "", fmt.Errorf("hadoop catalog: failed to create metadata directory: %w", err) + } + + compression := updated.Properties().Get(table.MetadataCompressionKey, table.MetadataCompressionDefault) + newMetaPath, err := c.metadataFilePathForCompression(ident, newVersion, compression) + if err != nil { + return nil, "", fmt.Errorf("hadoop catalog: failed to determine metadata file path: %w", err) + } + + tempPath := joinPath(c.isLocal, metaDir, uuid.New().String()+".metadata.json") + + if err := internal.WriteTableMetadata(updated, c.filesystem, tempPath, compression); err != nil { + _ = c.filesystem.Remove(tempPath) + + return nil, "", fmt.Errorf("hadoop catalog: failed to write table metadata: %w", err) + } + + if err := c.commitMetadataFile(ident, newVersion, tempPath, newMetaPath, table.ErrCommitFailed); err != nil { + return nil, "", err + } + + // Step 8: Best-effort version hint update. + c.writeVersionHint(ident, newVersion) + + return updated, newMetaPath, nil +} + +func (c *Catalog) commitMetadataFile(ident table.Identifier, version int, tempPath, metaPath string, conflictErr error) error { Review Comment: My #2067 adds ctx to `io.IO.Open`/`Remove`, and 7 of its 8 `hadoop.go` hunks are in `CreateTable`, `CommitTable`, `commitMetadataFile` and `DropNamespace`, which this PR moves here and into `namespaces.go` unchanged. The two conflict in `hadoop.go`, so whichever lands second has to re-apply the `Remove(ctx, ...)` and `commitMetadataFile(ctx, ...)` edits in the new files by hand. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
