Stabilize Spark timing-sensitive tests with unique table names.

Use per-test randomized table identifiers in OperationsTest and SparkMoRFunctionalTest to reduce cross-test table collisions and improve determinism under parallel execution.

Made-with: Cursor

Author: cbb330

apps/spark-3.5/src/test/java/com/linkedin/openhouse/catalog/e2e/SparkMoRFunctionalTest.java

@@ -9,6 +9,7 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.UUID;
 import java.util.function.BiFunction;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -38,7 +39,7 @@
 
 @Slf4j
 public class SparkMoRFunctionalTest extends OpenHouseSparkITest {
-  static final String tableName = "db.test_data_compaction";
+  private String tableName;
 
   final BiFunction<Operations, Table, RewriteDataFiles.Result> rewriteFunc =
       (ops, table) ->
@@ -58,13 +59,18 @@ public class SparkMoRFunctionalTest extends OpenHouseSparkITest {
   @BeforeEach
   public void setUp() throws Exception {
     ops = Operations.withCatalog(getSparkSession(), null);
+    tableName = uniqueTableName("db.test_data_compaction");
   }
 
   @AfterEach
   public void cleanUp() throws Exception {
     sql("DROP TABLE IF EXISTS %s", tableName);
   }
 
+  private static String uniqueTableName(String baseName) {
+    return baseName + "_" + UUID.randomUUID().toString().replace("-", "");
+  }
+
   protected void initTable() {
     sql(
         "ALTER TABLE %s SET TBLPROPERTIES ('write.delete.mode'='merge-on-read', 'write.update.mode'='merge-on-read', 'write.merge.mode'='merge-on-read', 'write.delete.distribution-mode'='range');",

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

@@ -26,6 +26,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 import java.util.function.BiFunction;
 import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
@@ -53,7 +54,7 @@ public class OperationsTest extends OpenHouseSparkITest {
 
   @Test
   public void testRetentionSparkApp() throws Exception {
-    final String tableName = "db.test_retention_sql";
+    final String tableName = uniqueTableName("db.test_retention_sql");
     try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) {
       prepareTableWithRetentionAndSharingPolicies(ops, tableName, "1d", true);
       populateTable(ops, tableName, 3);
@@ -66,12 +67,12 @@ public void testRetentionSparkApp() throws Exception {
 
   @Test
   public void testRetentionSparkAppWithStringPartitionColumns() throws Exception {
-    final String tableName1 = "db.test_retention_string_partition1";
-    final String tableName2 = "db.test_retention_string_partition2";
-    final String tableName3 = "db.test_retention_string_partition3";
-    final String tableName4 = "db.test_retention_string_partition4";
-    final String tableName5 = "db.test_retention_string_partition5";
-    final String tableName6 = "db.test_retention_string_partition6";
+    final String tableName1 = uniqueTableName("db.test_retention_string_partition1");
+    final String tableName2 = uniqueTableName("db.test_retention_string_partition2");
+    final String tableName3 = uniqueTableName("db.test_retention_string_partition3");
+    final String tableName4 = uniqueTableName("db.test_retention_string_partition4");
+    final String tableName5 = uniqueTableName("db.test_retention_string_partition5");
+    final String tableName6 = uniqueTableName("db.test_retention_string_partition6");
 
     List<String> rowValue = new ArrayList<>();
     try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) {
@@ -133,9 +134,7 @@ public void testRetentionSparkAppWithStringPartitionColumns() throws Exception {
       rowValue.add(currentDatesFormatMismatched.get(0).get(0).toString());
       runRetentionJobWithStringPartitionColumns(
           ops, tableName5, rowValue, "datePartition", "yyyy-MM.dd.HH", "day");
-      ops.spark()
-          .sql("select * from openhouse.db.test_retention_string_partition5")
-          .collectAsList();
+      ops.spark().sql(String.format("select * from %s", tableName5)).collectAsList();
       verifyRowCount(ops, tableName5, 0);
       rowValue.clear();
 
@@ -164,7 +163,7 @@ private void runRetentionJobWithStringPartitionColumns(
 
   @Test
   public void testRetentionCreatesSnapshotsOnNoOpDelete() throws Exception {
-    final String tableName = "db_test.test_retention_sql";
+    final String tableName = uniqueTableName("db_test.test_retention_sql");
     try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) {
       prepareTable(ops, tableName);
       populateTable(ops, tableName, 4);
@@ -182,7 +181,7 @@ public void testRetentionCreatesSnapshotsOnNoOpDelete() throws Exception {
   public void testRetentionDataManifestWithStringDatePartitionedTable() throws Exception {
     try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) {
       // set up
-      String tableName = "db.test_string_partition";
+      String tableName = uniqueTableName("db.test_string_partition");
       String columnName = "datepartition";
       String columnPattern = "yyyy-MM-dd-HH";
       String granularity = "DAY";
@@ -260,7 +259,7 @@ public void testRetentionDataManifestWithStringDatePartitionedTable() throws Exc
   public void testRetentionDataManifestWithTimestampPartitionedTable() throws Exception {
     try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) {
       // set up
-      String tableName = "db.test_time_partition";
+      String tableName = uniqueTableName("db.test_time_partition");
       String columnName = "ts";
       String columnPattern = "";
       String granularity = "DAY";
@@ -311,7 +310,7 @@ public void testRetentionDataManifestWithTimestampPartitionedTable() throws Exce
 
   @Test
   public void testOrphanFilesDeletionJavaAPI() throws Exception {
-    final String tableName = "db.test_ofd_java";
+    final String tableName = uniqueTableName("db.test_ofd_java");
     final String testOrphanFileName = "data/test_orphan_file.orc";
     final int numInserts = 3;
     try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) {
@@ -338,7 +337,7 @@ public void testOrphanFilesDeletionJavaAPI() throws Exception {
 
   @Test
   public void testOrphanFilesDeletionIgnoresFilesInBackupDir() throws Exception {
-    final String tableName = "db.test_ofd_java";
+    final String tableName = uniqueTableName("db.test_ofd_java");
     final String testOrphanFileName = "data/test_orphan_file.orc";
     final int numInserts = 3;
     try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) {
@@ -365,7 +364,7 @@ public void testOrphanFilesDeletionIgnoresFilesInBackupDir() throws Exception {
 
   @Test
   public void testOrphanFilesDeletionDeleteNonDataFiles() throws Exception {
-    final String tableName = "db.test_ofd";
+    final String tableName = uniqueTableName("db.test_ofd");
     final String testOrphanFileName = "metadata/test_orphan_file.avro";
     final int numInserts = 3;
     try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) {
@@ -392,7 +391,7 @@ public void testOrphanFilesDeletionDeleteNonDataFiles() throws Exception {
 
   @Test
   public void testOrphanFilesDeletionBackupDisabled() throws Exception {
-    final String tableName = "db.test_ofd";
+    final String tableName = uniqueTableName("db.test_ofd");
     final String testOrphanFileName = "data/test_orphan_file.orc";
     final int numInserts = 3;
     try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) {
@@ -419,7 +418,7 @@ public void testOrphanFilesDeletionBackupDisabled() throws Exception {
 
   @Test
   public void testOrphanFilesDeletionDeleteDataWhenDataManifestNotExists() throws Exception {
-    final String tableName = "db.test_ofd_java";
+    final String tableName = uniqueTableName("db.test_ofd_java");
     final String testOrphanFileName = "data/test_orphan_file.orc";
     final int numInserts = 3;
     try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) {
@@ -444,7 +443,7 @@ public void testOrphanFilesDeletionDeleteDataWhenDataManifestNotExists() throws
 
   @Test
   public void testSnapshotsExpirationMaxAge() throws Exception {
-    final String tableName = "db.test_es_maxage_java";
+    final String tableName = uniqueTableName("db.test_es_maxage_java");
     final int numInserts = 3;
     final int maxAge = 0;
     final String timeGranularity = "DAYS";
@@ -475,7 +474,7 @@ public void testSnapshotsExpirationMaxAge() throws Exception {
 
   @Test
   public void testSnapshotsExpirationMaxAgeNoop() throws Exception {
-    final String tableName = "db.test_es_maxage_noop_java";
+    final String tableName = uniqueTableName("db.test_es_maxage_noop_java");
     final int numInserts = 3;
     final int maxAge = 3;
     final String timeGranularity = "DAYS";
@@ -505,7 +504,7 @@ public void testSnapshotsExpirationMaxAgeNoop() throws Exception {
 
   @Test
   public void testSnapshotsExpirationVersionsNoop() throws Exception {
-    final String tableName = "db.test_es_versions_noop_java";
+    final String tableName = uniqueTableName("db.test_es_versions_noop_java");
     final int numInserts = 3;
     final int versionsToKeep = 5; // Should keep all versions given that there are fewer versions
     final int maxAge = 3;
@@ -535,7 +534,7 @@ public void testSnapshotsExpirationVersionsNoop() throws Exception {
 
   @Test
   public void testSnapshotsExpirationVersions() throws Exception {
-    final String tableName = "db.test_es_versions_java";
+    final String tableName = uniqueTableName("db.test_es_versions_java");
     final int numInserts = 3;
     final int versionsToKeep = 2;
     final int maxAge = 3;
@@ -569,7 +568,7 @@ public void testSnapshotsExpirationVersions() throws Exception {
 
   @Test
   public void testSnapshotsExpirationBothAgeAndVersions() throws Exception {
-    final String tableName = "db.test_es_age_and_versions_java";
+    final String tableName = uniqueTableName("db.test_es_age_and_versions_java");
     final int numInserts = 3;
     final int maxAge = 3;
     final String timeGranularity = "DAYS";
@@ -603,7 +602,7 @@ public void testSnapshotsExpirationBothAgeAndVersions() throws Exception {
 
   @Test
   public void testSnapshotsExpirationPrioritizeAge() throws Exception {
-    final String tableName = "db.test_es_age_prioritization_java";
+    final String tableName = uniqueTableName("db.test_es_age_prioritization_java");
     final int numInserts = 3;
     final int maxAge = 20;
     final String timeGranularity =
@@ -644,7 +643,7 @@ public void testSnapshotsExpirationPrioritizeAge() throws Exception {
 
   @Test
   public void testSnapshotExpirationWithHoursDaysMonthsYears() throws Exception {
-    final String tableName = "db.test_es_age_policy";
+    final String tableName = uniqueTableName("db.test_es_age_policy");
     final int numInserts = 3;
     final int maxAge = 20;
     final int versionsToKeep = 5;
@@ -676,7 +675,7 @@ public void testSnapshotExpirationWithHoursDaysMonthsYears() throws Exception {
 
   @Test
   public void testStagedFilesDelete() throws Exception {
-    final String tableName = "db.test_staged_delete";
+    final String tableName = uniqueTableName("db.test_staged_delete");
     final int numInserts = 3;
     final String testOrphanFile1 = "data/test_orphan_file1.orc";
     final String testOrphanFile2 = "data/test_orphan_file2.orc";
@@ -714,7 +713,7 @@ public void testStagedFilesDelete() throws Exception {
 
   @Test
   public void testDataCompactionPartialProgressNonPartitionedTable() throws Exception {
-    final String tableName = "db.test_data_compaction";
+    final String tableName = uniqueTableName("db.test_data_compaction");
     final int numInserts = 3;
 
     BiFunction<Operations, Table, RewriteDataFiles.Result> rewriteFunc =
@@ -771,7 +770,7 @@ public void testDataCompactionPartialProgressNonPartitionedTable() throws Except
 
   @Test
   public void testDataCompactionPartialProgressPartitionedTable() throws Exception {
-    final String tableName = "db.test_data_compaction_partitioned";
+    final String tableName = uniqueTableName("db.test_data_compaction_partitioned");
     final int numInsertsPerPartition = 3;
     final int numDailyPartitions = 10;
     final int maxCommits = 5;
@@ -857,7 +856,7 @@ public void testDataCompactionPartialProgressPartitionedTable() throws Exception
   public void testOrphanDirsDeletionJavaAPI() throws Exception {
     try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) {
       // test orphan delete
-      Path tbLoc = prepareOrphanTableDirectory(ops, "db1.test_odd_orphan");
+      Path tbLoc = prepareOrphanTableDirectory(ops, uniqueTableName("db1.test_odd_orphan"));
 
       long timeThreshold = System.currentTimeMillis();
       List<Path> matchingFilesBefore = new ArrayList<>();
@@ -912,7 +911,7 @@ private static Path prepareOrphanTableDirectory(Operations ops, String tableName
 
   @Test
   public void testCollectEarliestPartitionDateStat() throws Exception {
-    final String tableName = "db.test_collect_earliest_partition_date";
+    final String tableName = uniqueTableName("db.test_collect_earliest_partition_date");
     List<String> rowValue = new ArrayList<>();
 
     try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) {
@@ -958,7 +957,7 @@ public void testCollectEarliestPartitionDateStat() throws Exception {
 
   @Test
   public void testCollectTableStatsWithEmptyPartitions() throws Exception {
-    final String tableName = "db.test_empty_partitions";
+    final String tableName = uniqueTableName("db.test_empty_partitions");
 
     try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) {
       // Create table with partition but no data
@@ -980,7 +979,7 @@ public void testCollectTableStatsWithEmptyPartitions() throws Exception {
 
   @Test
   public void testCollectTablePolicyStats() throws Exception {
-    final String tableName = "db.test_collect_table_stats_with_policy";
+    final String tableName = uniqueTableName("db.test_collect_table_stats_with_policy");
     List<String> rowValue = new ArrayList<>();
 
     try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) {
@@ -1050,7 +1049,7 @@ public void testCollectTablePolicyStats() throws Exception {
 
   @Test
   public void testCollectTableStats() throws Exception {
-    final String tableName = "db.test_collect_table_stats";
+    final String tableName = uniqueTableName("db.test_collect_table_stats");
     final int numInserts = 3;
     try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) {
       prepareTable(ops, tableName);
@@ -1096,7 +1095,8 @@ public void testCollectTableStats() throws Exception {
 
   @Test
   public void testCollectHistoryPolicyStatsWithSnapshots() throws Exception {
-    final String tableName = "db.test_collect_table_stats_with_history_policy_snapshots";
+    final String tableName =
+        uniqueTableName("db.test_collect_table_stats_with_history_policy_snapshots");
 
     try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) {
       // Test table with both retention and sharing policies
@@ -1139,6 +1139,10 @@ private static void verifyRowCount(Operations ops, String tableName, int expecte
     Assertions.assertEquals(expectedRowCount, resultRows.size());
   }
 
+  private static String uniqueTableName(String tableName) {
+    return tableName + "_" + UUID.randomUUID().toString().replace("-", "");
+  }
+
   private static void populateTable(
       Operations ops, String tableName, int numRows, int dayLag, long timestampSeconds) {
     String timestampEntry =
@@ -1396,7 +1400,7 @@ private static List<Triple<String, String, Long>> getDataFiles(Operations ops, S
 
   @Test
   public void testCollectPartitionStatsForPartitionedTable() throws Exception {
-    final String tableName = "db.test_partition_stats_partitioned";
+    final String tableName = uniqueTableName("db.test_partition_stats_partitioned");
 
     try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) {
       // Setup: Create partitioned table
@@ -1438,7 +1442,7 @@ public void testCollectPartitionStatsForPartitionedTable() throws Exception {
 
   @Test
   public void testCollectPartitionStatsForUnpartitionedTable() throws Exception {
-    final String tableName = "db.test_partition_stats_unpartitioned";
+    final String tableName = uniqueTableName("db.test_partition_stats_unpartitioned");
 
     try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) {
       // Setup: Create unpartitioned table
@@ -1478,7 +1482,7 @@ public void testCollectPartitionStatsForUnpartitionedTable() throws Exception {
 
   @Test
   public void testCollectPartitionStatsWithMultipleCommitsToSamePartition() throws Exception {
-    final String tableName = "db.test_partition_stats_multiple_commits";
+    final String tableName = uniqueTableName("db.test_partition_stats_multiple_commits");
 
     try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) {
       // Setup: Create partitioned table
@@ -1516,7 +1520,7 @@ public void testCollectPartitionStatsWithMultipleCommitsToSamePartition() throws
 
   @Test
   public void testCollectPartitionStatsEmptyTable() throws Exception {
-    final String tableName = "db.test_partition_stats_empty";
+    final String tableName = uniqueTableName("db.test_partition_stats_empty");
 
     try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) {
       // Setup: Create empty partitioned table (no data)