Fail retention app when the columnPattern mismatch partition spec for… (#552)

Author: jiang95-dev

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

@@ -50,6 +50,7 @@
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.spark.actions.SparkActions;
@@ -339,6 +340,17 @@ private Map<String, List<String>> prepareBackupDataManifests(
     TableScan scan = table.newScan().filter(filter);
     try (CloseableIterable<FileScanTask> filesIterable = scan.planFiles()) {
       List<FileScanTask> filesList = Lists.newArrayList(filesIterable);
+      filesList.stream()
+          .filter(task -> !Expressions.alwaysTrue().isEquivalentTo(task.residual()))
+          .findFirst()
+          .ifPresent(
+              task -> {
+                throw new IllegalStateException(
+                    String.format(
+                        "Retention with backup enabled requires a metadata-only delete for table %s, "
+                            + "but file %s has residual filter %s, which would require a row-level rewrite.",
+                        fqtn, task.file().path(), task.residual()));
+              });
       return filesList.stream()
           .collect(
               Collectors.groupingBy(

apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java

@@ -309,6 +309,54 @@ public void testRetentionDataManifestWithTimestampPartitionedTable() throws Exce
     }
   }
 
+  @Test
+  public void testRetentionWithBackupFailsWhenColumnPatternMismatchesPartition() throws Exception {
+    final String tableName = "db.test_retention_backup_pattern_mismatch";
+    try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) {
+      // The table is partitioned on `datepartition`, but retention will filter
+      // on `time_col` using a pattern unrelated to the partitioning. For each
+      // file's per-file min/max to actually straddle the cutoff (and produce
+      // a non-trivial residual), both `time_col` values within a partition
+      // must live in the same data file — so we force a single writer task
+      // via the COALESCE(1) hint.
+      ops.spark().sql(String.format("DROP TABLE IF EXISTS %s", tableName)).show();
+      ops.spark()
+          .sql(
+              String.format(
+                  "CREATE TABLE %s (data string, datepartition string, time_col string) "
+                      + "PARTITIONED BY (datepartition)",
+                  tableName))
+          .show();
+      ops.spark()
+          .sql(
+              "SELECT data, datepartition, time_col FROM VALUES "
+                  + "('a', '2024-01', '2020-01-01-00'), "
+                  + "('b', '2024-01', '2030-01-01-00'), "
+                  + "('c', '2024-02', '2020-01-01-00'), "
+                  + "('d', '2024-02', '2030-01-01-00') "
+                  + "AS t(data, datepartition, time_col)")
+          .coalesce(1)
+          .writeTo(tableName)
+          .append();
+
+      // Fix `now` so the cutoff (now - 1 day, formatted yyyy-MM-dd-HH) falls
+      // strictly between each file's min ("2020-01-01-00") and max
+      // ("2030-01-01-00") — forcing a non-trivial residual on every file.
+      ZonedDateTime now = ZonedDateTime.of(2025, 6, 15, 10, 0, 0, 0, ZoneOffset.UTC);
+      IllegalStateException ex =
+          Assertions.assertThrows(
+              IllegalStateException.class,
+              () ->
+                  ops.runRetention(
+                      tableName, "time_col", "yyyy-MM-dd-HH", "day", 1, true, ".backup", now));
+      Assertions.assertTrue(
+          ex.getMessage().contains("metadata-only delete"),
+          "Expected metadata-only delete error, got: " + ex.getMessage());
+      // DELETE should not have executed: all 4 rows remain.
+      verifyRowCount(ops, tableName, 4);
+    }
+  }
+
   @Test
   public void testOrphanFilesDeletionJavaAPI() throws Exception {
     final String tableName = "db.test_ofd_java";