zeroshade commented on code in PR #1673:
URL: https://github.com/apache/iceberg-go/pull/1673#discussion_r3732159221
##########
table/snapshot_producers_test.go:
##########
@@ -719,6 +732,165 @@ func (t *trackingIO) GetWriterCount() int {
return len(t.writers)
}
+type closeErrorIO struct {
+ *trackingIO
+ closeErr error
+ removeErr error
+ removes []string
+}
+
+func newCloseErrorIO(closeErr, removeErr error) *closeErrorIO {
+ return &closeErrorIO{
+ trackingIO: newTrackingIO(),
+ closeErr: closeErr,
+ removeErr: removeErr,
+ }
+}
+
+func (c *closeErrorIO) Create(name string) (iceio.FileWriter, error) {
+ writer, err := c.trackingIO.Create(name)
+ if err == nil {
+ writer.(*trackingWriteCloser).closeErr = c.closeErr
+ }
+
+ return writer, err
+}
+
+func (c *closeErrorIO) Remove(name string) error {
+ c.removes = append(c.removes, name)
+ if c.removeErr != nil {
+ return c.removeErr
+ }
+
+ return c.trackingIO.Remove(name)
+}
+
+func TestCommitManifestsCloseFailureReturnsNoUpdates(t *testing.T) {
+ closeErr := errors.New("manifest list close failed")
+ removeErr := errors.New("manifest list cleanup failed")
+
+ for _, version := range []int{1, 2, 3} {
+ t.Run(strconv.Itoa(version), func(t *testing.T) {
+ fs := newCloseErrorIO(closeErr, removeErr)
+ spec := iceberg.NewPartitionSpec()
+ txn := createTestTransaction(t, fs, spec)
+ txn.meta.formatVersion = version
+ sp := newFastAppendFilesProducer(OpAppend, txn, fs,
nil, nil)
+ manifestVersion := version
+ if manifestVersion == 3 {
+ manifestVersion = 2
+ }
+ manifest := writeTestManifestFileWithVersion(
+ t, fs, spec, simpleSchema(), sp.snapshotID, 1,
manifestVersion,
+ )
+
+ updates, requirements, err :=
sp.commitManifests([]iceberg.ManifestFile{manifest}, nil)
+ require.ErrorIs(t, err, closeErr)
+ require.ErrorIs(t, err, removeErr)
+ require.Nil(t, updates)
+ require.Nil(t, requirements)
+ require.Len(t, fs.removes, 1)
+
+ if version == 3 {
+ fs.writersMu.Lock()
+ writerCount := len(fs.writers)
+ var output []byte
+ for _, writer := range fs.writers {
+ output = append([]byte(nil),
writer.buf.Bytes()...)
+ }
+ fs.writersMu.Unlock()
+
+ require.Equal(t, 1, writerCount)
+ writtenManifests, readErr :=
iceberg.ReadManifestList(bytes.NewReader(output))
+ require.NoError(t, readErr)
+ require.Len(t, writtenManifests, 1,
"manifest-list writer must flush before the output is closed")
Review Comment:
This assertion proves the flush happened, which is the right thing to check
— but it can't distinguish "flushed before the underlying close" from "flushed
at all", since a buffer that keeps accepting writes after `Close()` satisfies
it either way. To pin the ordering strictly, consider having
`trackingWriteCloser.Write` return an error once `closed` is set, or recording
an ordered event log across both layers and asserting inner-before-outer.
Non-blocking — the defers are in the right order today; this would just keep
them that way.
##########
table/snapshot_producers_test.go:
##########
@@ -719,6 +732,165 @@ func (t *trackingIO) GetWriterCount() int {
return len(t.writers)
}
+type closeErrorIO struct {
+ *trackingIO
+ closeErr error
+ removeErr error
+ removes []string
+}
+
+func newCloseErrorIO(closeErr, removeErr error) *closeErrorIO {
+ return &closeErrorIO{
+ trackingIO: newTrackingIO(),
+ closeErr: closeErr,
+ removeErr: removeErr,
+ }
+}
+
+func (c *closeErrorIO) Create(name string) (iceio.FileWriter, error) {
+ writer, err := c.trackingIO.Create(name)
+ if err == nil {
+ writer.(*trackingWriteCloser).closeErr = c.closeErr
+ }
+
+ return writer, err
+}
+
+func (c *closeErrorIO) Remove(name string) error {
+ c.removes = append(c.removes, name)
+ if c.removeErr != nil {
+ return c.removeErr
+ }
+
+ return c.trackingIO.Remove(name)
+}
+
+func TestCommitManifestsCloseFailureReturnsNoUpdates(t *testing.T) {
+ closeErr := errors.New("manifest list close failed")
+ removeErr := errors.New("manifest list cleanup failed")
+
+ for _, version := range []int{1, 2, 3} {
+ t.Run(strconv.Itoa(version), func(t *testing.T) {
+ fs := newCloseErrorIO(closeErr, removeErr)
+ spec := iceberg.NewPartitionSpec()
+ txn := createTestTransaction(t, fs, spec)
+ txn.meta.formatVersion = version
+ sp := newFastAppendFilesProducer(OpAppend, txn, fs,
nil, nil)
+ manifestVersion := version
+ if manifestVersion == 3 {
+ manifestVersion = 2
+ }
+ manifest := writeTestManifestFileWithVersion(
+ t, fs, spec, simpleSchema(), sp.snapshotID, 1,
manifestVersion,
+ )
+
+ updates, requirements, err :=
sp.commitManifests([]iceberg.ManifestFile{manifest}, nil)
+ require.ErrorIs(t, err, closeErr)
+ require.ErrorIs(t, err, removeErr)
+ require.Nil(t, updates)
+ require.Nil(t, requirements)
+ require.Len(t, fs.removes, 1)
Review Comment:
This confirms a removal was attempted. The complementary case — that a
*successful* cleanup leaves no partial manifest-list object behind — isn't
covered anywhere. Consider a variant with `removeErr == nil` that asserts the
path is actually gone from the backing store. Suggestion only.
##########
table/snapshot_producers_test.go:
##########
@@ -719,6 +732,165 @@ func (t *trackingIO) GetWriterCount() int {
return len(t.writers)
}
+type closeErrorIO struct {
+ *trackingIO
+ closeErr error
+ removeErr error
+ removes []string
+}
+
+func newCloseErrorIO(closeErr, removeErr error) *closeErrorIO {
+ return &closeErrorIO{
+ trackingIO: newTrackingIO(),
+ closeErr: closeErr,
+ removeErr: removeErr,
+ }
+}
+
+func (c *closeErrorIO) Create(name string) (iceio.FileWriter, error) {
+ writer, err := c.trackingIO.Create(name)
+ if err == nil {
+ writer.(*trackingWriteCloser).closeErr = c.closeErr
+ }
+
+ return writer, err
+}
+
+func (c *closeErrorIO) Remove(name string) error {
+ c.removes = append(c.removes, name)
+ if c.removeErr != nil {
+ return c.removeErr
+ }
+
+ return c.trackingIO.Remove(name)
+}
+
+func TestCommitManifestsCloseFailureReturnsNoUpdates(t *testing.T) {
Review Comment:
Worth adding while you're in here: an assertion that each writer layer is
closed exactly once. `trackingWriteCloser` already tracks `closed` as a bool;
making it a counter would let this test catch a double-close regression for
free. Optional.
##########
table/snapshot_producers_test.go:
##########
@@ -719,6 +732,165 @@ func (t *trackingIO) GetWriterCount() int {
return len(t.writers)
}
+type closeErrorIO struct {
+ *trackingIO
+ closeErr error
+ removeErr error
+ removes []string
+}
+
+func newCloseErrorIO(closeErr, removeErr error) *closeErrorIO {
+ return &closeErrorIO{
+ trackingIO: newTrackingIO(),
+ closeErr: closeErr,
+ removeErr: removeErr,
+ }
+}
+
+func (c *closeErrorIO) Create(name string) (iceio.FileWriter, error) {
+ writer, err := c.trackingIO.Create(name)
+ if err == nil {
+ writer.(*trackingWriteCloser).closeErr = c.closeErr
Review Comment:
Non-blocking observation about what this injects: the failure lands on the
*underlying* output writer only. The scenario this fix specifically exists for
— a failure surfacing during the OCF close-time flush while the outer `Close()`
succeeds — isn't reachable through this helper, so it currently goes untested.
Worth a follow-up variant that fails inside the manifest-list writer's own
close, which would exercise the join from the inner layer.
--
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]
