laskoviymishka commented on code in PR #1326:
URL: https://github.com/apache/iceberg-go/pull/1326#discussion_r3530090958
##########
catalog/hadoop/hadoop.go:
##########
@@ -689,18 +711,82 @@ func (c *Catalog) CommitTable(ctx context.Context, ident
table.Identifier, reqs
return updated, newMetaPath, nil
}
-func (c *Catalog) commitMetadataFile(ident table.Identifier, tempPath,
metaPath string, conflictErr error) error {
- if err := c.filesystem.RenameNoReplace(tempPath, metaPath); err != nil {
- _ = c.filesystem.Remove(tempPath)
+func (c *Catalog) commitMetadataFile(ident table.Identifier, version int,
tempPath, metaPath string, conflictErr error) error {
+ claimPath := c.metadataVersionClaimPath(ident, version)
+ for {
+ if err := c.filesystem.RenameNoReplace(tempPath, claimPath);
err != nil {
+ if !errors.Is(err, fs.ErrExist) {
+ _ = c.filesystem.Remove(tempPath)
+
+ return fmt.Errorf("hadoop catalog: failed to
claim metadata version: %w", err)
+ }
+
+ existingPath, exists :=
c.metadataVersionLocation(ident, version)
+ if exists {
+ _ = c.filesystem.Remove(tempPath)
+
+ return fmt.Errorf("%w: metadata file already
exists for table %s: %s",
+ conflictErr, strings.Join(ident, "."),
existingPath)
+ }
+
+ claimInfo, err := c.filesystem.Stat(claimPath)
+ if err != nil {
Review Comment:
Small one that folds into the crash-recovery thread zeroshade opened: if a
competitor cleans the claim and publishes between the failing rename above and
this `Stat`, `Stat` returns `fs.ErrNotExist` and we return a generic "failed to
inspect stale metadata claim" that isn't wrapped in `conflictErr` — so a caller
distinguishing retryable-conflict from abort-infra spuriously aborts on a race
that actually resolved fine.
I'd special-case `fs.ErrNotExist` here and just `continue` — the top of the
loop re-checks `metadataVersionLocation` and will surface a proper conflict if
it was published.
##########
catalog/hadoop/hadoop.go:
##########
@@ -450,6 +467,17 @@ func (c *Catalog) metadataVersionExists(ident
table.Identifier, version int) (bo
return found, nil
}
+func (c *Catalog) metadataVersionLocation(ident table.Identifier, version int)
(string, bool) {
+ files, _, err := c.scanMetadataFiles(ident)
+ if err != nil {
+ return "", false
Review Comment:
This is the path canonicalization from last round and it reads well — but
the helper it's built on drops its error: any `scanMetadataFiles` failure
returns `("", false)`, which is indistinguishable from "this version doesn't
exist."
That's load-bearing at the post-claim call to this in `commitMetadataFile` —
it's the only cross-codec conflict guard left. If `WalkDir` hits a transient
error there, we get `false`, fall through to `RenameNoReplace` onto the plain
filename (which succeeds because that path doesn't exist), and end up with
`v2.metadata.json` and `v2.gz.metadata.json` both at the same version.
`betterThan` then prefers the plain file and the compressed commit is silently
gone.
I'd give it an error return and fail the commit on it rather than folding it
into absence:
```go
func (c *Catalog) metadataVersionLocation(ident table.Identifier, version
int) (string, bool, error) {
files, _, err := c.scanMetadataFiles(ident)
if err != nil {
return "", false, err
}
file, ok := files[version]
return file.location, ok, nil
}
```
Then propagate it at the call site as an infra error (not wrapped in
`conflictErr`), so a scan failure aborts the commit instead of publishing over
a live version.
##########
catalog/hadoop/hadoop_test.go:
##########
@@ -1488,7 +1605,144 @@ func (s *HadoopCatalogTestSuite)
TestCreateTableConcurrentMetadataPublishConflic
s.Equal(1, successes)
s.Equal(1, conflicts)
- s.FileExists(s.cat.metadataFilePath([]string{"ns", "tbl"}, 1))
+ metaPath, err := s.cat.metadataFilePathForCompression(ident, 1,
table.MetadataCompressionCodecNone)
+ s.Require().NoError(err)
+ s.FileExists(metaPath)
+ s.NoFileExists(s.cat.metadataVersionClaimPath(ident, 1))
+}
+
+func (s *HadoopCatalogTestSuite)
TestCreateTableConcurrentMixedCodecVersionClaim() {
+ ctx := context.Background()
+ s.Require().NoError(os.Mkdir(filepath.Join(s.warehouse, "ns"), 0o755))
+ ident := []string{"ns", "tbl"}
+ s.cat.filesystem =
newBarrierRenameNoReplaceFS(filepath.Base(s.cat.metadataVersionClaimPath(ident,
1)))
+
+ type createResult struct {
+ metaLoc string
+ err error
+ }
+
+ results := make(chan createResult, 2)
+ var wg sync.WaitGroup
+ for _, props := range []iceberg.Properties{
+ nil,
+ {table.MetadataCompressionKey:
table.MetadataCompressionCodecGzip},
+ } {
+ props := props
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+
+ opts := []catalog.CreateTableOpt(nil)
+ if props != nil {
+ opts = append(opts,
catalog.WithProperties(props))
+ }
+
+ tbl, err := s.cat.CreateTable(ctx, ident,
s.testSchema(), opts...)
+ result := createResult{err: err}
+ if err == nil {
+ result.metaLoc = tbl.MetadataLocation()
+ }
+
+ results <- result
+ }()
+ }
+
+ wg.Wait()
+ close(results)
+
+ successes := 0
+ conflicts := 0
+ var successLoc string
+ for result := range results {
+ if result.err == nil {
+ successes++
+ successLoc = result.metaLoc
+ s.Contains([]string{"v1.metadata.json",
"v1.gz.metadata.json"}, filepath.Base(result.metaLoc))
+
+ continue
+ }
+
+ conflicts++
+ s.ErrorIs(result.err, catalog.ErrTableAlreadyExists)
+ }
+
+ s.Equal(1, successes)
+ s.Equal(1, conflicts)
+
+ plainPath, err := s.cat.metadataFilePathForCompression(ident, 1,
table.MetadataCompressionCodecNone)
+ s.Require().NoError(err)
+ gzipPath, err := s.cat.metadataFilePathForCompression(ident, 1,
table.MetadataCompressionCodecGzip)
+ s.Require().NoError(err)
+
+ existing := 0
+ for _, path := range []string{plainPath, gzipPath} {
+ if _, err := os.Stat(path); err == nil {
+ existing++
+ } else {
+ s.True(os.IsNotExist(err))
Review Comment:
`os.IsNotExist` doesn't unwrap, so this misses a wrapped not-exist.
`s.ErrorIs(err, fs.ErrNotExist)` is the unwrapping form — same pattern in the
other new mixed-codec test.
##########
catalog/hadoop/hadoop_test.go:
##########
@@ -1488,7 +1605,144 @@ func (s *HadoopCatalogTestSuite)
TestCreateTableConcurrentMetadataPublishConflic
s.Equal(1, successes)
s.Equal(1, conflicts)
- s.FileExists(s.cat.metadataFilePath([]string{"ns", "tbl"}, 1))
+ metaPath, err := s.cat.metadataFilePathForCompression(ident, 1,
table.MetadataCompressionCodecNone)
+ s.Require().NoError(err)
+ s.FileExists(metaPath)
+ s.NoFileExists(s.cat.metadataVersionClaimPath(ident, 1))
+}
+
+func (s *HadoopCatalogTestSuite)
TestCreateTableConcurrentMixedCodecVersionClaim() {
+ ctx := context.Background()
+ s.Require().NoError(os.Mkdir(filepath.Join(s.warehouse, "ns"), 0o755))
+ ident := []string{"ns", "tbl"}
+ s.cat.filesystem =
newBarrierRenameNoReplaceFS(filepath.Base(s.cat.metadataVersionClaimPath(ident,
1)))
+
+ type createResult struct {
+ metaLoc string
+ err error
+ }
+
+ results := make(chan createResult, 2)
+ var wg sync.WaitGroup
+ for _, props := range []iceberg.Properties{
+ nil,
+ {table.MetadataCompressionKey:
table.MetadataCompressionCodecGzip},
+ } {
+ props := props
Review Comment:
go.mod is on 1.25, so the per-iteration loop var is automatic — this copy
(and the one in the commit-side test) is redundant.
##########
catalog/hadoop/hadoop.go:
##########
@@ -655,31 +687,21 @@ func (c *Catalog) CommitTable(ctx context.Context, ident
table.Identifier, reqs
return nil, "", fmt.Errorf("hadoop catalog: failed to create
metadata directory: %w", err)
}
- newMetaPath := c.metadataFilePath(ident, newVersion)
- tempPath := joinPath(c.isLocal, metaDir,
uuid.New().String()+".metadata.json")
-
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)
}
- exists, err := c.metadataVersionExists(ident, newVersion)
- if err != nil {
- _ = c.filesystem.Remove(tempPath)
-
- return nil, "", fmt.Errorf("hadoop catalog: failed to inspect
metadata directory for version %d: %w",
- newVersion, err)
- }
- if exists {
- _ = c.filesystem.Remove(tempPath)
-
- return nil, "", fmt.Errorf("hadoop catalog: version %d already
exists for table %s",
- newVersion, strings.Join(ident, "."))
- }
- if err := c.commitMetadataFile(ident, tempPath, newMetaPath,
table.ErrCommitFailed); err != nil {
+ if err := c.commitMetadataFile(ident, newVersion, tempPath,
newMetaPath, table.ErrCommitFailed); err != nil {
Review Comment:
With this pre-check gone, `metadataVersionExists` loses its last caller —
the claim protocol supersedes it. I'd drop it while we're here so it doesn't
drift out of sync with the codec-aware path.
##########
catalog/hadoop/hadoop_test.go:
##########
@@ -564,18 +613,29 @@ func (s *HadoopCatalogTestSuite)
TestIsTableDirWithGzipMetadata() {
s.True(s.requireIsTableDir(tableDir))
}
+func (s *HadoopCatalogTestSuite) TestIsTableDirIgnoresZstdMetadata() {
+ tableDir := filepath.Join(s.warehouse, "ns", "tbl")
+ metaDir := filepath.Join(tableDir, "metadata")
+ s.Require().NoError(os.MkdirAll(metaDir, 0o755))
+ s.Require().NoError(os.WriteFile(filepath.Join(metaDir,
"v1.zstd.metadata.json"), nil, 0o644))
+
+ s.False(isTableDir(s.cat.filesystem, s.cat.isLocal, tableDir))
Review Comment:
`isTableDir` returns an error alongside the bool, and passing the call
straight into `s.False` slides it into the assertion's message args instead of
checking it. The sibling tests go through `requireIsTableDir`, which surfaces
it — I'd use that here too so a real scan failure can't masquerade as a clean
false.
--
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]
