Add IcebergCommitEventStats model for capturing commit events (#390)

## Summary

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

[DTQ-6354](https://linkedin.atlassian.net/browse/DTQ-6354) Briefly
discuss the summary of the changes made in this
pull request in 2-3 lines.

Implemented a Base class for all commit event statistics. 

## 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.
- [ ] 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

services/common/src/main/java/com/linkedin/openhouse/common/stats/model/BaseEventModels.java

@@ -0,0 +1,67 @@
+package com.linkedin.openhouse.common.stats.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.NonNull;
+import lombok.experimental.SuperBuilder;
+
+/** Container for base model classes used in the dataset event hierarchy. */
+public final class BaseEventModels {
+
+  private BaseEventModels() {
+    // Utility class - no instantiation
+  }
+
+  /** Base class containing table identification metadata. */
+  @Data
+  @SuperBuilder
+  @NoArgsConstructor
+  @AllArgsConstructor
+  public static class BaseTableIdentifier {
+
+    /** Name of the database for the dataset */
+    @NonNull private String databaseName;
+
+    /** Name of the table for the dataset */
+    @NonNull private String tableName;
+
+    /** Name of the cluster (e.g., holdem/war) */
+    @NonNull private String clusterName;
+
+    /**
+     * Fully qualified path to the root table metadata JSON. Useful to distinguish uniqueness in
+     * case the table is dropped and recreated with the same name.
+     */
+    @NonNull private String tableMetadataLocation;
+
+    /**
+     * String representation of the Iceberg PartitionSpec for the dataset.
+     *
+     * <p>Use the string from PartitionSpec.toString() to describe partitioning. For non-partitioned
+     * tables, use PartitionSpec.UNPARTITIONED_SPEC.toString(). See:
+     * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/PartitionSpec.java
+     */
+    @NonNull private String partitionSpec;
+  }
+
+  /** Base class for commit events that contains dataset information. */
+  @Data
+  @SuperBuilder
+  @NoArgsConstructor
+  @AllArgsConstructor
+  public abstract static class BaseCommitEvent {
+
+    /** Dataset information for this commit event */
+    @NonNull private BaseTableIdentifier dataset;
+
+    /** commit Metadata for this commit */
+    @NonNull private CommitMetadata commitMetadata;
+
+    /**
+     * Timestamp (in epoch milliseconds) representing when the collector job processed and ingested
+     * the corresponding event.
+     */
+    @NonNull private Long eventTimestampMs;
+  }
+}

services/common/src/main/java/com/linkedin/openhouse/common/stats/model/ColumnData.java

@@ -0,0 +1,50 @@
+package com.linkedin.openhouse.common.stats.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.NonNull;
+
+/**
+ * Column-level data interface for type-safe values used in events and stats.
+ *
+ * <p>Implementations encapsulate a specific value type to provide compile-time safety.
+ */
+public interface ColumnData {
+  /** Returns the column name this data applies to. */
+  String getColumnName();
+
+  /** Returns the underlying value. */
+  Object getValue();
+
+  /** Long-valued column data for counts and sizes in bytes. */
+  @Data
+  @Builder
+  @NoArgsConstructor
+  @AllArgsConstructor
+  class LongColumnData implements ColumnData {
+    @NonNull private String columnName;
+    @NonNull private Long value;
+  }
+
+  /** String-valued column data for min/max of strings, dates, and timestamps. */
+  @Data
+  @Builder
+  @NoArgsConstructor
+  @AllArgsConstructor
+  class StringColumnData implements ColumnData {
+    @NonNull private String columnName;
+    @NonNull private String value;
+  }
+
+  /** Double-valued column data for floating-point min/max values. */
+  @Data
+  @Builder
+  @NoArgsConstructor
+  @AllArgsConstructor
+  class DoubleColumnData implements ColumnData {
+    @NonNull private String columnName;
+    @NonNull private Double value;
+  }
+}

services/common/src/main/java/com/linkedin/openhouse/common/stats/model/CommitEventTable.java

@@ -0,0 +1,26 @@
+package com.linkedin.openhouse.common.stats.model;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+/**
+ * Data model for openhouseTableCommitEvents table.
+ *
+ * <p>Stores commit-level metadata for dataset changes. Each record represents a commit on table and
+ * can be linked to multiple partition events via commitId.
+ *
+ * <p><b>Cardinality</b>: One commit event can have N partition events. See {@link
+ * CommitEventTablePartitions} for partition-level details.
+ *
+ * @see CommitEventTablePartitions
+ */
+@Data
+@SuperBuilder
+@NoArgsConstructor
+@EqualsAndHashCode(callSuper = true)
+public class CommitEventTable extends BaseEventModels.BaseCommitEvent {
+  // All fields inherited from BaseCommitEvent and BaseTableIdentifier
+  // datasetType is available via dataset.datasetType
+}

services/common/src/main/java/com/linkedin/openhouse/common/stats/model/CommitEventTablePartitionStats.java

@@ -0,0 +1,89 @@
+package com.linkedin.openhouse.common.stats.model;
+
+import java.util.List;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+/**
+ * Data model for openhouseCommitEventTablePartitionStats table.
+ *
+ * <p>Stores partition-level metadata and statistics such as null count, NaN count, row count, and
+ * min/max values. Can represent both partition-level and table-level statistics.
+ *
+ * <p><b>Cardinality</b>: Each partition stats record references the latest commit that modified the
+ * partition via commitMetadata.commitId (Foreign Key to {@link CommitEventTable}).
+ *
+ * <p>Stats are updated/replaced when new commits modify the partition.
+ *
+ * @see CommitEventTable
+ * @see CommitEventTablePartitions
+ * @see CommitMetadata
+ */
+@Data
+@SuperBuilder
+@NoArgsConstructor
+@AllArgsConstructor
+@EqualsAndHashCode(callSuper = true)
+public class CommitEventTablePartitionStats extends BaseEventModels.BaseCommitEvent {
+
+  /**
+   * List of per-partition column values associated with the statistics.
+   *
+   * <p>Each element carries the partition column name and its typed value using {@link ColumnData}
+   * implementations. The list order should align with the partition spec order for determinism.
+   *
+   * <p>Can be null if the statistics is on a table level (non-partitioned table or table-level
+   * aggregates).
+   *
+   * <p>Example for non-null partition: [ new ColumnData.StringColumnData("datepartition",
+   * "2025-01-25"), new ColumnData.StringColumnData("hourpartition", "12") ]
+   */
+  private List<ColumnData> partitionData;
+
+  /**
+   * Total number of rows corresponding to the given partition specification if partitionData is not
+   * null; otherwise, the row count for the entire table.
+   */
+  private Long rowCount;
+
+  /**
+   * Total number of columns corresponding to the given partition specification if partitionData is
+   * not null; otherwise, the column count for the entire table.
+   */
+  private Long columnCount;
+
+  /**
+   * Stores null count statistics for each column in the dataset. Each element represents a column
+   * name and its corresponding number of null values.
+   */
+  private List<ColumnData> nullCount;
+
+  /**
+   * Stores NaN count statistics for each column in the dataset. Each element represents a column
+   * name and its corresponding number of NaN values.
+   *
+   * <p>Applicable only to numeric data types such as float, double, or decimal.
+   */
+  private List<ColumnData> nanCount;
+
+  /**
+   * Stores minimum value statistics for each column in the dataset. Each element represents a
+   * column name and its corresponding minimum value.
+   */
+  private List<ColumnData> minValue;
+
+  /**
+   * Stores maximum value statistics for each column in the dataset. Each element represents a
+   * column name and its corresponding maximum value.
+   */
+  private List<ColumnData> maxValue;
+
+  /**
+   * Stores column size in bytes statistics for each column in the dataset. Each element represents
+   * a column name and its corresponding size in bytes.
+   */
+  private List<ColumnData> columnSizeInBytes;
+}

services/common/src/main/java/com/linkedin/openhouse/common/stats/model/CommitEventTablePartitions.java

@@ -0,0 +1,43 @@
+package com.linkedin.openhouse.common.stats.model;
+
+import java.util.List;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+import lombok.NonNull;
+import lombok.experimental.SuperBuilder;
+
+/**
+ * Data model for openhouseTableCommitEventPartitions table.
+ *
+ * <p>Stores partition-level information for each commit. Maps commit events to specific partitions
+ * affected by that commit. One commit can correspond to multiple partition records. A commit event
+ * represents a single commit-partition pair, capturing what changed for that specific partition in
+ * that specific commit
+ *
+ * <p><b>Naming</b>: Represents "partitions of a commit event". This aligns with the conceptual
+ * model where CommitEvent has a "partitions" field that's been normalized into a separate table.
+ *
+ * <p><b>Cardinality</b>: N partition records linked to 1 commit event via commitId foreign key.
+ *
+ * @see CommitEventTable
+ */
+@Data
+@SuperBuilder
+@NoArgsConstructor
+@AllArgsConstructor
+@EqualsAndHashCode(callSuper = true)
+public class CommitEventTablePartitions extends BaseEventModels.BaseCommitEvent {
+
+  /**
+   * List of per-partition column values associated with the commit.
+   *
+   * <p>Each element carries the partition column name and its typed value using {@link ColumnData}
+   * implementations. The list order should align with the partition spec order for determinism.
+   *
+   * <p>Example: [ new ColumnData.StringColumnData("datepartition", "2025-01-25"), new
+   * ColumnData.StringColumnData("hourpartition", "12") ]
+   */
+  @NonNull private List<ColumnData> partitionData;
+}

services/common/src/main/java/com/linkedin/openhouse/common/stats/model/CommitMetadata.java

@@ -0,0 +1,36 @@
+package com.linkedin.openhouse.common.stats.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.NonNull;
+
+/** Standalone class representing commit metadata. */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class CommitMetadata {
+
+  /** Unique identifier for the commit event */
+  @NonNull private Long commitId;
+
+  /** Timestamp of the commit event captured in epoch milliseconds */
+  @NonNull private Long commitTimestampMs;
+
+  /**
+   * Unique application identifier (e.g., Spark Application ID) associated with the process or job
+   * that performed the commit
+   */
+  private String commitAppId;
+
+  /**
+   * Descriptive name of the application or job that executed the commit. Helps in identifying the
+   * pipeline or workflow responsible for the data change.
+   */
+  private String commitAppName;
+
+  /** Type of operation performed during the commit (e.g., APPEND, OVERWRITE, DELETE, REPLACE) */
+  private CommitOperation commitOperation;
+}

services/common/src/main/java/com/linkedin/openhouse/common/stats/model/CommitOperation.java

@@ -0,0 +1,20 @@
+package com.linkedin.openhouse.common.stats.model;
+
+/**
+ * Enum representing the type of operation performed during a dataset commit.
+ *
+ * <p>Defines the possible modification operations that can be applied to a dataset.
+ */
+public enum CommitOperation {
+  /** Append new data to the dataset */
+  APPEND,
+
+  /** Overwrite existing data in the dataset */
+  OVERWRITE,
+
+  /** Delete data from the dataset */
+  DELETE,
+
+  /** Replace data in the dataset */
+  REPLACE
+}