Core, Spark: Add added-dv-files metric for DV writes

[wangyum]

Why are the changes needed?

DV-heavy commits can produce many logical delete vectors while writing only a few physical Puffin container files. Today we expose added-dvs but not the number of Puffin files, which makes it hard to understand physical output and commit behavior.

What changes were proposed in this pull request?

• Add a new snapshot summary metric: added-dv-files
  • Counts distinct Puffin container file paths for added DVs.
• Propagate this metric through commit reporting:
  • CommitMetricsResult
  • CommitMetricsResultParser (to/from JSON)
• Expose the metric in Spark write custom metrics for Spark 4.x:
  • Add addedPuffinFiles custom metric class
  • Wire it into SparkWriteUtil.supportedCustomMetrics()
  • Wire it into SparkWriteUtil.customTaskMetrics(...)
• Update the spec table for optional snapshot summary fields.
• Extend core and Spark tests to validate the new metric.

How was this patch tested?

Unit test and manual test:

./gradlew -DsparkVersions=4.1 -DscalaVersion=2.13 :iceberg-spark:iceberg-spark-extensions-4.1_2.13:test --tests org.apache.iceberg.spark.extensions.TestAddedPuffinFilesUiManual

/*
 * 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 org.apache.iceberg.spark.extensions;

import static org.assertj.core.api.Assertions.assertThat;

import java.net.InetAddress;
import java.nio.file.Path;
import java.util.List;
import java.util.concurrent.TimeUnit;
import org.apache.iceberg.spark.SparkCatalog;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import scala.Option;

public class TestAddedPuffinFilesUiManual {

  @TempDir private Path warehouseDir;

  private SparkSession spark;

  @AfterEach
  public void stopSpark() {
    if (spark != null) {
      spark.stop();
      spark = null;
    }
  }

  @Test
  public void testV3MergeIntoWithUiPause() throws InterruptedException {
    int totalRows = 2_000;
    int initialFilePartitions = 200;

    String loopbackHost = InetAddress.getLoopbackAddress().getHostAddress();

    spark =
        SparkSession.builder()
            .master("local[2]")
            .appName("iceberg-added-dv-files-ui-manual")
            .config("spark.driver.host", loopbackHost)
            .config("spark.driver.bindAddress", loopbackHost)
            .config("spark.ui.enabled", "true")
            .config("spark.ui.port", "4040")
            .config("spark.sql.extensions", IcebergSparkSessionExtensions.class.getName())
            .config("spark.sql.catalog.local", SparkCatalog.class.getName())
            .config("spark.sql.catalog.local.type", "hadoop")
            .config("spark.sql.catalog.local.warehouse", "file:" + warehouseDir.toAbsolutePath())
            .config("spark.sql.shuffle.partitions", "2")
            .getOrCreate();

    spark.sql("CREATE NAMESPACE IF NOT EXISTS local.db");

    spark.sql(
        "CREATE TABLE local.db.target (id INT, dep STRING) USING iceberg "
            + "TBLPROPERTIES ("
            + "'format-version'='3',"
            + "'write.merge.mode'='merge-on-read',"
            + "'write.update.mode'='merge-on-read',"
            + "'write.delete.mode'='merge-on-read')");

    spark.sql(
        String.format(
            "INSERT INTO local.db.target "
                + "SELECT /*+ REPARTITION(%d) */ CAST(id AS INT), "
                + "CASE WHEN id %% 2 = 0 THEN 'hr' ELSE 'it' END "
                + "FROM range(1, %d)",
            initialFilePartitions, totalRows + 1));

    spark.sql(
        "CREATE OR REPLACE TEMP VIEW source AS "
            + String.format(
                "SELECT CAST(id AS INT) AS id, "
                    + "CONCAT('updated-', CAST(id %% 10 AS STRING)) AS dep "
                    + "FROM range(1, %d)",
                totalRows + 1));

    spark.sql(
        "MERGE INTO local.db.target AS t USING source AS s "
            + "ON t.id = s.id "
            + "WHEN MATCHED THEN UPDATE SET dep = s.dep");

    Option<String> uiWebUrl = spark.sparkContext().uiWebUrl();
    String uiUrl = uiWebUrl.isDefined() ? uiWebUrl.get() : "N/A";

    System.out.println("Spark application: " + spark.sparkContext().applicationId());
    System.out.println("Spark UI URL: " + uiUrl);

    Thread.sleep(TimeUnit.MINUTES.toMillis(10));
  }
}

UI:

Closes #16068

[singhpk234]

puffin container is used for stats tracking too, if we just want puffin files for DV then may be added-dv-files a better name

[wangyum]

renamed it to added-dv-files

[github-actions]

This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.