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 bcd9137b4 feat(metrics): Instrument commits and emit CommitReport (
#1236) (#1596)
bcd9137b4 is described below
commit bcd9137b4ec4b63039988e666414d36f3aaaf32b
Author: Tanmay Rauth <[email protected]>
AuthorDate: Wed Aug 5 03:31:30 2026 -0700
feat(metrics): Instrument commits and emit CommitReport ( #1236) (#1596)
* feat(metrics): instrument commits and emit CommitReport (#1236)
Fifth PR of the metrics reporting framework. Emits a CommitReport from
the commit path on successful completion:
- the retry loop in doCommit now records the total duration and the
number of attempts;
- the committed snapshot's summary is mapped into a CommitReport, under
Java's commit-report metric names. iceberg-go's summary keys differ
from those names (e.g. added-files-size -> added-files-size-bytes,
deleted-data-files -> removed-data-files, added-position-deletes ->
added-positional-deletes); the mapping translates them so the wire
format matches Java. Metrics with no iceberg-go summary key (DVs,
manifest counts) are omitted rather than reported as zero.
The committed table now also preserves its reporter. With the default
no-op reporter this is a pure no-op for existing users.
* docs(metrics): note deferred CommitReport.Metadata population
Java's CommitReport carries a metadata map (engine/commit context)
that iceberg-go does not thread into the commit path yet. Mark it as
a TODO in buildCommitReport rather than emitting an empty map.
* fix(metrics): address commit report review feedback
Bracket TotalDuration to the commit loop, populate manifest metrics,
guard report emission against panicking reporters on both commit and
scan paths, skip emission on unresolved branch head, and make
attemptsUsed int64. Add retry-attempt and manifest test coverage.
---
table/commit_metrics.go | 158 ++++++++++++++++++++++++++++++++++++
table/commit_metrics_test.go | 138 ++++++++++++++++++++++++++++++++
table/commit_retry_test.go | 47 +++++++++++
table/scanner.go | 2 +-
table/table.go | 39 +++++++++
table/table_metrics_commit_test.go | 159 +++++++++++++++++++++++++++++++++++++
6 files changed, 542 insertions(+), 1 deletion(-)
diff --git a/table/commit_metrics.go b/table/commit_metrics.go
new file mode 100644
index 000000000..9dedbd1a3
--- /dev/null
+++ b/table/commit_metrics.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
+
+import (
+ "context"
+ "log"
+ "strconv"
+ "time"
+
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/metrics"
+)
+
+// safeReport forwards a report to rep, recovering from a panic in a
misbehaving
+// third-party reporter so it can never fail a commit that has already durably
+// succeeded. The Reporter contract requires a reporter never affect the
observed
+// operation; compositeReporter isolates each inner reporter this way, but a
bare
+// non-composite reporter gets no such protection at the emit site.
+func safeReport(ctx context.Context, rep metrics.Reporter, report
metrics.MetricsReport) {
+ defer func() {
+ if v := recover(); v != nil {
+ log.Printf("Warning: metrics reporter %T panicked;
recovered: %v", rep, v)
+ }
+ }()
+ rep.Report(ctx, report)
+}
+
+// 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) are absent and therefore omitted — exactly as Java's
+// counterFrom omits summary keys it does not find.
+//
+// TODO: the report's Metadata map is left unset. Java's CommitReport carries a
+// metadata map (e.g. engine name/version and other commit context); iceberg-go
+// does not thread that context into the commit path yet, so it is omitted
+// rather than reported empty. Populate it in a follow-up once the context is
+// available.
+func buildCommitReport(tableName string, snap *Snapshot, attempts int64, dur
time.Duration) metrics.CommitReport {
+ var (
+ snapshotID int64
+ seqNum int64
+ operation string
+ props iceberg.Properties
+ )
+ if snap != nil {
+ snapshotID = snap.SnapshotID
+ seqNum = snap.SequenceNumber
+ if snap.Summary != nil {
+ operation = string(snap.Summary.Operation)
+ props = snap.Summary.Properties
+ }
+ }
+
+ count := func(key string) *metrics.CounterResult { return
summaryCounter(props, key, metrics.UnitCount) }
+ bytesOf := func(key string) *metrics.CounterResult { return
summaryCounter(props, key, metrics.UnitBytes) }
+
+ return metrics.CommitReport{
+ TableName: tableName,
+ SnapshotID: snapshotID,
+ SequenceNumber: seqNum,
+ Operation: operation,
+ Metrics: metrics.CommitMetricsResult{
+ TotalDuration: metrics.NewNanosTimerResult(1,
dur.Nanoseconds()),
+ Attempts:
metrics.NewCounterResult(metrics.UnitCount, attempts),
+
+ AddedDataFiles: count(addedDataFilesKey),
+ RemovedDataFiles: count(deletedDataFilesKey),
+ TotalDataFiles: count(totalDataFilesKey),
+
+ AddedDeleteFiles: count(addedDeleteFilesKey),
+ RemovedDeleteFiles: count(removedDeleteFilesKey),
+ TotalDeleteFiles: count(totalDeleteFilesKey),
+
+ AddedEqualityDeleteFiles:
count(addedEqDeleteFilesKey),
+ RemovedEqualityDeleteFiles:
count(removedEqDeleteFilesKey),
+
+ AddedPositionalDeleteFiles:
count(addedPosDeleteFilesKey),
+ RemovedPositionalDeleteFiles:
count(removedPosDeleteFilesKey),
+
+ AddedRecords: count(addedRecordsKey),
+ RemovedRecords: count(deletedRecordsKey),
+ TotalRecords: count(totalRecordsKey),
+
+ AddedFilesSizeBytes: bytesOf(addedFileSizeKey),
+ RemovedFilesSizeBytes: bytesOf(removedFileSizeKey),
+ TotalFilesSizeBytes: bytesOf(totalFileSizeKey),
+
+ AddedPositionalDeletes: count(addedPosDeletesKey),
+ RemovedPositionalDeletes: count(removedPosDeletesKey),
+ TotalPositionalDeletes: count(totalPosDeletesKey),
+
+ AddedEqualityDeletes: count(addedEqDeletesKey),
+ RemovedEqualityDeletes: count(removedEqDeletesKey),
+ TotalEqualityDeletes: count(totalEqDeletesKey),
+
+ ManifestsCreated: count(manifestsCreatedKey),
+ ManifestsReplaced: count(manifestsReplacedKey),
+ ManifestsKept: count(manifestsKeptKey),
+ // entries-processed in the summary maps to
manifest-entries-processed.
+ ManifestEntriesProcessed: count(entriesProcessedKey),
+ },
+ }
+}
diff --git a/table/commit_metrics_test.go b/table/commit_metrics_test.go
new file mode 100644
index 000000000..922b2a7fa
--- /dev/null
+++ b/table/commit_metrics_test.go
@@ -0,0 +1,138 @@
+// 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",
+ manifestsCreatedKey: "6",
+ manifestsReplacedKey: "2",
+ manifestsKeptKey: "4",
+ entriesProcessedKey: "9",
+ },
+ },
+ }
+
+ 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)
+ require.NotNil(t, m.TotalDataFiles)
+ assert.Equal(t, int64(10), m.TotalDataFiles.Value)
+ require.NotNil(t, m.AddedRecords)
+ assert.Equal(t, int64(12345), m.AddedRecords.Value)
+ require.NotNil(t, m.TotalRecords)
+ assert.Equal(t, int64(20000), m.TotalRecords.Value)
+
+ // Name-translated mappings (iceberg-go summary key -> Java report
name).
+ require.NotNil(t, m.RemovedDataFiles, "deleted-data-files ->
removed-data-files")
+ assert.Equal(t, int64(1), m.RemovedDataFiles.Value)
+ require.NotNil(t, m.RemovedRecords, "deleted-records ->
removed-records")
+ assert.Equal(t, int64(5), m.RemovedRecords.Value)
+ require.NotNil(t, m.AddedFilesSizeBytes, "added-files-size ->
added-files-size-bytes")
+ assert.Equal(t, metrics.UnitBytes, m.AddedFilesSizeBytes.Unit)
+ assert.Equal(t, int64(4096000), m.AddedFilesSizeBytes.Value)
+ require.NotNil(t, m.RemovedFilesSizeBytes, "removed-files-size ->
removed-files-size-bytes")
+ assert.Equal(t, int64(100), m.RemovedFilesSizeBytes.Value)
+ require.NotNil(t, m.TotalFilesSizeBytes)
+ assert.Equal(t, int64(5000000), m.TotalFilesSizeBytes.Value)
+ require.NotNil(t, m.AddedPositionalDeletes, "added-position-deletes ->
added-positional-deletes")
+ assert.Equal(t, int64(2), m.AddedPositionalDeletes.Value)
+ require.NotNil(t, m.AddedPositionalDeleteFiles,
"added-position-delete-files -> added-positional-delete-files")
+ assert.Equal(t, int64(1), m.AddedPositionalDeleteFiles.Value)
+ require.NotNil(t, m.AddedEqualityDeletes)
+ assert.Equal(t, int64(3), m.AddedEqualityDeletes.Value)
+ require.NotNil(t, m.AddedEqualityDeleteFiles)
+ assert.Equal(t, int64(1), m.AddedEqualityDeleteFiles.Value)
+
+ // Manifest metrics: entries-processed maps to
manifest-entries-processed.
+ require.NotNil(t, m.ManifestsCreated)
+ assert.Equal(t, int64(6), m.ManifestsCreated.Value)
+ require.NotNil(t, m.ManifestsReplaced)
+ assert.Equal(t, int64(2), m.ManifestsReplaced.Value)
+ require.NotNil(t, m.ManifestsKept)
+ assert.Equal(t, int64(4), m.ManifestsKept.Value)
+ require.NotNil(t, m.ManifestEntriesProcessed, "entries-processed ->
manifest-entries-processed")
+ assert.Equal(t, int64(9), m.ManifestEntriesProcessed.Value)
+
+ // Keys absent from the summary are omitted, not zeroed; metrics with no
+ // iceberg-go summary key (DVs) stay unset.
+ assert.Nil(t, m.RemovedDeleteFiles)
+ assert.Nil(t, m.AddedDVs)
+}
+
+func TestBuildCommitReportNilSnapshot(t *testing.T) {
+ cr := buildCommitReport("db.tbl", nil, 1, time.Millisecond)
+
+ assert.Equal(t, "db.tbl", cr.TableName)
+ assert.Equal(t, int64(0), cr.SnapshotID)
+ assert.Empty(t, cr.Operation)
+ require.NotNil(t, cr.Metrics.Attempts)
+ assert.Equal(t, int64(1), cr.Metrics.Attempts.Value)
+ assert.Nil(t, cr.Metrics.AddedDataFiles)
+}
+
+func TestBuildCommitReportUnparseableValueOmitted(t *testing.T) {
+ snap := &Snapshot{
+ Summary: &Summary{
+ Operation: OpAppend,
+ Properties: iceberg.Properties{addedDataFilesKey:
"not-a-number"},
+ },
+ }
+ cr := buildCommitReport("t", snap, 1, time.Millisecond)
+ assert.Nil(t, cr.Metrics.AddedDataFiles, "unparseable summary value is
omitted")
+}
diff --git a/table/commit_retry_test.go b/table/commit_retry_test.go
index e698ed08b..a3c14e142 100644
--- a/table/commit_retry_test.go
+++ b/table/commit_retry_test.go
@@ -31,6 +31,7 @@ import (
"github.com/apache/iceberg-go"
iceio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/metrics"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -525,6 +526,52 @@ func TestDoCommit_OrphanCleanedOnSuccess(t *testing.T) {
"live committed manifest list must be preserved by the cleanup
defer")
}
+// TestDoCommit_CommitReportRecordsRetryAttempts drives one retryable
+// ErrCommitFailed before success and asserts the emitted CommitReport records
+// Attempts == 2. This exercises the attempt-counting on the emit path, which
no
+// other test does — every other commit test is a clean single-attempt success.
+func TestDoCommit_CommitReportRecordsRetryAttempts(t *testing.T) {
+ spec := iceberg.NewPartitionSpec()
+ wfs, meta := newMemIOWithRetryMeta(t, spec)
+
+ tbl := newOCCTable(t, meta, wfs, nil)
+ txn := tbl.NewTransaction()
+ sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+ sp.appendDataFile(newTestDataFile(t, spec,
"mem://default/table-location/data/f.parquet", nil))
+
+ updates, reqs, err := sp.commit(context.Background())
+ require.NoError(t, err)
+
+ // Fail once with ErrCommitFailed (one retry), then succeed on the 2nd
attempt.
+ cat := &sequentialCatalog{
+ metadata: meta,
+ errs: []error{ErrCommitFailed},
+ }
+ sink := &metrics.InMemoryReporter{}
+ tbl = New(
+ Identifier{"db", "commit-report-retry"},
+ meta,
+ "mem://default/table-location/metadata/v1.metadata.json",
+ func(context.Context) (iceio.IO, error) { return wfs, nil },
+ cat,
+ WithMetricsReporter(sink),
+ )
+
+ _, err = tbl.doCommit(t.Context(), updates, reqs,
withCommitBranch(MainBranch))
+ require.NoError(t, err, "doCommit must succeed on the second attempt")
+
+ var cr *metrics.CommitReport
+ for _, r := range sink.Reports() {
+ if c, ok := r.(metrics.CommitReport); ok {
+ cr = &c
+ }
+ }
+ require.NotNil(t, cr, "a successful commit must emit a CommitReport")
+ require.NotNil(t, cr.Metrics.Attempts)
+ assert.Equal(t, int64(2), cr.Metrics.Attempts.Value,
+ "one retry then success must record 2 attempts")
+}
+
// TestDoCommit_OrphanNotCleanedOnUnknownError verifies that manifest-list
// files are NOT removed when CommitTable returns an unknown
non-ErrCommitFailed
// error (5xx / gateway timeout). In that case the catalog may have silently
diff --git a/table/scanner.go b/table/scanner.go
index 28e5f4b4a..c48e8c457 100644
--- a/table/scanner.go
+++ b/table/scanner.go
@@ -864,7 +864,7 @@ func (scan *Scan) PlanFiles(ctx context.Context)
([]FileScanTask, error) {
// fields. Report assembly must never fail a scan, so a
projection error
// just yields a report that omits projected fields.
projected, _ := scan.Projection()
- rep.Report(ctx, scan.buildScanReport(&acc, schema,
projected, planningDuration))
+ safeReport(ctx, rep, scan.buildScanReport(&acc, schema,
projected, planningDuration))
}
}
diff --git a/table/table.go b/table/table.go
index eb1f01e16..83422d8e5 100644
--- a/table/table.go
+++ b/table/table.go
@@ -487,8 +487,14 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
newLoc string
timer *time.Timer
orphanedManifests []string // manifest-list files orphaned by
rebuilds
+ commitDuration time.Duration
)
+ // attemptsUsed initializes to 1: the emit block below is reached only
via the
+ // success break, which always runs at least one attempt. Deriving it
from a
+ // 0-based counter risks emitting 0 if a second success exit is ever
added.
+ var attemptsUsed int64 = 1
+
// cleanupOrphans controls whether the defer below removes orphaned
manifest-list
// files on exit. It defaults to true (clean on all safe exits) and is
set to
// false only for the one unsafe case: a non-ErrCommitFailed error from
@@ -515,6 +521,11 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
// numRetries counts retries; total attempts = 1 initial + numRetries.
totalAttempts := cfg.numRetries + 1
+ // commitStart brackets the commit loop itself (not FS resolution above
or
+ // metadata cleanup below) so TotalDuration times only the CommitTable
+ // submission loop, matching Java's CommitReport.
+ commitStart := time.Now()
+
for attempt := range totalAttempts {
if attempt != 0 {
wait := backoffDuration(attempt-1, cfg.minWaitMs,
cfg.maxWaitMs)
@@ -596,6 +607,11 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
newMeta, newLoc, err = t.cat.CommitTable(retryCtx,
slices.Clone(t.identifier), reqs, updates)
if err == nil {
+ attemptsUsed = int64(attempt) + 1
+ // Capture elapsed time at the commit boundary, before
orphan
+ // cleanup and deleteOldMetadata below (both can do
I/O).
+ commitDuration = time.Since(commitStart)
+
break
}
@@ -632,6 +648,29 @@ 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 != "" {
+ // A nil lookup means the branch head could not be
resolved (e.g. a
+ // fresh non-default branch); attributing
CurrentSnapshot() would
+ // carry the wrong snapshot, so skip emission entirely.
+ committed = newMeta.SnapshotByName(co.branch)
+ }
+ if committed != nil {
+ safeReport(ctx, rep,
+ buildCommitReport(strings.Join(t.identifier,
"."), committed, attemptsUsed, commitDuration))
+ }
+ }
+
return New(t.identifier, newMeta, newLoc, t.fsF, t.cat,
withReporterState(t.reporter, t.reporterSet)), nil
}
diff --git a/table/table_metrics_commit_test.go
b/table/table_metrics_commit_test.go
new file mode 100644
index 000000000..af9d31581
--- /dev/null
+++ b/table/table_metrics_commit_test.go
@@ -0,0 +1,159 @@
+// 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()
+ 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", "metadata_only_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}]`})
+ require.NoError(t, err)
+ defer arrTable.Release()
+
+ // Seed a snapshot so the property-only commit below has an existing
branch
+ // head that must NOT be reported.
+ tbl, err = tbl.AppendTable(ctx, arrTable, arrTable.NumRows(), nil)
+ require.NoError(t, err)
+
+ // Discard the append's report; only the property-only commit is under
test.
+ commitReportSink.Reset()
+
+ txn := tbl.NewTransaction()
+ require.NoError(t, txn.SetProperties(iceberg.Properties{"free-form":
"value"}))
+ _, err = txn.Commit(ctx)
+ require.NoError(t, err)
+
+ for _, r := range commitReportSink.Reports() {
+ _, ok := r.(metrics.CommitReport)
+ assert.False(t, ok, "metadata-only commit must not emit a
CommitReport")
+ }
+}
