Implementation[openhouseTableCommitEvents]: Commit job for freshness in TableStatsCollectionSparkApp (#398)

## Summary

<!--- HINT: Replace #nnn with corresponding Issue number, if you are
fixing an existing issue -->

I extended the existing TableStatsCollectionSparkApp to implement the
logic for populating the openhouseTableCommitEvents table.
This new table will serve as the single source of truth for
commit-related metadata across all OpenHouse datasets, including:

- 	Commit ID
- 	Commit timestamp
- 	Commit operation
- 	Spark App ID
- 	Spark App Name 

This enables a unified, consistent, and efficient way to access commit
events for all OpenHouse tables.


## Output / Result

1. This PR populates the openhouseTableCommitEvents table by pushing
commit events from Snapshot Metadata table for all OH datasets.
2. Creates one row per commit across all OpenHouse tables.
3. Table will be updated daily via the TableStatsCollection job.
4. At every scheduled run, we will be processing all the active commit
events(non-expired) in the Snapshot Metadata table.
5. Every Partition will have Commit events for all the non-expired
Snapshots at the time of Job run.
6. This will have a lot of duplicates across partitions, but we can
handle it at query time in the downstream consumer.


## Changes

- [ ] Client-facing API Changes
- [ ] Internal API Changes
- [ ] Bug Fixes
- [x] New Features
- [ ] Performance Improvements
- [ ] Code Style
- [ ] Refactoring
- [ ] Documentation
- [ ] Tests

For all the boxes checked, please include additional details of the
changes made in this pull request.

## Testing Done
<!--- Check any relevant boxes with "x" -->

- [ ] Manually Tested on local docker setup. Please include commands
ran, and their output.
- [x] Added new tests for the changes made.
- [ ] Updated existing tests to reflect the changes made.
- [ ] No tests added or updated. Please explain why. If unsure, please
feel free to ask for help.
- [ ] Some other form of testing like staging or soak time in
production. Please explain.

For all the boxes checked, include a detailed description of the testing
done for the changes made in this pull request.

# Additional Information

- [ ] Breaking Changes
- [ ] Deprecations
- [ ] Large PR broken into smaller PRs, and PR plan linked in the
description.

For all the boxes checked, include additional details of the changes
made in this pull request.

---------

Co-authored-by: srawat <[email protected]>
Co-authored-by: Stas Pak <[email protected]>
Co-authored-by: Sumedh Sakdeo <[email protected]>

Author: 4 people

apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java

@@ -4,13 +4,15 @@
 import com.google.common.collect.Lists;
 import com.google.gson.Gson;
 import com.linkedin.openhouse.common.metrics.OtelEmitter;
+import com.linkedin.openhouse.common.stats.model.CommitEventTable;
 import com.linkedin.openhouse.common.stats.model.IcebergTableStats;
 import com.linkedin.openhouse.jobs.util.SparkJobUtil;
 import com.linkedin.openhouse.jobs.util.TableStatsCollector;
 import java.io.IOException;
 import java.nio.file.Paths;
 import java.time.ZonedDateTime;
 import java.time.temporal.ChronoUnit;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -530,15 +532,36 @@ static String partitionToString(StructLike partition) {
   public IcebergTableStats collectTableStats(String fqtn) {
     Table table = getTable(fqtn);
 
-    TableStatsCollector tableStatsCollector;
     try {
-      tableStatsCollector = new TableStatsCollector(fs(), spark, fqtn, table);
+      TableStatsCollector tableStatsCollector = new TableStatsCollector(fs(), spark, table);
+      return tableStatsCollector.collectTableStats();
     } catch (IOException e) {
       log.error("Unable to initialize file system for table stats collection", e);
       return null;
+    } catch (Exception e) {
+      log.error("Failed to collect table stats for table: {}", fqtn, e);
+      return null;
     }
+  }
 
-    IcebergTableStats tableStats = tableStatsCollector.collectTableStats();
-    return tableStats;
+  /**
+   * Collect commit events for a given fully-qualified table name.
+   *
+   * @param fqtn fully-qualified table name
+   * @return List of CommitEventTable objects (event_timestamp_ms will be set at publish time)
+   */
+  public List<CommitEventTable> collectCommitEventTable(String fqtn) {
+    Table table = getTable(fqtn);
+
+    try {
+      TableStatsCollector tableStatsCollector = new TableStatsCollector(fs(), spark, table);
+      return tableStatsCollector.collectCommitEventTable();
+    } catch (IOException e) {
+      log.error("Unable to initialize file system for commit events collection", e);
+      return Collections.emptyList();
+    } catch (Exception e) {
+      log.error("Failed to collect commit events for table: {}", fqtn, e);
+      return Collections.emptyList();
+    }
   }
 }

apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/TableStatsCollectionSparkApp.java

@@ -3,12 +3,15 @@
 import com.google.gson.Gson;
 import com.linkedin.openhouse.common.metrics.DefaultOtelConfig;
 import com.linkedin.openhouse.common.metrics.OtelEmitter;
+import com.linkedin.openhouse.common.stats.model.CommitEventTable;
 import com.linkedin.openhouse.common.stats.model.IcebergTableStats;
 import com.linkedin.openhouse.jobs.spark.state.StateManager;
 import com.linkedin.openhouse.jobs.util.AppsOtelEmitter;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Supplier;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.Option;
@@ -31,8 +34,46 @@ public TableStatsCollectionSparkApp(
   protected void runInner(Operations ops) {
     log.info("Running TableStatsCollectorApp for table {}", fqtn);
 
-    IcebergTableStats icebergTableStats = ops.collectTableStats(fqtn);
-    publishStats(icebergTableStats);
+    // Run stats collection and commit events collection in parallel
+    long startTime = System.currentTimeMillis();
+
+    CompletableFuture<IcebergTableStats> statsFuture =
+        executeWithTimingAsync(
+            "table stats collection",
+            () -> ops.collectTableStats(fqtn),
+            result -> String.format("%s", fqtn));
+
+    CompletableFuture<List<CommitEventTable>> commitEventsFuture =
+        executeWithTimingAsync(
+            "commit events collection",
+            () -> ops.collectCommitEventTable(fqtn),
+            result -> String.format("%s (%d events)", fqtn, result.size()));
+
+    // Wait for both to complete
+    CompletableFuture.allOf(statsFuture, commitEventsFuture).join();
+
+    long endTime = System.currentTimeMillis();
+    log.info(
+        "Total collection time for table: {} in {} ms (parallel execution)",
+        fqtn,
+        (endTime - startTime));
+
+    // Publish results
+    IcebergTableStats icebergTableStats = statsFuture.join();
+    if (icebergTableStats != null) {
+      publishStats(icebergTableStats);
+    } else {
+      log.warn("Skipping stats publishing for table: {} due to collection failure", fqtn);
+    }
+
+    List<CommitEventTable> commitEvents = commitEventsFuture.join();
+    if (commitEvents != null && !commitEvents.isEmpty()) {
+      publishCommitEvents(commitEvents);
+    } else {
+      log.warn(
+          "Skipping commit events publishing for table: {} due to collection failure or no events",
+          fqtn);
+    }
   }
 
   /**
@@ -45,12 +86,57 @@ protected void publishStats(IcebergTableStats icebergTableStats) {
     log.info(new Gson().toJson(icebergTableStats));
   }
 
+  /**
+   * Publish commit events. Override this method in li-openhouse to send to Kafka.
+   *
+   * @param commitEvents List of commit events to publish
+   */
+  protected void publishCommitEvents(List<CommitEventTable> commitEvents) {
+    // Set event timestamp at publish time
+    long eventTimestampInEpochMs = System.currentTimeMillis();
+    commitEvents.forEach(event -> event.setEventTimestampMs(eventTimestampInEpochMs));
+
+    log.info("Publishing commit events for table: {}", fqtn);
+    log.info(new Gson().toJson(commitEvents));
+  }
+
   public static void main(String[] args) {
     OtelEmitter otelEmitter =
         new AppsOtelEmitter(Arrays.asList(DefaultOtelConfig.getOpenTelemetry()));
     createApp(args, otelEmitter).run();
   }
 
+  /**
+   * Execute a supplier asynchronously with timing and logging.
+   *
+   * @param operationName Name of the operation for logging
+   * @param supplier The operation to execute
+   * @param resultFormatter Function to format the result for logging
+   * @param <T> Return type of the operation
+   * @return CompletableFuture wrapping the operation result
+   */
+  private <T> CompletableFuture<T> executeWithTimingAsync(
+      String operationName,
+      Supplier<T> supplier,
+      java.util.function.Function<T, String> resultFormatter) {
+    return CompletableFuture.supplyAsync(
+        () -> {
+          long startTime = System.currentTimeMillis();
+          log.info("Starting {} for table: {}", operationName, fqtn);
+          T result = supplier.get();
+          long endTime = System.currentTimeMillis();
+
+          String resultDescription =
+              (result != null) ? resultFormatter.apply(result) : "null (collection failed)";
+          log.info(
+              "Completed {} for table: {} in {} ms",
+              operationName,
+              resultDescription,
+              (endTime - startTime));
+          return result;
+        });
+  }
+
   public static TableStatsCollectionSparkApp createApp(String[] args, OtelEmitter otelEmitter) {
     List<Option> extraOptions = new ArrayList<>();
     extraOptions.add(new Option("t", "tableName", true, "Fully-qualified table name"));

apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/TableStatsCollector.java

@@ -1,6 +1,8 @@
 package com.linkedin.openhouse.jobs.util;
 
+import com.linkedin.openhouse.common.stats.model.CommitEventTable;
 import com.linkedin.openhouse.common.stats.model.IcebergTableStats;
+import java.util.List;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hadoop.fs.FileSystem;
@@ -14,7 +16,6 @@ public class TableStatsCollector {
 
   private FileSystem fs;
   private SparkSession spark;
-  String fqtn;
   Table table;
 
   /** Collect table stats. */
@@ -25,14 +26,25 @@ public IcebergTableStats collectTableStats() {
         TableStatsCollectorUtil.populateTableMetadata(table, stats);
     IcebergTableStats statsWithReferenceFiles =
         TableStatsCollectorUtil.populateStatsOfAllReferencedFiles(
-            fqtn, table, spark, statsWithMetadataData);
+            table, spark, statsWithMetadataData);
     IcebergTableStats statsWithCurrentSnapshot =
-        TableStatsCollectorUtil.populateStatsForSnapshots(
-            fqtn, table, spark, statsWithReferenceFiles);
+        TableStatsCollectorUtil.populateStatsForSnapshots(table, spark, statsWithReferenceFiles);
 
     IcebergTableStats tableStats =
-        TableStatsCollectorUtil.populateStorageStats(fqtn, table, fs, statsWithCurrentSnapshot);
+        TableStatsCollectorUtil.populateStorageStats(table, fs, statsWithCurrentSnapshot);
 
     return tableStats;
   }
+
+  /**
+   * Collect commit events for the table.
+   *
+   * <p>Note: Returns List (loads into memory). This is acceptable because Iceberg retention limits
+   * active snapshots to a manageable number (typically <10k per table).
+   *
+   * @return List of CommitEventTable objects (event_timestamp_ms will be set at publish time)
+   */
+  public List<CommitEventTable> collectCommitEventTable() {
+    return TableStatsCollectorUtil.populateCommitEventTable(table, spark);
+  }
 }