Use more robust Iceberg CDC metrics

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitterMetrics.java

@@ -33,12 +33,14 @@
 final class CommitterMetrics implements Serializable {
 
   final Counter snapshotsCreated = Metrics.counter(CommitDeltas.class, "snapshotsCreated");
-  final Counter committedDataFiles = Metrics.counter(CommitDeltas.class, "committedDataFiles");
-  final Counter committedDeleteFiles = Metrics.counter(CommitDeltas.class, "committedDeleteFiles");
-  final Counter committedRecords = Metrics.counter(CommitDeltas.class, "committedRecords");
-  final Counter committedEqualityDeleteRecords =
-      Metrics.counter(CommitDeltas.class, "committedEqualityDeleteRecords");
-  final Counter committedBytes = Metrics.counter(CommitDeltas.class, "committedBytes");
+  final Distribution committedDataFileRecordCount =
+      Metrics.distribution(CommitDeltas.class, "committedDataFileRecordCount");
+  final Distribution committedDataFileByteSize =
+      Metrics.distribution(CommitDeltas.class, "committedDataFileByteSize");
+  final Distribution committedEqualityDeleteRecordCount =
+      Metrics.distribution(CommitDeltas.class, "committedEqualityDeleteRecordCount");
+  final Distribution committedEqualityDeleteByteSize =
+      Metrics.distribution(CommitDeltas.class, "committedEqualityDeleteByteSize");
   final Distribution commitDurationMs =
       Metrics.distribution(CommitDeltas.class, "commitDurationMs");
   final Counter commitFailures = Metrics.counter(CommitDeltas.class, "commitFailures");
@@ -57,79 +59,58 @@ final class CommitterMetrics implements Serializable {
 
   /** Records a successful commit's volume metrics. */
   void recordCommit(CommitSummary summary) {
-    committedDataFiles.inc(summary.dataFileCount);
-    committedDeleteFiles.inc(summary.deleteFileCount);
-    committedRecords.inc(summary.dataRecords);
-    committedEqualityDeleteRecords.inc(summary.equalityDeleteRecords);
-    committedBytes.inc(summary.bytes);
+    for (DataFile f : summary.dataFiles) {
+      committedDataFileRecordCount.update(f.recordCount());
+      committedDataFileByteSize.update(f.fileSizeInBytes());
+    }
+    for (DeleteFile f : summary.deleteFiles) {
+      if (f.content() == FileContent.EQUALITY_DELETES) {
+        committedEqualityDeleteRecordCount.update(f.recordCount());
+        committedEqualityDeleteByteSize.update(f.fileSizeInBytes());
+      }
+    }
     snapshotsCreated.inc();
   }
 
-  /** One commit's volume and partition-spec summary, computed from its files. */
+  /** One commit's files and partition-spec summary. */
   static final class CommitSummary {
-    final long dataFileCount;
-    final long deleteFileCount;
-    final long dataRecords;
-    final long equalityDeleteRecords;
-    final long bytes;
+    final List<DataFile> dataFiles;
+    final List<DeleteFile> deleteFiles;
     final @Nullable Integer firstSpecId;
     final Set<Integer> specIds;
     final boolean hasEqualityDeletes;
 
     private CommitSummary(
-        long dataFileCount,
-        long deleteFileCount,
-        long dataRecords,
-        long equalityDeleteRecords,
-        long bytes,
+        List<DataFile> dataFiles,
+        List<DeleteFile> deleteFiles,
         @Nullable Integer firstSpecId,
         Set<Integer> specIds,
         boolean hasEqualityDeletes) {
-      this.dataFileCount = dataFileCount;
-      this.deleteFileCount = deleteFileCount;
-      this.dataRecords = dataRecords;
-      this.equalityDeleteRecords = equalityDeleteRecords;
-      this.bytes = bytes;
+      this.dataFiles = dataFiles;
+      this.deleteFiles = deleteFiles;
       this.firstSpecId = firstSpecId;
       this.specIds = specIds;
       this.hasEqualityDeletes = hasEqualityDeletes;
     }
 
     static CommitSummary of(List<DataFile> dataFiles, List<DeleteFile> deleteFiles) {
-      long dataRecords = 0;
-      long bytes = 0;
       @Nullable Integer firstSpecId = null;
       Set<Integer> specIds = new HashSet<>();
       for (DataFile f : dataFiles) {
-        dataRecords += f.recordCount();
-        bytes += f.fileSizeInBytes();
         if (firstSpecId == null) {
           firstSpecId = f.specId();
         }
         specIds.add(f.specId());
       }
-      long equalityDeleteRecords = 0;
       boolean hasEqualityDeletes = false;
       for (DeleteFile f : deleteFiles) {
-        bytes += f.fileSizeInBytes();
         if (firstSpecId == null) {
           firstSpecId = f.specId();
         }
         specIds.add(f.specId());
-        if (f.content() == FileContent.EQUALITY_DELETES) {
-          equalityDeleteRecords += f.recordCount();
-          hasEqualityDeletes = true;
-        }
+        hasEqualityDeletes |= f.content() == FileContent.EQUALITY_DELETES;
       }
-      return new CommitSummary(
-          dataFiles.size(),
-          deleteFiles.size(),
-          dataRecords,
-          equalityDeleteRecords,
-          bytes,
-          firstSpecId,
-          specIds,
-          hasEqualityDeletes);
+      return new CommitSummary(dataFiles, deleteFiles, firstSpecId, specIds, hasEqualityDeletes);
     }
   }
 }

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/package-info.java

@@ -41,12 +41,12 @@
  *   <li><b>snapshotsCreated</b>: CDC snapshots committed. In streaming this should track one per
  *       commit window per destination. A flat line while data is flowing means commits have
  *       stalled.
- *   <li><b>committedDataFiles</b> / <b>committedDeleteFiles</b>: files published. Divide by {@code
- *       snapshotsCreated} for files per commit. A large ratio points to the small-file problem
+ *   <li><b>committedDataFileRecordCount</b> / <b>committedEqualityDeleteRecordCount</b>
+ *       (distributions, per file): rows in each committed data file and equality-delete file. Their
+ *       counts are the files committed; divide by {@code snapshotsCreated} for files per commit.
+ *   <li><b>committedDataFileByteSize</b> / <b>committedEqualityDeleteByteSize</b> (distributions,
+ *       per file): size of each committed file. A low mean points to the small-file problem
  *       described below.
- *   <li><b>committedRecords</b>: rows committed.
- *   <li><b>committedEqualityDeleteRecords</b>: equality-delete rows committed.
- *   <li><b>committedBytes</b>: total bytes of committed files.
  *   <li><b>commitDurationMs</b> (distribution): wall-clock time spent waiting on the Iceberg commit
  *       operation. Its tail is your catalog's health: a growing maximum usually means catalog
  *       contention, not a Beam problem.

sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitDeltasTest.java

@@ -56,6 +56,7 @@
 import org.apache.beam.sdk.io.iceberg.SerializableDataFile;
 import org.apache.beam.sdk.io.iceberg.SerializableDeleteFile;
 import org.apache.beam.sdk.io.iceberg.SnapshotInfo;
+import org.apache.beam.sdk.metrics.DistributionResult;
 import org.apache.beam.sdk.metrics.MetricNameFilter;
 import org.apache.beam.sdk.metrics.MetricResult;
 import org.apache.beam.sdk.metrics.Metrics;
@@ -302,6 +303,19 @@ private static long counter(PipelineResult result, String name) {
     return total;
   }
 
+  /** The named {@link CommitDeltas} distribution committed by the pipeline. */
+  private static DistributionResult distribution(PipelineResult result, String name) {
+    return Iterables.getOnlyElement(
+            result
+                .metrics()
+                .queryMetrics(
+                    MetricsFilter.builder()
+                        .addNameFilter(MetricNameFilter.named(CommitDeltas.class, name))
+                        .build())
+                .getDistributions())
+        .getCommitted();
+  }
+
   /**
    * Seeds a real prior CDC snapshot for {@code sinkId}, token keys written as the committer would.
    * The keys are literal strings on purpose: they double as a pin of the on-disk contract.
@@ -1243,7 +1257,7 @@ public void describeSkippedFilesListsUpToFivePathsAndFallsThroughToDeleteFiles()
   // Commit volume metrics.
   // ---------------------------------------------------------------------------------------------
 
-  /** A commit records its volume metrics: file/record counts, bytes, snapshots created. */
+  /** A commit records its volume metrics: records and bytes per committed file. */
   @Test
   public void recordsCommitVolumeMetrics() {
     TestTable tt = v2Table();
@@ -1283,12 +1297,18 @@ public void recordsCommitVolumeMetrics() {
     result.waitUntilFinish();
 
     assertThat(counter(result, "snapshotsCreated"), equalTo(1L));
-    assertThat(counter(result, "committedDataFiles"), equalTo(2L));
-    assertThat(counter(result, "committedDeleteFiles"), equalTo(2L));
-    // 2 data files carry 1 record each (id=1, id=2).
-    assertThat(counter(result, "committedRecords"), equalTo(2L));
-    assertThat(counter(result, "committedEqualityDeleteRecords"), equalTo(2L));
-    assertThat(counter(result, "committedBytes"), greaterThan(0L));
+    // One commit with 2 data files and 2 equality-delete files, each carrying 1 record (rows
+    // id=1, id=2; deletes for id=1, id=3).
+    assertThat(
+        distribution(result, "committedDataFileRecordCount"),
+        equalTo(DistributionResult.create(2, 2, 1, 1)));
+    assertThat(
+        distribution(result, "committedEqualityDeleteRecordCount"),
+        equalTo(DistributionResult.create(2, 2, 1, 1)));
+    for (String bytes : List.of("committedDataFileByteSize", "committedEqualityDeleteByteSize")) {
+      assertThat(distribution(result, bytes).getCount(), equalTo(2L));
+      assertThat(distribution(result, bytes).getMin(), greaterThan(0L));
+    }
   }
 
   // ---------------------------------------------------------------------------------------------

sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitterMetricsTest.java

@@ -69,20 +69,17 @@ private static DeleteFile deleteFile(
         .build();
   }
 
-  // Counts, records, and bytes cover both file kinds; only equality deletes count as delete
-  // records.
+  // The summary keeps both file lists for the per-file metrics.
   @Test
-  public void summarizesCountsRecordsAndBytes() {
-    CommitSummary summary =
-        CommitSummary.of(
-            ImmutableList.of(dataFile(SPEC_0, 10, 100), dataFile(SPEC_0, 20, 200)),
-            ImmutableList.of(deleteFile(SPEC_0, true, 4, 40), deleteFile(SPEC_0, false, 7, 70)));
+  public void keepsFilesAndFlagsEqualityDeletes() {
+    ImmutableList<DataFile> dataFiles =
+        ImmutableList.of(dataFile(SPEC_0, 10, 100), dataFile(SPEC_0, 20, 200));
+    ImmutableList<DeleteFile> deleteFiles =
+        ImmutableList.of(deleteFile(SPEC_0, true, 4, 40), deleteFile(SPEC_0, false, 7, 70));
+    CommitSummary summary = CommitSummary.of(dataFiles, deleteFiles);
 
-    assertThat(summary.dataFileCount, equalTo(2L));
-    assertThat(summary.deleteFileCount, equalTo(2L));
-    assertThat(summary.dataRecords, equalTo(30L));
-    assertThat(summary.equalityDeleteRecords, equalTo(4L));
-    assertThat(summary.bytes, equalTo(410L));
+    assertThat(summary.dataFiles, equalTo(dataFiles));
+    assertThat(summary.deleteFiles, equalTo(deleteFiles));
     assertThat(summary.hasEqualityDeletes, equalTo(true));
   }
 
@@ -92,9 +89,7 @@ public void positionDeletesAreNotEqualityDeletes() {
     CommitSummary summary =
         CommitSummary.of(ImmutableList.of(), ImmutableList.of(deleteFile(SPEC_0, false, 7, 70)));
 
-    assertThat(summary.equalityDeleteRecords, equalTo(0L));
     assertThat(summary.hasEqualityDeletes, equalTo(false));
-    assertThat(summary.bytes, equalTo(70L));
   }
 
   // The first spec id comes from the data files when there are any, and every spec id is collected.
@@ -116,8 +111,8 @@ public void firstSpecIdPrefersDataFilesAndSpecIdsCollectsAll() {
   public void emptyWindowHasNoSpec() {
     CommitSummary summary = CommitSummary.of(ImmutableList.of(), ImmutableList.of());
 
-    assertThat(summary.dataFileCount, equalTo(0L));
-    assertThat(summary.bytes, equalTo(0L));
+    assertThat(summary.dataFiles, empty());
+    assertThat(summary.deleteFiles, empty());
     assertThat(summary.firstSpecId, nullValue());
     assertThat(summary.specIds, empty());
   }