(iceberg-go) branch main updated: feat(catalog): add version-hint management for Hadoop catalog

[at...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

laskoviymishka 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 f026751e feat(catalog): add version-hint management for Hadoop catalog
f026751e is described below

commit f026751eee9710275b6e25590a3388d1e816fcc4
Author: Tanmay Rauth <t_...@apple.com>
AuthorDate: Fri May 1 13:16:23 2026 -0700

    feat(catalog): add version-hint management for Hadoop catalog
    
    [Version-hint
    management](https://github.com/apache/iceberg-go/issues/798#issuecomment-4320784323)
    Implement readVersionHint (reads the integer from version-hint.text,
    returns 0 on any error), writeVersionHint (writes to a temp file then
    renames, best-effort with errors logged but not returned), and
    findVersion with the full three-tier fallback: read hint and validate
    the file exists, scan forward from the hinted version until no file is
    found, and if there's no hint at all, list the metadata directory to
    find the max version then scan forward. Tests cover round-trip
    read/write, missing hint, corrupt hint, stale hint (v3 exists but hint
    says 2), scan-forward behavior, directory listing fallback, and the
    no-metadata-files case.
    
    
    Relates to #798
---
 catalog/hadoop/hadoop.go      |  98 +++++++++++++++++++++++++
 catalog/hadoop/hadoop_test.go | 166 ++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 264 insertions(+)

diff --git a/catalog/hadoop/hadoop.go b/catalog/hadoop/hadoop.go
index 3d73c1f1..e44fc697 100644
--- a/catalog/hadoop/hadoop.go
+++ b/catalog/hadoop/hadoop.go
@@ -22,15 +22,18 @@ import (
 	"errors"
 	"fmt"
 	"iter"
+	"log"
 	"net/url"
 	"os"
 	"path/filepath"
 	"regexp"
+	"strconv"
 	"strings"
 
 	"github.com/apache/iceberg-go"
 	"github.com/apache/iceberg-go/catalog"
 	"github.com/apache/iceberg-go/table"
+	"github.com/google/uuid"
 )
 
 func init() {
@@ -137,6 +140,101 @@ func isTableDir(path string) bool {
 	return false
 }
 
+func (c *Catalog) readVersionHint(ident table.Identifier) int {
+	data, err := os.ReadFile(c.versionHintPath(ident))
+	if err != nil {
+		return 0
+	}
+
+	ver, err := strconv.Atoi(strings.TrimSpace(string(data)))
+	if err != nil || ver <= 0 {
+		return 0
+	}
+
+	return ver
+}
+
+func (c *Catalog) writeVersionHint(ident table.Identifier, version int) {
+	dir := c.metadataDir(ident)
+	tempPath := filepath.Join(dir, uuid.New().String()+"-version-hint.temp")
+	hintPath := c.versionHintPath(ident)
+
+	content := []byte(strconv.Itoa(version))
+	if err := os.WriteFile(tempPath, content, 0o644); err != nil {
+		log.Printf("hadoop catalog: failed to write version hint temp file: %v", err)
+
+		return
+	}
+
+	if err := os.Rename(tempPath, hintPath); err != nil {
+		log.Printf("hadoop catalog: failed to rename version hint: %v", err)
+		os.Remove(tempPath)
+	}
+}
+
+// metadataVersionExists checks whether a metadata file for the given version
+// exists in either plain or gzip-compressed form.
+func (c *Catalog) metadataVersionExists(ident table.Identifier, version int) bool {
+	dir := c.metadataDir(ident)
+	plain := filepath.Join(dir, fmt.Sprintf("v%d.metadata.json", version))
+
+	if _, err := os.Stat(plain); err == nil {
+		return true
+	}
+
+	gz := filepath.Join(dir, fmt.Sprintf("v%d.gz.metadata.json", version))
+
+	_, err := os.Stat(gz)
+
+	return err == nil
+}
+
+func (c *Catalog) scanForward(ident table.Identifier, start int) int {
+	ver := start
+	for c.metadataVersionExists(ident, ver+1) {
+		ver++
+	}
+
+	return ver
+}
+
+func (c *Catalog) findVersion(ident table.Identifier) (int, error) {
+	hint := c.readVersionHint(ident)
+	if hint > 0 && c.metadataVersionExists(ident, hint) {
+		return c.scanForward(ident, hint), nil
+	}
+
+	dir := c.metadataDir(ident)
+
+	entries, err := os.ReadDir(dir)
+	if err != nil {
+		return 0, fmt.Errorf("hadoop catalog: cannot read metadata directory for %s: %w",
+			strings.Join(ident, "."), catalog.ErrNoSuchTable)
+	}
+
+	maxVer := 0
+	for _, e := range entries {
+		if e.IsDir() {
+			continue
+		}
+
+		matches := versionPattern.FindStringSubmatch(e.Name())
+		if len(matches) == 2 {
+			v, _ := strconv.Atoi(matches[1])
+			if v > maxVer {
+				maxVer = v
+			}
+		}
+	}
+
+	if maxVer == 0 {
+		return 0, fmt.Errorf("hadoop catalog: no metadata files found for table %s: %w",
+			strings.Join(ident, "."), catalog.ErrNoSuchTable)
+	}
+
+	return c.scanForward(ident, maxVer), nil
+}
+
 func (c *Catalog) CreateTable(_ context.Context, _ table.Identifier, _ *iceberg.Schema, _ ...catalog.CreateTableOpt) (*table.Table, error) {
 	return nil, errors.New("hadoop catalog: CreateTable not yet implemented")
 }
diff --git a/catalog/hadoop/hadoop_test.go b/catalog/hadoop/hadoop_test.go
index c9eef7ae..23ecaff2 100644
--- a/catalog/hadoop/hadoop_test.go
+++ b/catalog/hadoop/hadoop_test.go
@@ -25,6 +25,7 @@ import (
 
 	"github.com/apache/iceberg-go"
 	"github.com/apache/iceberg-go/catalog"
+	"github.com/apache/iceberg-go/table"
 	"github.com/stretchr/testify/suite"
 )
 
@@ -220,3 +221,168 @@ func (s *HadoopCatalogTestSuite) TestIsTableDirWithGzipMetadata() {
 func (s *HadoopCatalogTestSuite) TestIsTableDirNonExistentPath() {
 	s.False(isTableDir(filepath.Join(s.warehouse, "does", "not", "exist")))
 }
+
+func (s *HadoopCatalogTestSuite) createMetadataFile(ident table.Identifier, version int) {
+	path := s.cat.metadataFilePath(ident, version)
+	s.Require().NoError(os.MkdirAll(filepath.Dir(path), 0o755))
+	s.Require().NoError(os.WriteFile(path, nil, 0o644))
+}
+
+func (s *HadoopCatalogTestSuite) TestReadWriteVersionHint() {
+	ident := []string{"ns", "tbl"}
+	s.Require().NoError(os.MkdirAll(s.cat.metadataDir(ident), 0o755))
+	s.cat.writeVersionHint(ident, 42)
+	s.Equal(42, s.cat.readVersionHint(ident))
+}
+
+func (s *HadoopCatalogTestSuite) TestReadVersionHintMissing() {
+	s.Equal(0, s.cat.readVersionHint([]string{"ns", "tbl"}))
+}
+
+func (s *HadoopCatalogTestSuite) TestReadVersionHintCorrupt() {
+	ident := []string{"ns", "tbl"}
+	dir := s.cat.metadataDir(ident)
+	s.Require().NoError(os.MkdirAll(dir, 0o755))
+	s.Require().NoError(os.WriteFile(s.cat.versionHintPath(ident), []byte("garbage"), 0o644))
+	s.Equal(0, s.cat.readVersionHint(ident))
+}
+
+func (s *HadoopCatalogTestSuite) TestWriteVersionHintCreatesFile() {
+	ident := []string{"ns", "tbl"}
+	s.Require().NoError(os.MkdirAll(s.cat.metadataDir(ident), 0o755))
+	s.cat.writeVersionHint(ident, 7)
+	data, err := os.ReadFile(s.cat.versionHintPath(ident))
+	s.Require().NoError(err)
+	s.Equal("7", string(data))
+}
+
+func (s *HadoopCatalogTestSuite) TestFindVersionFromHint() {
+	ident := []string{"ns", "tbl"}
+	s.createMetadataFile(ident, 1)
+	s.cat.writeVersionHint(ident, 1)
+	ver, err := s.cat.findVersion(ident)
+	s.Require().NoError(err)
+	s.Equal(1, ver)
+}
+
+func (s *HadoopCatalogTestSuite) TestFindVersionNoHint() {
+	ident := []string{"ns", "tbl"}
+	for i := 1; i <= 3; i++ {
+		s.createMetadataFile(ident, i)
+	}
+
+	ver, err := s.cat.findVersion(ident)
+	s.Require().NoError(err)
+	s.Equal(3, ver)
+}
+
+func (s *HadoopCatalogTestSuite) TestFindVersionNoMetadata() {
+	ident := []string{"ns", "tbl"}
+	s.Require().NoError(os.MkdirAll(s.cat.metadataDir(ident), 0o755))
+	_, err := s.cat.findVersion(ident)
+	s.Require().Error(err)
+	s.ErrorIs(err, catalog.ErrNoSuchTable)
+}
+
+func (s *HadoopCatalogTestSuite) TestFindVersionScanForward() {
+	ident := []string{"ns", "tbl"}
+	for i := 1; i <= 5; i++ {
+		s.createMetadataFile(ident, i)
+	}
+
+	s.cat.writeVersionHint(ident, 2)
+	ver, err := s.cat.findVersion(ident)
+	s.Require().NoError(err)
+	s.Equal(5, ver)
+}
+
+func (s *HadoopCatalogTestSuite) TestFindVersionNoMetadataDir() {
+	_, err := s.cat.findVersion([]string{"ns", "tbl"})
+	s.Require().Error(err)
+	s.ErrorIs(err, catalog.ErrNoSuchTable)
+}
+
+func (s *HadoopCatalogTestSuite) TestFindVersionStaleHintPointsToDeletedVersion() {
+	// hint=5, but only v1-v3 exist. Hint file points to a version whose
+	// metadata was deleted. Should fall through to dir listing and find v3.
+	ident := []string{"ns", "tbl"}
+	for i := 1; i <= 3; i++ {
+		s.createMetadataFile(ident, i)
+	}
+
+	s.cat.writeVersionHint(ident, 5)
+	ver, err := s.cat.findVersion(ident)
+	s.Require().NoError(err)
+	s.Equal(3, ver)
+}
+
+func (s *HadoopCatalogTestSuite) TestReadVersionHintZeroValue() {
+	ident := []string{"ns", "tbl"}
+	s.Require().NoError(os.MkdirAll(s.cat.metadataDir(ident), 0o755))
+	s.Require().NoError(os.WriteFile(s.cat.versionHintPath(ident), []byte("0"), 0o644))
+	s.Equal(0, s.cat.readVersionHint(ident))
+}
+
+func (s *HadoopCatalogTestSuite) TestReadVersionHintWithWhitespace() {
+	ident := []string{"ns", "tbl"}
+	s.Require().NoError(os.MkdirAll(s.cat.metadataDir(ident), 0o755))
+	s.Require().NoError(os.WriteFile(s.cat.versionHintPath(ident), []byte("  42\n"), 0o644))
+	s.Equal(42, s.cat.readVersionHint(ident))
+}
+
+func (s *HadoopCatalogTestSuite) TestWriteVersionHintNoMetadataDir() {
+	// metadata dir doesn't exist — writeVersionHint should not panic,
+	// just log a warning silently.
+	ident := []string{"nonexistent", "tbl"}
+	s.NotPanics(func() {
+		s.cat.writeVersionHint(ident, 1)
+	})
+}
+
+func (s *HadoopCatalogTestSuite) TestFindVersionIgnoresTempFiles() {
+	// metadata dir has both valid metadata files and orphaned temp files
+	// from crashed commits. findVersion should ignore temp files.
+	ident := []string{"ns", "tbl"}
+	dir := s.cat.metadataDir(ident)
+	s.Require().NoError(os.MkdirAll(dir, 0o755))
+	s.createMetadataFile(ident, 1)
+	s.createMetadataFile(ident, 2)
+	s.Require().NoError(os.WriteFile(filepath.Join(dir, "abc123-def456.metadata.json"), nil, 0o644))
+	s.Require().NoError(os.WriteFile(filepath.Join(dir, "abc123-version-hint.temp"), nil, 0o644))
+
+	ver, err := s.cat.findVersion(ident)
+	s.Require().NoError(err)
+	s.Equal(2, ver)
+}
+
+func (s *HadoopCatalogTestSuite) TestFindVersionGzipOnlyWithHint() {
+	// Table has only gzip metadata. Hint validation and scanForward
+	// should recognize gzip-compressed metadata files.
+	ident := []string{"ns", "tbl"}
+	dir := s.cat.metadataDir(ident)
+	s.Require().NoError(os.MkdirAll(dir, 0o755))
+	s.Require().NoError(os.WriteFile(filepath.Join(dir, "v1.gz.metadata.json"), nil, 0o644))
+	s.Require().NoError(os.WriteFile(filepath.Join(dir, "v2.gz.metadata.json"), nil, 0o644))
+	s.cat.writeVersionHint(ident, 1)
+
+	ver, err := s.cat.findVersion(ident)
+	s.Require().NoError(err)
+	// Hint=1 validated via gzip path, scanForward finds v2.gz → returns 2
+	s.Equal(2, ver)
+}
+
+func (s *HadoopCatalogTestSuite) TestFindVersionMixedGzipAndPlain() {
+	// v1 and v2 are plain, v3 is gzip-compressed. scanForward must
+	// check both formats to avoid returning a stale version.
+	ident := []string{"ns", "tbl"}
+	dir := s.cat.metadataDir(ident)
+	s.Require().NoError(os.MkdirAll(dir, 0o755))
+	s.createMetadataFile(ident, 1)
+	s.createMetadataFile(ident, 2)
+	s.Require().NoError(os.WriteFile(filepath.Join(dir, "v3.gz.metadata.json"), nil, 0o644))
+	s.cat.writeVersionHint(ident, 1)
+
+	ver, err := s.cat.findVersion(ident)
+	s.Require().NoError(err)
+	s.Equal(3, ver)
+}