laskoviymishka commented on code in PR #1256:
URL: https://github.com/apache/iceberg-go/pull/1256#discussion_r3467411854
##########
table/transaction.go:
##########
@@ -510,8 +514,13 @@ func (t *Transaction) ReplaceDataFiles(ctx
context.Context, filesToDelete, files
}
}
+ wfs, err := requireWriteFileIO(fs)
Review Comment:
If `ensureNameMapping`/`SetProperties(DefaultNameMappingKey)` runs above
this guard in `ReplaceDataFiles` (and the same shape in
`ReplaceDataFilesWithDataFiles`, `ReplaceFiles`, `AddDataFiles`, `AddFiles`),
then on a read-only FS we've already mutated the transaction's in-memory
name-mapping before bailing out — the txn is left half-applied.
`performCopyOnWriteDeletion` got this right by guarding immediately after
`fsF`, before any name-mapping write. I'd move `requireWriteFileIO` up to right
after `t.tbl.fsF(ctx)` in each of these so the gate runs before any state
mutation. Worth a one-line comment on `requireWriteFileIO` noting the canonical
placement so a new call site doesn't reintroduce this.
##########
table/table.go:
##########
@@ -54,12 +54,19 @@ import (
// issue #830).
var ErrCommitFailed = errors.New("commit failed, refresh and try again")
-// ErrWriteIORequired is returned by doCommit when the table's file system
-// does not implement io.WriteFileIO. Manifest-list rebuild on retry requires
-// write access; failing fast here is preferable to silently skipping the
-// rebuild and reintroducing the stale-parent data-loss bug. Callers that
-// need to detect this condition should use errors.Is(err, ErrWriteIORequired).
-var ErrWriteIORequired = errors.New("commit: file system does not implement
WriteFileIO")
+// ErrWriteIORequired is returned by write paths when the table's file system
+// does not implement io.WriteFileIO. Callers that need to detect this
+// condition should use errors.Is(err, ErrWriteIORequired).
+var ErrWriteIORequired = errors.New("file system does not implement
WriteFileIO")
Review Comment:
Generalizing this to "write paths" makes sense, but it drops the
manifest-list rebuild rationale entirely — and this doc is the only place a
consumer reading `go doc`/IDE hover learns *why* `doCommit` fails fast instead
of skipping the rebuild and reintroducing the stale-parent data-loss bug
(#830). The inline comment at the `doCommit` site has it, but the exported
symbol is what people actually read.
I'd keep a sentence of that rationale here while generalizing the rest.
Separately, the sentinel string also changed (`"commit: file system..."` →
`"file system..."`); `ErrWriteIORequired` is new as of the prior PR so the
window is small, but worth a line in the PR description for anyone matching on
the message.
##########
table/transaction.go:
##########
@@ -1224,10 +1251,14 @@ func (t *Transaction) Overwrite(ctx context.Context,
rdr array.RecordReader, sna
if err != nil {
return err
}
+ wfs, err := requireWriteFileIO(fs)
Review Comment:
`Overwrite` already calls `performCopyOnWriteDeletion` above this, which now
guards immediately after its own `fsF`. So on a read-only FS we return from
there long before reaching this second guard — it's dead for the read-only
case, and we've done two `fsF(ctx)` round-trips.
I'd have `performCopyOnWriteDeletion` return the `WriteFileIO` it already
resolved and thread it down here, so there's one fetch and one guard. If
there's a reason the second fetch has to be independent (credential refresh,
etc.) let's drop a comment, since otherwise it reads as redundant.
##########
table/write_records.go:
##########
@@ -178,9 +177,9 @@ func WriteRecords(ctx context.Context, tbl *Table,
return internal.SingleErrorIter[iceberg.DataFile](err)
}
- writeFS, ok := fs.(iceio.WriteFileIO)
- if !ok {
- return
internal.SingleErrorIter[iceberg.DataFile](fmt.Errorf("%w: filesystem does not
support writing", iceberg.ErrNotImplemented))
+ writeFS, err := requireWriteFileIO(fs)
Review Comment:
This drops the `iceberg.ErrNotImplemented` wrapping. `WriteRecords` is
exported, so anyone matching on `errors.Is(err, iceberg.ErrNotImplemented)`
silently stops matching after this — a quiet breaking change.
I'd have `ErrWriteIORequired` wrap `ErrNotImplemented` so both `errors.Is`
checks keep passing, e.g. define it as `fmt.Errorf("%w: file system does not
implement WriteFileIO", ErrNotImplemented)`, or wrap here: `fmt.Errorf("%w:
%w", iceberg.ErrNotImplemented, ErrWriteIORequired)`. `ErrWriteIORequired` is
the more accurate sentinel, so I lean toward the former and keeping the
relationship. wdyt?
##########
table/write_file_io_test.go:
##########
@@ -0,0 +1,147 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+ "context"
+ "strings"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/stretchr/testify/require"
+)
+
+func newReadOnlyWriteTable(t *testing.T, overrides ...iceberg.Properties)
*Table {
+ t.Helper()
+
+ props := iceberg.Properties{PropertyFormatVersion: "2"}
+ for _, override := range overrides {
+ for k, v := range override {
+ props[k] = v
+ }
+ }
+
+ schema := simpleSchema()
+ meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec,
UnsortedSortOrder, "file:///tmp/read-only-write",
+ props)
+ require.NoError(t, err)
+
+ cat := &flakyCatalog{metadata: meta}
+
+ return New(
+ Identifier{"db", "read-only-write"},
+ meta,
+ "file:///tmp/read-only-write/metadata/v1.metadata.json",
+ func(context.Context) (iceio.IO, error) { return readOnlyIO{},
nil },
+ cat,
+ )
+}
+
+func newSingleRowReader(t *testing.T, schema *iceberg.Schema)
array.RecordReader {
+ t.Helper()
+
+ arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false)
+ require.NoError(t, err)
+
+ rec, _, err := array.RecordFromJSON(memory.DefaultAllocator,
arrowSchema, strings.NewReader(`[{"id": 1}]`))
+ require.NoError(t, err)
+ defer rec.Release()
+
+ tbl := array.NewTableFromRecords(arrowSchema, []arrow.RecordBatch{rec})
+ rdr := array.NewTableReader(tbl, -1)
+ t.Cleanup(func() {
+ rdr.Release()
+ tbl.Release()
+ })
+
+ return rdr
+}
+
+func newWriteIORequiredDataFile(t *testing.T) iceberg.DataFile {
+ t.Helper()
+
+ builder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec,
+ iceberg.EntryContentData,
+ "file:///tmp/read-only-write/data.parquet",
+ iceberg.ParquetFile,
+ nil,
+ nil,
+ nil,
+ 1,
+ 1,
+ )
+ require.NoError(t, err)
+
+ return builder.Build()
+}
+
+func TestWritePathsReturnErrorForReadOnlyIO(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ props iceberg.Properties
+ run func(context.Context, *Table) error
+ }{
+ {
+ name: "append",
+ run: func(ctx context.Context, tbl *Table) error {
+ return tbl.NewTransaction().Append(ctx,
newSingleRowReader(t, tbl.Schema()), nil)
+ },
+ },
+ {
+ name: "overwrite",
+ run: func(ctx context.Context, tbl *Table) error {
+ return tbl.NewTransaction().Overwrite(ctx,
newSingleRowReader(t, tbl.Schema()), nil)
+ },
+ },
+ {
+ name: "copy-on-write delete",
+ run: func(ctx context.Context, tbl *Table) error {
+ return tbl.NewTransaction().Delete(ctx,
iceberg.AlwaysTrue{}, nil)
+ },
+ },
+ {
+ name: "merge-on-read delete",
+ props: iceberg.Properties{WriteDeleteModeKey:
WriteModeMergeOnRead},
+ run: func(ctx context.Context, tbl *Table) error {
+ return tbl.NewTransaction().Delete(ctx,
iceberg.AlwaysTrue{}, nil)
+ },
+ },
+ {
+ name: "row delta",
+ run: func(ctx context.Context, tbl *Table) error {
+ return tbl.NewTransaction().NewRowDelta(nil).
+ AddRows(newWriteIORequiredDataFile(t)).
+ Commit(ctx)
+ },
+ },
Review Comment:
This covers append, overwrite, CoW delete, MoR delete, and row delta — but
the PR patches ~11 write paths. `AddDataFiles`, `AddFiles`,
`WriteEqualityDeletes`, `ReplaceDataFiles`, `ReplaceDataFilesWithDataFiles`,
and `ReplaceFiles` all gained guards and none are exercised here.
`AddDataFiles` and `ReplaceDataFilesWithDataFiles` are cheap to add since
`newWriteIORequiredDataFile` already exists. I'd add at least those plus
`WriteEqualityDeletes`. Also worth noting: `rewriteSingleFile` and
`writePositionDeletesForFiles` only run when the table has existing data
partially matching the filter, so the empty-table cases never reach them —
those are exactly the spots that would have panicked on the old unsafe
assertion, so a case seeded with one data file would be the most valuable
addition.
##########
table/write_file_io_test.go:
##########
@@ -0,0 +1,147 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+ "context"
+ "strings"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/stretchr/testify/require"
+)
+
+func newReadOnlyWriteTable(t *testing.T, overrides ...iceberg.Properties)
*Table {
+ t.Helper()
+
+ props := iceberg.Properties{PropertyFormatVersion: "2"}
+ for _, override := range overrides {
+ for k, v := range override {
+ props[k] = v
+ }
+ }
+
+ schema := simpleSchema()
+ meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec,
UnsortedSortOrder, "file:///tmp/read-only-write",
+ props)
+ require.NoError(t, err)
+
+ cat := &flakyCatalog{metadata: meta}
+
+ return New(
+ Identifier{"db", "read-only-write"},
+ meta,
+ "file:///tmp/read-only-write/metadata/v1.metadata.json",
+ func(context.Context) (iceio.IO, error) { return readOnlyIO{},
nil },
+ cat,
+ )
+}
+
+func newSingleRowReader(t *testing.T, schema *iceberg.Schema)
array.RecordReader {
+ t.Helper()
+
+ arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false)
+ require.NoError(t, err)
+
+ rec, _, err := array.RecordFromJSON(memory.DefaultAllocator,
arrowSchema, strings.NewReader(`[{"id": 1}]`))
+ require.NoError(t, err)
+ defer rec.Release()
+
+ tbl := array.NewTableFromRecords(arrowSchema, []arrow.RecordBatch{rec})
+ rdr := array.NewTableReader(tbl, -1)
+ t.Cleanup(func() {
+ rdr.Release()
+ tbl.Release()
+ })
+
+ return rdr
+}
+
+func newWriteIORequiredDataFile(t *testing.T) iceberg.DataFile {
+ t.Helper()
+
+ builder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec,
+ iceberg.EntryContentData,
+ "file:///tmp/read-only-write/data.parquet",
+ iceberg.ParquetFile,
+ nil,
+ nil,
+ nil,
+ 1,
+ 1,
+ )
+ require.NoError(t, err)
+
+ return builder.Build()
+}
+
+func TestWritePathsReturnErrorForReadOnlyIO(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ props iceberg.Properties
+ run func(context.Context, *Table) error
+ }{
+ {
+ name: "append",
+ run: func(ctx context.Context, tbl *Table) error {
+ return tbl.NewTransaction().Append(ctx,
newSingleRowReader(t, tbl.Schema()), nil)
+ },
+ },
+ {
+ name: "overwrite",
+ run: func(ctx context.Context, tbl *Table) error {
+ return tbl.NewTransaction().Overwrite(ctx,
newSingleRowReader(t, tbl.Schema()), nil)
+ },
+ },
+ {
+ name: "copy-on-write delete",
+ run: func(ctx context.Context, tbl *Table) error {
+ return tbl.NewTransaction().Delete(ctx,
iceberg.AlwaysTrue{}, nil)
+ },
+ },
+ {
+ name: "merge-on-read delete",
+ props: iceberg.Properties{WriteDeleteModeKey:
WriteModeMergeOnRead},
+ run: func(ctx context.Context, tbl *Table) error {
+ return tbl.NewTransaction().Delete(ctx,
iceberg.AlwaysTrue{}, nil)
+ },
+ },
+ {
+ name: "row delta",
+ run: func(ctx context.Context, tbl *Table) error {
+ return tbl.NewTransaction().NewRowDelta(nil).
+ AddRows(newWriteIORequiredDataFile(t)).
+ Commit(ctx)
+ },
+ },
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ tbl := newReadOnlyWriteTable(t, tc.props)
+ var err error
+ require.NotPanics(t, func() {
Review Comment:
The `NotPanics` wrapper is load-bearing here — it's guarding against the
pre-fix `fs.(io.WriteFileIO)` panic — but that's not obvious to a future
reader. A one-line comment would make it clear this isn't just ceremony, e.g.
`// NotPanics guards against the pre-fix unsafe type assertion on a read-only
FS.`
##########
table/table.go:
##########
@@ -54,12 +54,19 @@ import (
// issue #830).
var ErrCommitFailed = errors.New("commit failed, refresh and try again")
-// ErrWriteIORequired is returned by doCommit when the table's file system
-// does not implement io.WriteFileIO. Manifest-list rebuild on retry requires
-// write access; failing fast here is preferable to silently skipping the
-// rebuild and reintroducing the stale-parent data-loss bug. Callers that
-// need to detect this condition should use errors.Is(err, ErrWriteIORequired).
-var ErrWriteIORequired = errors.New("commit: file system does not implement
WriteFileIO")
+// ErrWriteIORequired is returned by write paths when the table's file system
+// does not implement io.WriteFileIO. Callers that need to detect this
+// condition should use errors.Is(err, ErrWriteIORequired).
+var ErrWriteIORequired = errors.New("file system does not implement
WriteFileIO")
+
+func requireWriteFileIO(fs icebergio.IO) (icebergio.WriteFileIO, error) {
Review Comment:
Minor, while we're here: `requireWriteFileIO` lives in `table.go` but is now
used across every write-path file. Not wrong, but a small `write_helpers.go`
(or a comment here noting the canonical "call immediately after `fsF`"
placement, tied to the guard-ordering point) would make the shared contract
easier to find. Non-blocking.
--
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]
