You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by at...@apache.org on 2026/08/07 15:53:35 UTC
(iceberg-go) branch main updated: fix(manifest): make empty writer close terminal (#1610)
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 79edbbd1 fix(manifest): make empty writer close terminal (#1610)
79edbbd1 is described below
commit 79edbbd19f81fd85647ebfdf0fbddfe54b32a4c2
Author: Minh Vu <vu...@gmail.com>
AuthorDate: Fri Aug 7 17:53:30 2026 +0200
fix(manifest): make empty writer close terminal (#1610)
* fix(manifest): make empty writer close terminal
Signed-off-by: Minh Vu <vu...@gmail.com>
* fix(manifest): avoid duplicate close errors
* fix(manifest): finalize terminal writer close
---------
Signed-off-by: Minh Vu <vu...@gmail.com>
---
errors.go | 1 +
manifest.go | 41 +++++++---
manifest_test.go | 189 ++++++++++++++++++++++++++++++++++++++++++++
table/snapshot_producers.go | 32 +++++++-
4 files changed, 250 insertions(+), 13 deletions(-)
diff --git a/errors.go b/errors.go
index 1d031c6f..45531c3a 100644
--- a/errors.go
+++ b/errors.go
@@ -29,6 +29,7 @@ var (
ErrInvalidFormatVersion = fmt.Errorf("%w: invalid format version", ErrInvalidArgument)
ErrInvalidSchema = errors.New("invalid schema")
ErrInvalidPartitionSpec = errors.New("invalid partition spec")
+ ErrEmptyManifest = errors.New("empty manifest file has been written")
ErrInvalidTransform = errors.New("invalid transform syntax")
ErrType = errors.New("type error")
ErrBadCast = errors.New("could not cast value")
diff --git a/manifest.go b/manifest.go
index f2912eb8..f54538a0 100644
--- a/manifest.go
+++ b/manifest.go
@@ -1275,9 +1275,10 @@ func constructPartitionSummaries(spec PartitionSpec, schema *Schema, partitions
}
type ManifestWriter struct {
- closed bool
- version int
- impl writerImpl
+ closed bool
+ closeErr error
+ version int
+ impl writerImpl
output io.Writer
writer *ocf.Writer
@@ -1372,16 +1373,30 @@ func NewManifestWriter(version int, out io.Writer, spec PartitionSpec, schema *S
func (w *ManifestWriter) Close() error {
if w.closed {
- return nil
+ return w.closeErr
}
+ w.closed = true
+ var emptyErr error
if w.addedFiles+w.existingFiles+w.deletedFiles == 0 {
- return errors.New("empty manifest file has been written")
+ emptyErr = ErrEmptyManifest
}
- w.closed = true
+ var writerErr error
+ if w.writer != nil {
+ writerErr = w.writer.Close()
+ }
- return w.writer.Close()
+ switch {
+ case emptyErr == nil:
+ w.closeErr = writerErr
+ case writerErr == nil:
+ w.closeErr = emptyErr
+ default:
+ w.closeErr = errors.Join(emptyErr, writerErr)
+ }
+
+ return w.closeErr
}
type ManifestFileOption func(mf *manifestFile)
@@ -1810,7 +1825,11 @@ func WriteManifest(
if err != nil {
return nil, err
}
- defer internal.CheckedClose(w, &err)
+ defer func() {
+ if !w.closed {
+ internal.CheckedClose(w, &err)
+ }
+ }()
for _, entry := range entries {
if err := w.addEntry(entry.(*manifestEntry)); err != nil {
@@ -1851,7 +1870,11 @@ func WriteManifestV3(
if err != nil {
return nil, 0, err
}
- defer internal.CheckedClose(w, &err)
+ defer func() {
+ if !w.closed {
+ internal.CheckedClose(w, &err)
+ }
+ }()
for _, entry := range entries {
if err := w.addEntry(entry.(*manifestEntry)); err != nil {
diff --git a/manifest_test.go b/manifest_test.go
index 7cc73030..735c07e1 100644
--- a/manifest_test.go
+++ b/manifest_test.go
@@ -514,6 +514,41 @@ type ManifestTestSuite struct {
v3ManifestEntries bytes.Buffer
}
+type manifestFailingWriter struct {
+ err error
+}
+
+func (w manifestFailingWriter) Write([]byte) (int, error) {
+ return 0, w.err
+}
+
+type manifestCloseErrorCodec struct {
+ ocf.Codec
+ err error
+}
+
+func (c manifestCloseErrorCodec) Close() error {
+ return c.err
+}
+
+func (m *ManifestTestSuite) setManifestWriterCloseError(writer *ManifestWriter, closeErr error) {
+ m.T().Helper()
+
+ avroSchema := writer.writer.Schema()
+ m.Require().NoError(writer.writer.Close())
+
+ replacement, err := ocf.NewWriter(
+ io.Discard,
+ avroSchema,
+ ocf.WithCodec(manifestCloseErrorCodec{
+ Codec: ocf.DeflateCodec(flate.DefaultCompression),
+ err: closeErr,
+ }),
+ )
+ m.Require().NoError(err)
+ writer.writer = replacement
+}
+
func (m *ManifestTestSuite) writeManifestList() {
err := WriteManifestList(1, &m.v1ManifestList, snapshotID, nil, nil, 0, manifestFileRecordsV1)
m.Require().NoError(err)
@@ -2152,6 +2187,160 @@ func (m *ManifestTestSuite) TestManifestWriterMeta() {
m.Equal("[]", string(md["partition-spec"]))
}
+func (m *ManifestTestSuite) TestEmptyManifestWriterCloseIsTerminal() {
+ for _, version := range []int{1, 2, 3} {
+ m.Run("v"+strconv.Itoa(version), func() {
+ var out bytes.Buffer
+ writer, err := NewManifestWriter(version, &out, *UnpartitionedSpec, testSchema, snapshotID)
+ m.Require().NoError(err)
+
+ firstErr := writer.Close()
+ m.Require().ErrorIs(firstErr, ErrEmptyManifest)
+ m.ErrorContains(writer.Add(manifestEntryV2Records[0]), "closed manifest writer")
+ m.Same(firstErr, writer.Close())
+
+ _, err = writer.ToManifestFile("manifest.avro", int64(out.Len()))
+ m.Same(firstErr, err)
+
+ manifest := NewManifestFile(
+ version,
+ "manifest.avro",
+ int64(out.Len()),
+ int32(UnpartitionedSpec.ID()),
+ snapshotID,
+ ).Build()
+ entries, err := ReadManifest(manifest, bytes.NewReader(out.Bytes()), false)
+ m.Require().NoError(err)
+ m.Empty(entries)
+ })
+ }
+}
+
+func (m *ManifestTestSuite) TestManifestWriterSuccessfulCloseIsTerminal() {
+ entries := map[int]ManifestEntry{
+ 1: manifestEntryV1Records[0],
+ 2: manifestEntryV2Records[0],
+ 3: manifestEntryV3Records[0],
+ }
+ for _, version := range []int{1, 2, 3} {
+ m.Run("v"+strconv.Itoa(version), func() {
+ writer, err := NewManifestWriter(version, io.Discard, *UnpartitionedSpec, testSchema, snapshotID)
+ m.Require().NoError(err)
+ m.Require().NoError(writer.Add(entries[version]))
+
+ m.NoError(writer.Close())
+ m.NoError(writer.Close())
+ })
+ }
+}
+
+func (m *ManifestTestSuite) TestEmptyManifestWriterCloseAfterConstructionFailure() {
+ writeErr := errors.New("write failed")
+
+ writer, err := NewManifestWriter(
+ 2,
+ manifestFailingWriter{err: writeErr},
+ *UnpartitionedSpec,
+ testSchema,
+ snapshotID,
+ )
+ m.Require().ErrorIs(err, writeErr)
+ m.Require().NotNil(writer)
+
+ var closeErr error
+ m.NotPanics(func() {
+ closeErr = writer.Close()
+ })
+ m.ErrorIs(closeErr, ErrEmptyManifest)
+
+ m.Same(closeErr, writer.Close())
+}
+
+func (m *ManifestTestSuite) TestEmptyManifestWriterCloseJoinsUnderlyingError() {
+ for _, version := range []int{1, 2, 3} {
+ m.Run("v"+strconv.Itoa(version), func() {
+ writer, err := NewManifestWriter(
+ version,
+ io.Discard,
+ *UnpartitionedSpec,
+ testSchema,
+ snapshotID,
+ )
+ m.Require().NoError(err)
+
+ underlyingErr := errors.New("underlying close failed")
+ m.setManifestWriterCloseError(writer, underlyingErr)
+
+ firstErr := writer.Close()
+ m.Require().ErrorIs(firstErr, ErrEmptyManifest)
+ m.ErrorIs(firstErr, underlyingErr)
+ m.EqualError(
+ firstErr,
+ "empty manifest file has been written\nunderlying close failed",
+ )
+ m.Same(firstErr, writer.Close())
+ })
+ }
+}
+
+func (m *ManifestTestSuite) TestManifestWriterCloseAfterAddEntryFailure() {
+ entries := map[int]ManifestEntry{
+ 1: manifestEntryV1Records[0],
+ 2: manifestEntryV2Records[0],
+ 3: manifestEntryV3Records[0],
+ }
+ for _, version := range []int{1, 2, 3} {
+ m.Run("v"+strconv.Itoa(version), func() {
+ writer, err := NewManifestWriter(version, io.Discard, *UnpartitionedSpec, testSchema, snapshotID)
+ m.Require().NoError(err)
+ m.Require().NoError(writer.Add(entries[version]))
+
+ m.Require().Error(writer.addEntry(&manifestEntry{EntryStatus: ManifestEntryStatus(-1)}))
+
+ underlyingErr := errors.New("underlying close failed")
+ m.setManifestWriterCloseError(writer, underlyingErr)
+
+ firstErr := writer.Close()
+ m.Require().Same(underlyingErr, firstErr)
+ m.False(errors.Is(firstErr, ErrEmptyManifest))
+ m.Same(firstErr, writer.Close())
+ })
+ }
+}
+
+func (m *ManifestTestSuite) TestWriteManifestEmptyErrorIsNotDuplicated() {
+ for _, version := range []int{1, 2, 3} {
+ m.Run("v"+strconv.Itoa(version), func() {
+ var out bytes.Buffer
+
+ _, err := WriteManifest(
+ "manifest.avro",
+ &out,
+ version,
+ *UnpartitionedSpec,
+ testSchema,
+ snapshotID,
+ nil,
+ )
+ m.ErrorIs(err, ErrEmptyManifest)
+ m.Same(ErrEmptyManifest, err)
+ })
+ }
+
+ var out bytes.Buffer
+ _, _, err := WriteManifestV3(
+ "manifest.avro",
+ &out,
+ 0,
+ *UnpartitionedSpec,
+ testSchema,
+ snapshotID,
+ nil,
+ )
+ m.ErrorIs(err, ErrEmptyManifest)
+ m.Same(ErrEmptyManifest, err)
+}
+
func TestReadManifestDecodesNilLogicalPartitionValueFromNullableUnion(t *testing.T) {
schema := NewSchema(
0,
diff --git a/table/snapshot_producers.go b/table/snapshot_producers.go
index b305d3f4..324b3c3f 100644
--- a/table/snapshot_producers.go
+++ b/table/snapshot_producers.go
@@ -220,7 +220,12 @@ func (of *overwriteFiles) existingManifests(parent *Snapshot) ([]iceberg.Manifes
return nil, err
}
defer internal.CheckedClose(fileCloser, &retErr)
- defer internal.CheckedClose(wr, &retErr)
+ writerClosed := false
+ defer func() {
+ if !writerClosed {
+ internal.CheckedClose(wr, &retErr)
+ }
+ }()
for _, entry := range notDeleted {
if err := wr.Existing(entry); err != nil {
@@ -229,6 +234,7 @@ func (of *overwriteFiles) existingManifests(parent *Snapshot) ([]iceberg.Manifes
}
// close the writer to force a flush and ensure counter.Count is accurate
+ writerClosed = true
if err := wr.Close(); err != nil {
return nil, err
}
@@ -375,7 +381,12 @@ func (m *manifestMergeManager) createManifest(specID int, bin []iceberg.Manifest
return nil, err
}
defer internal.CheckedClose(fileCloser, &err)
- defer internal.CheckedClose(wr, &err)
+ writerClosed := false
+ defer func() {
+ if !writerClosed {
+ internal.CheckedClose(wr, &err)
+ }
+ }()
for _, manifest := range bin {
for entry, err := range m.snap.iterManifestEntries(manifest, false) {
@@ -402,6 +413,7 @@ func (m *manifestMergeManager) createManifest(specID int, bin []iceberg.Manifest
}
// close the writer to force a flush and ensure counter.Count is accurate
+ writerClosed = true
if err := wr.Close(); err != nil {
return nil, err
}
@@ -834,7 +846,12 @@ func (sp *snapshotProducer) parentDependentManifests(ctx context.Context, parent
if err != nil {
return nil, err
}
- defer internal.CheckedClose(wr, &retErr)
+ writerClosed := false
+ defer func() {
+ if !writerClosed {
+ internal.CheckedClose(wr, &retErr)
+ }
+ }()
for _, entry := range entries {
if err := wr.Delete(entry); err != nil {
@@ -842,6 +859,7 @@ func (sp *snapshotProducer) parentDependentManifests(ctx context.Context, parent
}
}
+ writerClosed = true
if err := wr.Close(); err != nil {
return nil, err
}
@@ -904,7 +922,12 @@ func (sp *snapshotProducer) writeAddedManifest(content iceberg.ManifestContent,
return nil, err
}
defer internal.CheckedClose(out, &retErr)
- defer internal.CheckedClose(wr, &retErr)
+ writerClosed := false
+ defer func() {
+ if !writerClosed {
+ internal.CheckedClose(wr, &retErr)
+ }
+ }()
for _, df := range files {
err := wr.Add(iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &sp.snapshotID,
@@ -915,6 +938,7 @@ func (sp *snapshotProducer) writeAddedManifest(content iceberg.ManifestContent,
}
// close the writer to force a flush and ensure counter.Count is accurate
+ writerClosed = true
if err := wr.Close(); err != nil {
return nil, err
}