Initial draft for timestamping DLO generation and compaction.

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

@@ -1,15 +1,21 @@
 package com.linkedin.openhouse.jobs.spark;
 
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.reflect.TypeToken;
 import com.linkedin.openhouse.datalayout.config.DataCompactionConfig;
 import com.linkedin.openhouse.jobs.spark.state.StateManager;
 import com.linkedin.openhouse.jobs.util.AppConstants;
+import com.linkedin.openhouse.jobs.util.CompactionEvent;
 import io.opentelemetry.api.common.AttributeKey;
 import io.opentelemetry.api.common.Attributes;
+import java.lang.reflect.Type;
 import java.util.ArrayList;
 import java.util.List;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.Option;
+import org.apache.commons.lang.StringEscapeUtils;
 import org.apache.commons.lang3.math.NumberUtils;
 import org.apache.iceberg.actions.RewriteDataFiles;
 
@@ -24,6 +30,7 @@
 @Slf4j
 public class DataCompactionSparkApp extends BaseTableSparkApp {
   private final DataCompactionConfig config;
+  public static final String DATA_LAYOUT_COMPACTION_PROPERTY_KEY = "write.data-layout.compaction";
 
   protected DataCompactionSparkApp(
       String jobId, StateManager stateManager, String fqtn, DataCompactionConfig config) {
@@ -58,6 +65,23 @@ protected void runInner(Operations ops) {
           fileGroupRewriteResult.rewrittenDataFilesCount(),
           fileGroupRewriteResult.rewrittenBytesCount());
     }
+    // Add compaction timestamp to table properties.
+    CompactionEvent event =
+        CompactionEvent.builder()
+            .compactionTimestamp(System.currentTimeMillis())
+            .addedDataFilesCount(result.addedDataFilesCount())
+            .rewrittenDataFilesCount(result.rewrittenDataFilesCount())
+            .rewrittenBytesCount(result.rewrittenBytesCount())
+            .build();
+    Gson gson = new GsonBuilder().create();
+    Type type = new TypeToken<CompactionEvent>() {}.getType();
+    ops.spark()
+        .sql(
+            String.format(
+                "ALTER TABLE %s SET TBLPROPERTIES ('%s' = '%s')",
+                ops.getTable(fqtn),
+                DATA_LAYOUT_COMPACTION_PROPERTY_KEY,
+                StringEscapeUtils.escapeJava(gson.toJson(event, type))));
     METER
         .counterBuilder(AppConstants.ADDED_DATA_FILE_COUNT)
         .build()

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

@@ -0,0 +1,14 @@
+package com.linkedin.openhouse.jobs.util;
+
+import lombok.Builder;
+import lombok.Data;
+
+/** Compaction event to be logged as part of the table properties. */
+@Data
+@Builder
+public class CompactionEvent {
+  private final long compactionTimestamp;
+  private final long addedDataFilesCount;
+  private final long rewrittenDataFilesCount;
+  private final long rewrittenBytesCount;
+}

libs/datalayout/src/main/java/com/linkedin/openhouse/datalayout/generator/OpenHouseDataLayoutStrategyGenerator.java

@@ -124,6 +124,7 @@ private Optional<DataLayoutStrategy> generateCompactionStrategy() {
     return Optional.of(
         DataLayoutStrategy.builder()
             .config(configBuilder.build())
+            .computationTimestamp(System.currentTimeMillis())
             .cost(computeGbHr)
             .gain(reducedFileCount)
             .score(reducedFileCountPerComputeGbHr)

libs/datalayout/src/main/java/com/linkedin/openhouse/datalayout/strategy/DataLayoutStrategy.java

@@ -11,6 +11,7 @@
 @Data
 @Builder
 public class DataLayoutStrategy {
+  private final long computationTimestamp;
   private final double score;
   private final double entropy;
   private final double cost;