laskoviymishka commented on code in PR #1596:
URL: https://github.com/apache/iceberg-go/pull/1596#discussion_r3705522676
##########
table/table.go:
##########
@@ -464,6 +464,9 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
return nil, err
}
+ commitStart := time.Now()
Review Comment:
`commitStart` is captured here, before `t.fsF(ctx)` and `requireWriteFileIO`
resolve the filesystem (both can do I/O), and `time.Since(commitStart)` is
sampled at the emit site after `deleteOldMetadata`, which can also hit the
network. So `TotalDuration` starts before the commit boundary and ends after
it: on a REST catalog with dynamic FS resolution plus old-metadata cleanup it's
systematically larger than Java, which times only the `CommitTable` submission
loop.
Since the whole point here is to mirror Java's CommitReport, an inflated
duration makes cross-implementation comparison misleading. I'd move
`commitStart` to just before the retry loop and capture the elapsed time right
after the `break` (before orphan cleanup and `deleteOldMetadata`), so the
window brackets the commit itself.
##########
table/commit_metrics.go:
##########
@@ -0,0 +1,136 @@
+// 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 (
+ "strconv"
+ "time"
+
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/metrics"
+)
+
+// commitAddedSnapshot reports whether the commit produced a new snapshot.
+// Metadata-only commits (property or schema changes) carry no
addSnapshotUpdate
+// and must not emit a commit report: they create no snapshot, so the branch
head
+// is unchanged and reporting it would attribute a prior snapshot's metrics to
+// this commit. This mirrors Java, whose CommitReport is emitted only from the
+// snapshot-producing path (SnapshotProducer.commit).
+func commitAddedSnapshot(updates []Update) bool {
+ for _, u := range updates {
+ if _, ok := u.(*addSnapshotUpdate); ok {
+ return true
+ }
+ }
+
+ return false
+}
+
+// summaryCounter reads a snapshot-summary property and returns it as a
+// CounterResult, or nil if the key is absent or unparseable. This mirrors
+// Java's CommitMetricsResult.counterFrom exactly: a metric is omitted rather
+// than reported as a zero unless the snapshot summary carries a parseable
value
+// for it. As a result iceberg-go and Java emit the same present/absent metric
+// set for an equivalent commit.
+func summaryCounter(props iceberg.Properties, key string, unit metrics.Unit)
*metrics.CounterResult {
+ v, ok := props[key]
+ if !ok {
+ return nil
+ }
+ n, err := strconv.ParseInt(v, 10, 64)
+ if err != nil {
+ return nil
+ }
+
+ return metrics.NewCounterResult(unit, n)
+}
+
+// buildCommitReport assembles a CommitReport from the committed snapshot's
+// summary, mirroring Java's CommitMetricsResult.from(commitMetrics,
+// snapshotSummary): attempts and total-duration come from the commit itself,
+// and every other metric is read from the snapshot summary under its spec key
+// and emitted under Java's commit-report field name so dashboards line up
+// across implementations. Metrics whose summary key iceberg-go does not yet
+// populate (DVs, manifest counts) are absent and therefore omitted — exactly
as
Review Comment:
This comment says iceberg-go doesn't yet populate manifest counts, but
`rewrite_manifests.go` already writes `manifests-created`,
`manifests-replaced`, `manifests-kept`, and `entries-processed` into the
snapshot summary on every non-trivial RewriteManifests commit. So a
CommitReport emitted after a RewriteManifests silently drops all four manifest
metrics that Java's `CommitMetricsResult.from` would populate, and
RewriteManifests is the one path that always produces them.
The four summary keys are package-level constants in this same package, so
we can populate them directly in the `CommitMetricsResult` below:
```go
ManifestsCreated: count(manifestsCreatedKey),
ManifestsReplaced: count(manifestsReplacedKey),
ManifestsKept: count(manifestsKeptKey),
ManifestEntriesProcessed: count(entriesProcessedKey),
```
One thing to watch on the wire mapping: `entries-processed` in the summary
emits under `manifest-entries-processed`, so the lookup uses
`entriesProcessedKey` but the field is `ManifestEntriesProcessed` (the fixture
already has this right). And drop the "does not yet populate ... manifest
counts" line from the comment.
##########
table/table.go:
##########
@@ -632,6 +637,28 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
deleteOldMetadata(fs, t.metadata, newMeta)
+ // Emit a commit report on success. Prefer the just-committed branch
head
+ // over the table's current snapshot so commits to a non-default branch
+ // report the snapshot they actually created.
+ //
+ // Mirrors the scan path: building the report is skipped for a no-op
+ // reporter (the opt-in default), since a nop discards it and
assembling one
+ // would be pure overhead. A metadata-only commit produces no snapshot
and
+ // must be skipped too — its branch head is unchanged, so reporting it
would
+ // attribute a prior snapshot's metrics to this commit.
+ if rep := t.MetricsReporter(); !metrics.IsNop(rep) &&
commitAddedSnapshot(updates) {
+ committed := newMeta.CurrentSnapshot()
+ if co.branch != "" {
+ if s := newMeta.SnapshotByName(co.branch); s != nil {
Review Comment:
If `co.branch` is set but `SnapshotByName` returns nil (e.g. the first
commit that creates a new non-default branch), `committed` silently stays as
`CurrentSnapshot()`, the default-branch head. That's a different snapshot than
the one we just committed, so the report would carry the wrong snapshot ID,
sequence number, and operation. This is exactly the case the "prefer branch
head" comment is meant to handle, so the silent fallback inverts the intent.
I'd either compare the resolved snapshot's ID against what was committed, or
skip emission when the branch lookup comes back nil rather than attributing the
wrong snapshot. A test that commits to a fresh branch would pin this down.
##########
table/table.go:
##########
@@ -632,6 +637,28 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
deleteOldMetadata(fs, t.metadata, newMeta)
+ // Emit a commit report on success. Prefer the just-committed branch
head
+ // over the table's current snapshot so commits to a non-default branch
+ // report the snapshot they actually created.
+ //
+ // Mirrors the scan path: building the report is skipped for a no-op
+ // reporter (the opt-in default), since a nop discards it and
assembling one
+ // would be pure overhead. A metadata-only commit produces no snapshot
and
+ // must be skipped too — its branch head is unchanged, so reporting it
would
+ // attribute a prior snapshot's metrics to this commit.
+ if rep := t.MetricsReporter(); !metrics.IsNop(rep) &&
commitAddedSnapshot(updates) {
+ committed := newMeta.CurrentSnapshot()
+ if co.branch != "" {
+ if s := newMeta.SnapshotByName(co.branch); s != nil {
+ committed = s
+ }
+ }
+ if committed != nil {
+ rep.Report(ctx,
Review Comment:
This `Report` call isn't guarded, and it runs after `CommitTable` has
already durably succeeded. The `Reporter` contract says a reporter must never
fail the operation it observes, but nothing structurally stops a buggy
third-party impl from panicking here, and if one does, the panic propagates out
of `doCommit` and the caller sees an error for a commit that actually landed.
That's the worst kind of confusion: a retry or rollback against
already-committed state.
`compositeReporter` recovers around each inner report, but a bare
non-composite reporter gets no protection. I'd wrap this in a small
`safeReport` helper that recovers and logs, mirroring the composite's
isolation. The scan path in `scanner.go` has the same gap, though that one's
pre-existing and can be a follow-up.
##########
table/table_metrics_commit_test.go:
##########
@@ -0,0 +1,158 @@
+// 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_test
+
+import (
+ "context"
+ "sync"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/catalog"
+ "github.com/apache/iceberg-go/catalog/sql"
+ "github.com/apache/iceberg-go/metrics"
+ "github.com/apache/iceberg-go/table"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+ "github.com/uptrace/bun/driver/sqliteshim"
+)
+
+// commitReportSink is registered once as a named reporter so an in-memory
+// catalog can select it via metrics-reporter-impl. Register panics on a
+// duplicate name, so registration is guarded by sync.Once.
+var (
+ commitReportSink = &metrics.InMemoryReporter{}
+ registerCommitSinkOnce sync.Once
+ commitReportReporterName = "test-commit-report-sink"
+)
+
+func registerCommitReportSink() {
+ registerCommitSinkOnce.Do(func() {
+ metrics.Register(commitReportReporterName,
func(map[string]string) (metrics.Reporter, error) {
+ return commitReportSink, nil
+ })
+ })
+}
+
+func TestCommitEmitsCommitReport(t *testing.T) {
+ registerCommitReportSink()
+ commitReportSink.Reset()
+
+ ctx := context.Background()
+ cat, err := catalog.Load(ctx, "default", iceberg.Properties{
+ "uri": ":memory:",
+ "type": "sql",
+ sql.DriverKey: sqliteshim.ShimName,
+ sql.DialectKey: string(sql.SQLite),
+ "warehouse": "file://" + t.TempDir(),
+ metrics.ReporterImplKey: commitReportReporterName,
+ })
+ require.NoError(t, err)
+
+ ident := table.Identifier{"default", "commit_report_tbl"}
+ require.NoError(t, cat.CreateNamespace(ctx,
catalog.NamespaceFromIdent(ident), nil))
+
+ sc := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ )
+ tbl, err := cat.CreateTable(ctx, ident, sc)
+ require.NoError(t, err)
+
+ arrowSchema, err := table.SchemaToArrowSchema(sc, nil, true, false)
+ require.NoError(t, err)
+ arrTable, err := array.TableFromJSON(memory.DefaultAllocator,
arrowSchema,
+ []string{`[{"id": 1}, {"id": 2}, {"id": 3}]`})
+ require.NoError(t, err)
+ defer arrTable.Release()
+
+ _, err = tbl.AppendTable(ctx, arrTable, arrTable.NumRows(), nil)
+ require.NoError(t, err)
+
+ var commits []metrics.CommitReport
+ for _, r := range commitReportSink.Reports() {
+ if cr, ok := r.(metrics.CommitReport); ok {
+ commits = append(commits, cr)
+ }
+ }
+ require.NotEmpty(t, commits, "an append commit must emit a
CommitReport")
+
+ cr := commits[len(commits)-1]
+ assert.Equal(t, "append", cr.Operation)
+ require.NotNil(t, cr.Metrics.Attempts)
+ assert.GreaterOrEqual(t, cr.Metrics.Attempts.Value, int64(1))
+ require.NotNil(t, cr.Metrics.TotalDuration)
+ require.NotNil(t, cr.Metrics.AddedDataFiles)
+ assert.Positive(t, cr.Metrics.AddedDataFiles.Value)
+ require.NotNil(t, cr.Metrics.AddedRecords)
+ assert.Equal(t, int64(3), cr.Metrics.AddedRecords.Value)
+}
+
+// TestMetadataOnlyCommitEmitsNoCommitReport pins that a commit which produces
no
+// snapshot (here a property-only change) emits no CommitReport, even though
the
+// table already has a snapshot. Reporting the unchanged branch head would
+// misattribute the earlier append's metrics to this commit.
+func TestMetadataOnlyCommitEmitsNoCommitReport(t *testing.T) {
+ registerCommitReportSink()
Review Comment:
`commitReportSink` is a package-level singleton shared with
`TestCommitEmitsCommitReport`, and this test never resets it at the top: it
only resets mid-test after the seeding append. Under sequential order it
happens to pass, but with `-shuffle=on`, `-count=2`, or a future committing
test running first, stale CommitReports would already be in the sink and this
assertion passes for the wrong reason. Worse, a real regression where the
metadata-only commit does emit would get masked.
I'd add `commitReportSink.Reset()` as the first statement here. Better
still, drop the singleton and inject a fresh `InMemoryReporter` per test via
`WithMetricsReporter` so there's no shared state to reason about at all.
##########
table/table.go:
##########
@@ -464,6 +464,9 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
return nil, err
}
+ commitStart := time.Now()
+ attemptsUsed := 0
Review Comment:
Two small things on `attemptsUsed`. It's declared `int` and later set with
`int(attempt) + 1`, but the `Attempts` counter is `int64` and `attempt` ranges
over a `uint` bounded by `maxRetryCount = math.MaxUint32`. On a 32-bit build a
large attempt index wraps negative through the `int` cast and emits a negative
counter. I'd declare it `int64`, use `int64(attempt) + 1`, and take `attempts
int64` in `buildCommitReport` to drop the cast.
Separately, initializing to `0` is a latent trap: today the budget-exhausted
path returns an error before the emit block so `0` never escapes, but nothing
structurally enforces that. If someone later adds a second success exit, a `0`
attempts counter would slip out. Initializing to `1` (or deriving from
`attempt` at the emit site) makes the invariant hold by construction.
##########
table/table_metrics_commit_test.go:
##########
@@ -0,0 +1,158 @@
+// 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_test
+
+import (
+ "context"
+ "sync"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/catalog"
+ "github.com/apache/iceberg-go/catalog/sql"
+ "github.com/apache/iceberg-go/metrics"
+ "github.com/apache/iceberg-go/table"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+ "github.com/uptrace/bun/driver/sqliteshim"
+)
+
+// commitReportSink is registered once as a named reporter so an in-memory
+// catalog can select it via metrics-reporter-impl. Register panics on a
+// duplicate name, so registration is guarded by sync.Once.
+var (
+ commitReportSink = &metrics.InMemoryReporter{}
+ registerCommitSinkOnce sync.Once
+ commitReportReporterName = "test-commit-report-sink"
+)
+
+func registerCommitReportSink() {
+ registerCommitSinkOnce.Do(func() {
+ metrics.Register(commitReportReporterName,
func(map[string]string) (metrics.Reporter, error) {
+ return commitReportSink, nil
+ })
+ })
+}
+
+func TestCommitEmitsCommitReport(t *testing.T) {
+ registerCommitReportSink()
+ commitReportSink.Reset()
+
+ ctx := context.Background()
+ cat, err := catalog.Load(ctx, "default", iceberg.Properties{
+ "uri": ":memory:",
+ "type": "sql",
+ sql.DriverKey: sqliteshim.ShimName,
+ sql.DialectKey: string(sql.SQLite),
+ "warehouse": "file://" + t.TempDir(),
+ metrics.ReporterImplKey: commitReportReporterName,
+ })
+ require.NoError(t, err)
+
+ ident := table.Identifier{"default", "commit_report_tbl"}
+ require.NoError(t, cat.CreateNamespace(ctx,
catalog.NamespaceFromIdent(ident), nil))
+
+ sc := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ )
+ tbl, err := cat.CreateTable(ctx, ident, sc)
+ require.NoError(t, err)
+
+ arrowSchema, err := table.SchemaToArrowSchema(sc, nil, true, false)
+ require.NoError(t, err)
+ arrTable, err := array.TableFromJSON(memory.DefaultAllocator,
arrowSchema,
+ []string{`[{"id": 1}, {"id": 2}, {"id": 3}]`})
+ require.NoError(t, err)
+ defer arrTable.Release()
+
+ _, err = tbl.AppendTable(ctx, arrTable, arrTable.NumRows(), nil)
+ require.NoError(t, err)
+
+ var commits []metrics.CommitReport
+ for _, r := range commitReportSink.Reports() {
+ if cr, ok := r.(metrics.CommitReport); ok {
+ commits = append(commits, cr)
+ }
+ }
+ require.NotEmpty(t, commits, "an append commit must emit a
CommitReport")
+
+ cr := commits[len(commits)-1]
+ assert.Equal(t, "append", cr.Operation)
+ require.NotNil(t, cr.Metrics.Attempts)
+ assert.GreaterOrEqual(t, cr.Metrics.Attempts.Value, int64(1))
Review Comment:
This only asserts `Attempts >= 1`, and nothing in the suite ever drives
`attemptsUsed` above 1; every test is a clean single-attempt success. The
`int(attempt) + 1` assignment is the most novel bit of the emit path, so the
coverage here is effectively vacuous for it.
I'd add a test that injects a one-time retryable `ErrCommitFailed` and
asserts `Attempts.Value == 2`, so the retry-count logic is actually exercised.
##########
table/commit_metrics_test.go:
##########
@@ -0,0 +1,118 @@
+// 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 (
+ "testing"
+ "time"
+
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/metrics"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestBuildCommitReport(t *testing.T) {
+ snap := &Snapshot{
+ SnapshotID: 42,
+ SequenceNumber: 3,
+ Summary: &Summary{
+ Operation: OpAppend,
+ Properties: iceberg.Properties{
+ addedDataFilesKey: "4",
+ deletedDataFilesKey: "1",
+ totalDataFilesKey: "10",
+ addedRecordsKey: "12345",
+ deletedRecordsKey: "5",
+ totalRecordsKey: "20000",
+ addedFileSizeKey: "4096000",
+ removedFileSizeKey: "100",
+ totalFileSizeKey: "5000000",
+ addedPosDeletesKey: "2",
+ addedPosDeleteFilesKey: "1",
+ addedEqDeletesKey: "3",
+ addedEqDeleteFilesKey: "1",
+ },
+ },
+ }
+
+ cr := buildCommitReport("db.tbl", snap, 2, 7*time.Millisecond)
+
+ assert.Equal(t, "db.tbl", cr.TableName)
+ assert.Equal(t, int64(42), cr.SnapshotID)
+ assert.Equal(t, int64(3), cr.SequenceNumber)
+ assert.Equal(t, "append", cr.Operation)
+
+ m := cr.Metrics
+ require.NotNil(t, m.TotalDuration)
+ assert.Equal(t, (7 * time.Millisecond).Nanoseconds(),
m.TotalDuration.TotalDuration)
+ require.NotNil(t, m.Attempts)
+ assert.Equal(t, int64(2), m.Attempts.Value)
+
+ // Direct (same-name) mappings.
+ require.NotNil(t, m.AddedDataFiles)
+ assert.Equal(t, int64(4), m.AddedDataFiles.Value)
+ assert.Equal(t, int64(10), m.TotalDataFiles.Value)
+ assert.Equal(t, int64(12345), m.AddedRecords.Value)
Review Comment:
These `.Value` dereferences aren't nil-guarded: `m.TotalDataFiles`,
`m.AddedRecords`, `m.TotalRecords`, and the ones below (`m.TotalFilesSizeBytes`
/ `m.AddedPositionalDeletes` / `m.AddedEqualityDeletes` /
`m.AddedEqualityDeleteFiles`). The test guards `AddedDataFiles` and
`RemovedDataFiles` with `require.NotNil` first but drops the pattern for the
rest. If `buildCommitReport` ever returns nil for one of these (key rename,
ParseInt failure, refactor), the test panics with a nil deref instead of a
clean FAIL. I'd add `require.NotNil` before each.
While we're here: the summary seeds `removedFileSizeKey: "100"` but nothing
asserts `RemovedFilesSizeBytes` maps through, so that translation isn't
actually covered. Worth an assert, since round-trip coverage on every populated
field is the point of this test.
--
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]
