You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2025/05/08 16:52:31 UTC
(iceberg-go) branch main updated: fix(table): logging error when deleting old metadata (#422)
This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 027d0a1 fix(table): logging error when deleting old metadata (#422)
027d0a1 is described below
commit 027d0a14fe2a2cd263370ebc293355ba50fb40a9
Author: Shreyas Mishra <55...@users.noreply.github.com>
AuthorDate: Thu May 8 22:22:26 2025 +0530
fix(table): logging error when deleting old metadata (#422)
Fixes #411
---------
Signed-off-by: Shreyas220 <sh...@gmail.com>
---
table/table.go | 13 ++---
table/table_test.go | 143 ++++++++++++++++++++++++++++++++++++++++++++++++++++
2 files changed, 148 insertions(+), 8 deletions(-)
diff --git a/table/table.go b/table/table.go
index 05f40b0..c91fdfd 100644
--- a/table/table.go
+++ b/table/table.go
@@ -19,8 +19,8 @@ package table
import (
"context"
- "fmt"
"iter"
+ "log"
"runtime"
"slices"
@@ -196,9 +196,7 @@ func (t Table) doCommit(ctx context.Context, updates []Update, reqs []Requiremen
return nil, err
}
- if err := deleteOldMetadata(t.fs, t.metadata, newMeta); err != nil {
- return nil, err
- }
+ deleteOldMetadata(t.fs, t.metadata, newMeta)
return New(t.identifier, newMeta, newLoc, t.fs, t.cat), nil
}
@@ -219,7 +217,7 @@ func getFiles(it iter.Seq[MetadataLogEntry]) iter.Seq[string] {
}
}
-func deleteOldMetadata(fs io.IO, baseMeta, newMeta Metadata) error {
+func deleteOldMetadata(fs io.IO, baseMeta, newMeta Metadata) {
deleteAfterCommit := newMeta.Properties().GetBool(MetadataDeleteAfterCommitEnabledKey,
MetadataDeleteAfterCommitEnabledDefault)
@@ -230,12 +228,11 @@ func deleteOldMetadata(fs io.IO, baseMeta, newMeta Metadata) error {
for _, file := range toRemove {
if err := fs.Remove(file); err != nil {
- return fmt.Errorf("failed to delete old metadata file %s: %w", file, err)
+ // Log the error instead of raising it when deleting old metadata files, as an external entity like a compactor may have already deleted them
+ log.Printf("Warning: Failed to delete old metadata file: %s error: %v", file, err)
}
}
}
-
- return nil
}
type ScanOption func(*Scan)
diff --git a/table/table_test.go b/table/table_test.go
index 3a8338f..108493b 100644
--- a/table/table_test.go
+++ b/table/table_test.go
@@ -21,10 +21,14 @@ import (
"bytes"
"context"
"fmt"
+ "log"
+ "os"
+ "path/filepath"
"slices"
"strconv"
"strings"
"testing"
+ "time"
"github.com/apache/arrow-go/v18/arrow"
"github.com/apache/arrow-go/v18/arrow/array"
@@ -1205,3 +1209,142 @@ func TestNullableStructRequiredField(t *testing.T) {
require.NoError(t, err)
require.NotNil(t, stagedTbl)
}
+
+type DeleteOldMetadataMockedCatalog struct{}
+
+func (m *DeleteOldMetadataMockedCatalog) LoadTable(ctx context.Context, ident table.Identifier, props iceberg.Properties) (*table.Table, error) {
+ return nil, nil
+}
+
+func (m *DeleteOldMetadataMockedCatalog) CommitTable(ctx context.Context, tbl *table.Table, reqs []table.Requirement, updates []table.Update) (table.Metadata, string, error) {
+ bldr, err := table.MetadataBuilderFromBase(tbl.Metadata())
+ if err != nil {
+ return nil, "", err
+ }
+
+ location := tbl.Metadata().Location()
+
+ randid := uuid.New().String()
+ metdatafile := fmt.Sprintf("%s/metadata/%s.metadata.json", location, randid)
+
+ // removing old metadata files
+ bldr.TrimMetadataLogs(0)
+
+ bldr.AppendMetadataLog(table.MetadataLogEntry{
+ MetadataFile: metdatafile,
+ TimestampMs: time.Now().UnixMilli(),
+ })
+
+ for _, u := range updates {
+ if err := u.Apply(bldr); err != nil {
+ return nil, "", err
+ }
+ }
+
+ meta, err := bldr.Build()
+ if err != nil {
+ return nil, "", err
+ }
+
+ return meta, metdatafile, nil
+}
+
+func createMetadataFile(metadatadir, metadataFile string) error {
+ // Ensure the directory exists
+ metadataDir := filepath.Dir(metadatadir)
+ err := os.MkdirAll(metadataDir, 0o755)
+ if err != nil {
+ return err
+ }
+
+ err = os.WriteFile(metadataFile, []byte("first commit metadata content"), 0o644)
+
+ return err
+}
+
+func (t *TableWritingTestSuite) TestDeleteOldMetadataLogsErrorOnFileNotFound() {
+ // capture logs to validate that no error is logged
+ var logBuf bytes.Buffer
+ log.SetOutput(&logBuf)
+ defer log.SetOutput(os.Stderr) // restore default output
+
+ fs := iceio.LocalFS{}
+ files := make([]string, 0)
+ for i := range 5 {
+ filePath := fmt.Sprintf("%s/file_v%d/data-%d.parquet", t.location, t.formatVersion, i)
+ t.writeParquet(fs, filePath, t.arrTablePromotedTypes)
+ files = append(files, filePath)
+ }
+
+ ident := table.Identifier{"default", "file_v" + strconv.Itoa(t.formatVersion)}
+ meta, err := table.NewMetadata(t.tableSchemaPromotedTypes, iceberg.UnpartitionedSpec,
+ table.UnsortedSortOrder, t.location, iceberg.Properties{"format-version": strconv.Itoa(t.formatVersion), "write.metadata.delete-after-commit.enabled": "true"})
+ t.Require().NoError(err)
+
+ tbl := table.New(ident, meta, t.getMetadataLoc(), fs, &DeleteOldMetadataMockedCatalog{})
+ ctx := context.Background()
+
+ // transaction 1 to create metadata file
+ tx := tbl.NewTransaction()
+ tx.AddFiles(ctx, files[0:1], nil, false)
+ tbl_new, err := tx.Commit(ctx)
+ t.Require().NoError(err)
+
+ // transaction 2 to add files
+ tx_new := tbl_new.NewTransaction()
+ tx_new.AddFiles(ctx, files[1:2], nil, false)
+
+ _, err = tx_new.Commit(ctx)
+ t.Require().NoError(err)
+
+ // validate that error is logged
+ logOutput := logBuf.String()
+ t.Contains(logOutput, "Warning: Failed to delete old metadata file")
+ t.Contains(logOutput, "no such file or directory")
+}
+
+func (t *TableWritingTestSuite) TestDeleteOldMetadataNoErrorLogsOnFileFound() {
+ // capture logs to validate that no error is logged
+ var logBuf bytes.Buffer
+ log.SetOutput(&logBuf)
+ defer log.SetOutput(os.Stderr) // restore default output
+
+ fs := iceio.LocalFS{}
+ files := make([]string, 0)
+ for i := range 5 {
+ filePath := fmt.Sprintf("%s/file_v%d/data-%d.parquet", t.location, t.formatVersion, i)
+ t.writeParquet(fs, filePath, t.arrTablePromotedTypes)
+ files = append(files, filePath)
+ }
+
+ ident := table.Identifier{"default", "file_v" + strconv.Itoa(t.formatVersion)}
+ meta, err := table.NewMetadata(t.tableSchemaPromotedTypes, iceberg.UnpartitionedSpec, table.UnsortedSortOrder, t.location, iceberg.Properties{"format-version": strconv.Itoa(t.formatVersion), "write.metadata.delete-after-commit.enabled": "true"})
+ t.Require().NoError(err)
+
+ tbl := table.New(ident, meta, t.getMetadataLoc(), fs, &DeleteOldMetadataMockedCatalog{})
+
+ ctx := context.Background()
+
+ // transaction 1 to create metadata file
+ tx := tbl.NewTransaction()
+ tx.AddFiles(ctx, files[0:1], nil, false)
+ tbl_new, err := tx.Commit(ctx)
+ t.Require().NoError(err)
+
+ // Now we have the first metadata location - create the file there so that deleteOldMetadata does not log an error
+ firstMetadataLoc := tbl_new.MetadataLocation()
+ metadataFile := tbl_new.MetadataLocation()
+ err = createMetadataFile(firstMetadataLoc, metadataFile)
+ t.Require().NoError(err)
+
+ // transaction 2 to add files
+ tx_new := tbl_new.NewTransaction()
+ tx_new.AddFiles(ctx, files[1:2], nil, false)
+ _, err = tx_new.Commit(ctx)
+ t.Require().NoError(err)
+
+ // validate that no error is logged
+ logOutput := logBuf.String()
+ t.NotContains(logOutput, "Warning: Failed to delete old metadata file")
+ t.NotContains(logOutput, "no such file or directory")
+}