diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ChainTablePartitionExpire.java b/paimon-core/src/main/java/org/apache/paimon/operation/ChainTablePartitionExpire.java index 0e70854363ad..4d20908413e8 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/ChainTablePartitionExpire.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/ChainTablePartitionExpire.java @@ -352,11 +352,9 @@ private void batchDropPartitions( for (List> batch : Lists.partition(partitionSpecs, expireBatchSize)) { partitionModification.dropPartitions(batch); - partitionModification.dropPartitions(toDonePartitions(batch)); } } else { partitionModification.dropPartitions(partitionSpecs); - partitionModification.dropPartitions(toDonePartitions(partitionSpecs)); } } catch (Catalog.TableNotExistException e) { throw new RuntimeException(e); @@ -373,23 +371,6 @@ private void batchDropPartitions( } } - private List> toDonePartitions( - List> expiredPartitions) { - List> donePartitions = new ArrayList<>(expiredPartitions.size()); - for (Map partition : expiredPartitions) { - LinkedHashMap donePartition = new LinkedHashMap<>(partition); - Map.Entry lastEntry = null; - for (Map.Entry entry : donePartition.entrySet()) { - lastEntry = entry; - } - if (lastEntry != null) { - donePartition.put(lastEntry.getKey(), lastEntry.getValue() + ".done"); - donePartitions.add(donePartition); - } - } - return donePartitions; - } - private Map> groupByGroupKey(List partitionEntries) { Map> grouped = new LinkedHashMap<>(); for (PartitionEntry entry : partitionEntries) { diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/NormalPartitionExpire.java b/paimon-core/src/main/java/org/apache/paimon/operation/NormalPartitionExpire.java index aaceb6a17afa..6abffab58d7e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/NormalPartitionExpire.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/NormalPartitionExpire.java @@ -39,7 +39,6 @@ import java.util.ArrayList; import java.util.Collection; import java.util.Comparator; -import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.concurrent.ThreadLocalRandom; @@ -198,36 +197,14 @@ private void doBatchExpire( if (partitionModification != null) { try { partitionModification.dropPartitions(expiredBatchPartitions); - // also drop corresponding .done partitions - partitionModification.dropPartitions(toDonePartitions(expiredBatchPartitions)); } catch (Catalog.TableNotExistException e) { throw new RuntimeException(e); } } else { - // .done partitions only exist when partitionModification != null - // (metastore.partitioned-table = true), so no need to handle them here commit.dropPartitions(expiredBatchPartitions, commitIdentifier); } } - private List> toDonePartitions( - List> expiredPartitions) { - List> donePartitions = new ArrayList<>(expiredPartitions.size()); - for (Map partition : expiredPartitions) { - LinkedHashMap donePartition = new LinkedHashMap<>(partition); - // append .done suffix to the last partition field value - Map.Entry lastEntry = null; - for (Map.Entry entry : donePartition.entrySet()) { - lastEntry = entry; - } - if (lastEntry != null) { - donePartition.put(lastEntry.getKey(), lastEntry.getValue() + ".done"); - donePartitions.add(donePartition); - } - } - return donePartitions; - } - private List> convertToPartitionString( List> expiredPartValues) { // Sort on the joined form, but build the partition from the original values: a value diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/ChainTablePartitionExpireTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/ChainTablePartitionExpireTest.java index ac7176dcbf82..03e196e7c866 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/ChainTablePartitionExpireTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/ChainTablePartitionExpireTest.java @@ -889,6 +889,16 @@ public void dropPartitions(List> partitions) throws Catalog.TableNotExistException { for (Map partition : partitions) { droppedPartitions.add(new HashMap<>(partition)); + // Model the Hive metastore's cleanup of the corresponding marker. + Map donePartition = new HashMap<>(partition); + String lastKey = null; + for (String key : partition.keySet()) { + lastKey = key; + } + if (lastKey != null) { + donePartition.put(lastKey, partition.get(lastKey) + ".done"); + droppedPartitions.add(donePartition); + } } } diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java index e0a2372287f8..4b07a3fbbf46 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java @@ -139,12 +139,7 @@ public void createPartitions(List> partitions) @Override public void dropPartitions(List> partitions) throws Catalog.TableNotExistException { - for (Map partition : partitions) { - // only record partitions that were created - if (createdPartitions.remove(partition)) { - deletedPartitions.add(partition); - } - } + dropPartitionMetadata(partitions); try (FileStoreCommit commit = table.store() .newCommit( @@ -153,6 +148,27 @@ public void dropPartitions(List> partitions) null)) { commit.dropPartitions(partitions, BatchWriteBuilder.COMMIT_IDENTIFIER); } + // Model HiveCatalog's cleanup of markers after deleting ordinary data. + List> donePartitions = new ArrayList<>(); + String lastKey = + tableSchema + .partitionKeys() + .get(tableSchema.partitionKeys().size() - 1); + for (Map partition : partitions) { + Map donePartition = new HashMap<>(partition); + donePartition.put(lastKey, partition.get(lastKey) + ".done"); + donePartitions.add(donePartition); + } + dropPartitionMetadata(donePartitions); + } + + private void dropPartitionMetadata(List> partitions) { + for (Map partition : partitions) { + // only record partitions that were created + if (createdPartitions.remove(partition)) { + deletedPartitions.add(partition); + } + } } @Override diff --git a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java index 94e561264428..e2ca397882fd 100644 --- a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java +++ b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java @@ -35,6 +35,7 @@ import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; import org.apache.paimon.hive.pool.CachedClientPool; +import org.apache.paimon.operation.FileStoreScan; import org.apache.paimon.operation.Lock; import org.apache.paimon.options.CatalogOptions; import org.apache.paimon.options.Options; @@ -488,35 +489,72 @@ private void createPartitionsUtil( public void dropPartitions(Identifier identifier, List> partitions) throws TableNotExistException { TableSchema schema = this.loadTableSchema(identifier); - CoreOptions options = CoreOptions.fromMap(schema.options()); - boolean tagToPart = options.tagToPartitionField() != null; - if (metastorePartitioned(schema)) { - List> metaPartitions = + boolean tagToPart = CoreOptions.fromMap(schema.options()).tagToPartitionField() != null; + boolean metastorePartitioned = metastorePartitioned(schema); + if (metastorePartitioned) { + dropPartitionsFromMetastore( + identifier, tagToPart ? partitions - : removePartitionsExistsInOtherBranches(identifier, partitions); - for (Map part : metaPartitions) { - List partitionValues = - new ArrayList<>( - orderByPartitionKeys(schema.partitionKeys(), part).values()); - try { - clients() - .execute( - client -> - client.dropPartition( - identifier.getDatabaseName(), - identifier.getTableName(), - partitionValues, - false)); - } catch (NoSuchObjectException e) { - // do nothing if the partition not exists - } catch (Exception e) { - throw new RuntimeException(e); - } - } + : removePartitionsStillInUse(identifier, partitions, false), + schema); } if (!tagToPart) { super.dropPartitions(identifier, partitions); + if (metastorePartitioned) { + dropDonePartitions(identifier, partitions, schema); + } + } + } + + private void dropDonePartitions( + Identifier identifier, List> partitions, TableSchema schema) + throws TableNotExistException { + List> donePartitions = new ArrayList<>(partitions.size()); + String lastKey = schema.partitionKeys().get(schema.partitionKeys().size() - 1); + for (Map partition : partitions) { + Map donePartition = new LinkedHashMap<>(partition); + donePartition.put(lastKey, partition.get(lastKey) + ".done"); + donePartitions.add(donePartition); + } + + // HMS partition names are shared by all branches. A marker is still needed if a branch + // owns its ordinary partition or a real STRING partition with the same literal spec. + Set> specsToCheck = new HashSet<>(partitions); + specsToCheck.addAll(donePartitions); + Set> removable = + new HashSet<>( + removePartitionsStillInUse( + identifier, new ArrayList<>(specsToCheck), true)); + List> markersToDrop = new ArrayList<>(); + for (int i = 0; i < partitions.size(); i++) { + if (removable.contains(partitions.get(i)) + && removable.contains(donePartitions.get(i))) { + markersToDrop.add(donePartitions.get(i)); + } + } + dropPartitionsFromMetastore(identifier, markersToDrop, schema); + } + + private void dropPartitionsFromMetastore( + Identifier identifier, List> partitions, TableSchema schema) { + for (Map part : partitions) { + List partitionValues = + new ArrayList<>(orderByPartitionKeys(schema.partitionKeys(), part).values()); + try { + clients() + .execute( + client -> + client.dropPartition( + identifier.getDatabaseName(), + identifier.getTableName(), + partitionValues, + false)); + } catch (NoSuchObjectException e) { + // do nothing if the partition not exists + } catch (Exception e) { + throw new RuntimeException(e); + } } } @@ -753,26 +791,23 @@ public List listPartitionsFromHms(Identifier identifier) Short.MAX_VALUE)); } - private List> removePartitionsExistsInOtherBranches( - Identifier identifier, List> inputs) throws TableNotExistException { + private List> removePartitionsStillInUse( + Identifier identifier, List> inputs, boolean syntheticPartitions) + throws TableNotExistException { FileStoreTable mainTable = (FileStoreTable) getTable( new Identifier( identifier.getDatabaseName(), identifier.getTableName())); - InternalRowPartitionComputer partitionComputer = - new InternalRowPartitionComputer( - mainTable.coreOptions().partitionDefaultName(), - mainTable.rowType().project(mainTable.partitionKeys()), - mainTable.partitionKeys().toArray(new String[0]), - mainTable.coreOptions().legacyPartitionName()); List branchNames = new ArrayList<>(mainTable.branchManager().branches()); branchNames.add(DEFAULT_MAIN_BRANCH); Set> inputsToRemove = new HashSet<>(inputs); for (String branchName : branchNames) { - if (branchName.equals(identifier.getBranchNameOrDefault())) { + // Ordinary partitions are checked before their data is dropped. Marker cleanup runs + // afterwards and must also protect real, colliding partitions on the current branch. + if (!syntheticPartitions && branchName.equals(identifier.getBranchNameOrDefault())) { continue; } @@ -782,8 +817,22 @@ private List> removePartitionsExistsInOtherBranches( continue; } - branchTableForPartitionExistence(mainTable, branchName).newScan() - .withPartitionsFilter(new ArrayList<>(inputsToRemove)).listPartitions().stream() + FileStoreTable branchTable = branchTableForPartitionExistence(mainTable, branchName); + InternalRowPartitionComputer partitionComputer = + new InternalRowPartitionComputer( + branchTable.coreOptions().partitionDefaultName(), + branchTable.rowType().project(branchTable.partitionKeys()), + branchTable.partitionKeys().toArray(new String[0]), + branchTable.coreOptions().legacyPartitionName()); + // Ownership depends on the latest physical snapshot. A table batch scan may instead + // follow persisted query selectors such as scan.snapshot-id or scan.tag-name. + FileStoreScan scan = branchTable.store().newScan(); + if (!syntheticPartitions) { + scan.withPartitionsFilter(new ArrayList<>(inputsToRemove)); + } + // Synthetic values need not fit the partition type. Compare their literal specs with + // actual branch partitions, including real STRING values that collide with markers. + scan.listPartitions().stream() .map(partitionComputer::generatePartValues) .forEach(inputsToRemove::remove); } diff --git a/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogTest.java b/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogTest.java index 9e22b024e483..ed972cfa3942 100644 --- a/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogTest.java +++ b/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogTest.java @@ -32,6 +32,7 @@ import org.apache.paimon.options.Options; import org.apache.paimon.partition.Partition; import org.apache.paimon.partition.PartitionStatistics; +import org.apache.paimon.partition.actions.AddDonePartitionAction; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaChange; import org.apache.paimon.schema.SchemaManager; @@ -59,6 +60,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.ValueSource; import java.lang.reflect.Field; @@ -833,6 +835,575 @@ public void testCreateAndDropPartitionsWithSpecNotInPartitionKeyOrder() throws E .endsWith("/dt=d9/hh=h9"); } + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testExpireIntegerPartitionsWithDoneMarkers(boolean markDone) throws Exception { + String databaseName = "test_expire_integer_partitions"; + String tableName = "t"; + catalog.createDatabase(databaseName, false); + Identifier identifier = Identifier.create(databaseName, tableName); + catalog.createTable( + identifier, + Schema.newBuilder() + .option(METASTORE_PARTITIONED_TABLE.key(), "true") + .option(PARTITION_TIMESTAMP_FORMATTER.key(), "yyyyMMdd") + .option(PARTITION_TIMESTAMP_PATTERN.key(), "$dt") + .column("col", DataTypes.INT()) + .column("dt", DataTypes.INT()) + .partitionKeys("dt") + .build(), + false); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = writeBuilder.newWrite(); + BatchTableCommit commit = writeBuilder.newCommit()) { + write.write(GenericRow.of(1, 20230101)); + write.write(GenericRow.of(2, 20990101)); + commit.commit(write.prepareCommit()); + } + if (markDone) { + AddDonePartitionAction action = + new AddDonePartitionAction(table.catalogEnvironment().partitionModification()); + action.markDone("dt=20230101"); + action.markDone("dt=20990101"); + } + Map expirationOptions = new HashMap<>(); + expirationOptions.put(CoreOptions.PARTITION_EXPIRATION_TIME.key(), "1 d"); + expirationOptions.put(CoreOptions.PARTITION_EXPIRATION_CHECK_INTERVAL.key(), "0 s"); + table = table.copy(expirationOptions); + long snapshotId = table.snapshotManager().latestSnapshotId(); + table.store().newPartitionExpire("expire", table).expire(Long.MAX_VALUE); + + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId + 1); + assertThat(table.newScan().listPartitions()) + .extracting(partition -> partition.getInt(0)) + .containsExactly(20990101); + if (markDone) { + assertHmsDts(databaseName, tableName, "20990101", "20990101.done"); + } else { + assertHmsDts(databaseName, tableName, "20990101"); + } + } + + @Test + public void testDropPartitionsCleansDoneMarkersInPartitionKeyOrder() throws Exception { + String databaseName = "test_drop_multi_key_done_markers"; + String tableName = "t"; + catalog.createDatabase(databaseName, false); + Identifier identifier = Identifier.create(databaseName, tableName); + catalog.createTable( + identifier, + Schema.newBuilder() + .option(METASTORE_PARTITIONED_TABLE.key(), "true") + .column("col", DataTypes.INT()) + .column("dt", DataTypes.STRING()) + .column("hh", DataTypes.INT()) + .partitionKeys("dt", "hh") + .build(), + false); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = writeBuilder.newWrite(); + BatchTableCommit commit = writeBuilder.newCommit()) { + write.write(GenericRow.of(1, BinaryString.fromString("20230101"), 1)); + write.write(GenericRow.of(2, BinaryString.fromString("20230101.done"), 1)); + commit.commit(write.prepareCommit()); + } + table.createBranch("empty"); + AddDonePartitionAction action = + new AddDonePartitionAction(table.catalogEnvironment().partitionModification()); + action.markDone("dt=20230101/hh=1"); + action.markDone("dt=20230101.done/hh=1"); + long snapshotId = table.snapshotManager().latestSnapshotId(); + + catalog.dropPartitions( + identifier, Collections.singletonList(orderedSpec("hh", "1", "dt", "20230101"))); + + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId + 1); + assertThat(table.newScan().listPartitions()) + .extracting(partition -> partition.getString(0).toString()) + .containsExactly("20230101.done"); + assertThat(((HiveCatalog) catalog).listPartitionsFromHms(identifier)) + .extracting(partition -> partition.getValues()) + .containsExactlyInAnyOrder( + Arrays.asList("20230101.done", "1"), + Arrays.asList("20230101.done", "1.done")); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testExpireIntegerPartitionsWithEmptyBranch(boolean markDone) throws Exception { + String databaseName = "test_expire_integer_partitions"; + String tableName = "t"; + catalog.createDatabase(databaseName, false); + Identifier identifier = Identifier.create(databaseName, tableName); + catalog.createTable( + identifier, + Schema.newBuilder() + .option(METASTORE_PARTITIONED_TABLE.key(), "true") + .option(PARTITION_TIMESTAMP_FORMATTER.key(), "yyyyMMdd") + .option(PARTITION_TIMESTAMP_PATTERN.key(), "$dt") + .column("col", DataTypes.INT()) + .column("dt", DataTypes.INT()) + .partitionKeys("dt") + .build(), + false); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = writeBuilder.newWrite(); + BatchTableCommit commit = writeBuilder.newCommit()) { + write.write(GenericRow.of(1, 20230101)); + write.write(GenericRow.of(2, 20990101)); + commit.commit(write.prepareCommit()); + } + if (markDone) { + AddDonePartitionAction action = + new AddDonePartitionAction(table.catalogEnvironment().partitionModification()); + action.markDone("dt=20230101"); + action.markDone("dt=20990101"); + } + Map expirationOptions = new HashMap<>(); + expirationOptions.put(CoreOptions.PARTITION_EXPIRATION_TIME.key(), "1 d"); + expirationOptions.put(CoreOptions.PARTITION_EXPIRATION_CHECK_INTERVAL.key(), "0 s"); + table.createBranch("empty"); + assertThat(table.switchToBranch("empty").newScan().listPartitions()).isEmpty(); + table = table.copy(expirationOptions); + long snapshotId = table.snapshotManager().latestSnapshotId(); + table.store().newPartitionExpire("expire", table).expire(Long.MAX_VALUE); + + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId + 1); + assertThat(table.newScan().listPartitions()) + .extracting(partition -> partition.getInt(0)) + .containsExactly(20990101); + if (markDone) { + assertHmsDts(databaseName, tableName, "20990101", "20990101.done"); + } else { + assertHmsDts(databaseName, tableName, "20990101"); + } + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testExpireIntegerPartitionsSharedWithOtherBranch(boolean markDone) + throws Exception { + String databaseName = "test_expire_shared_integer_partitions"; + String tableName = "t"; + catalog.createDatabase(databaseName, false); + Identifier identifier = Identifier.create(databaseName, tableName); + catalog.createTable( + identifier, + Schema.newBuilder() + .option(METASTORE_PARTITIONED_TABLE.key(), "true") + .option(PARTITION_TIMESTAMP_FORMATTER.key(), "yyyyMMdd") + .option(PARTITION_TIMESTAMP_PATTERN.key(), "$dt") + .column("col", DataTypes.INT()) + .column("dt", DataTypes.INT()) + .partitionKeys("dt") + .build(), + false); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = writeBuilder.newWrite(); + BatchTableCommit commit = writeBuilder.newCommit()) { + write.write(GenericRow.of(1, 20230101)); + write.write(GenericRow.of(2, 20230102)); + write.write(GenericRow.of(3, 20990101)); + commit.commit(write.prepareCommit()); + } + table.createBranch("other"); + FileStoreTable other = table.switchToBranch("other"); + BatchWriteBuilder otherWriteBuilder = other.newBatchWriteBuilder(); + try (BatchTableWrite write = otherWriteBuilder.newWrite(); + BatchTableCommit commit = otherWriteBuilder.newCommit()) { + write.write(GenericRow.of(2, 20230102)); + commit.commit(write.prepareCommit()); + } + if (markDone) { + AddDonePartitionAction action = + new AddDonePartitionAction(table.catalogEnvironment().partitionModification()); + action.markDone("dt=20230101"); + action.markDone("dt=20230102"); + action.markDone("dt=20990101"); + } + Map expirationOptions = new HashMap<>(); + expirationOptions.put(CoreOptions.PARTITION_EXPIRATION_TIME.key(), "1 d"); + expirationOptions.put(CoreOptions.PARTITION_EXPIRATION_CHECK_INTERVAL.key(), "0 s"); + table = table.copy(expirationOptions); + table.store().newPartitionExpire("expire-main", table).expire(Long.MAX_VALUE); + + assertThat(table.newScan().listPartitions()) + .extracting(partition -> partition.getInt(0)) + .containsExactly(20990101); + assertThat(other.newScan().listPartitions()) + .extracting(partition -> partition.getInt(0)) + .containsExactly(20230102); + if (markDone) { + assertHmsDts( + databaseName, + tableName, + "20230102", + "20230102.done", + "20990101", + "20990101.done"); + } else { + assertHmsDts(databaseName, tableName, "20230102", "20990101"); + } + + // Only the last branch holding a partition may remove its shared HMS entry and marker. + other = other.copy(expirationOptions); + other.store().newPartitionExpire("expire-other", other).expire(Long.MAX_VALUE); + assertThat(other.newScan().listPartitions()).isEmpty(); + assertThat(table.newScan().listPartitions()) + .extracting(partition -> partition.getInt(0)) + .containsExactly(20990101); + if (markDone) { + assertHmsDts(databaseName, tableName, "20990101", "20990101.done"); + } else { + assertHmsDts(databaseName, tableName, "20990101"); + } + } + + @Test + public void testDropRealDoneSuffixPartitionAndItsMarker() throws Exception { + String databaseName = "test_drop_partition_metadata"; + String tableName = "t"; + catalog.createDatabase(databaseName, false); + Identifier identifier = Identifier.create(databaseName, tableName); + catalog.createTable( + identifier, + Schema.newBuilder() + .option(METASTORE_PARTITIONED_TABLE.key(), "true") + .column("col", DataTypes.INT()) + .column("dt", DataTypes.STRING()) + .partitionKeys("dt") + .build(), + false); + writeAppendPartition(identifier, "20230101.done"); + writeAppendPartition(identifier, "20990101"); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + AddDonePartitionAction action = + new AddDonePartitionAction(table.catalogEnvironment().partitionModification()); + action.markDone("dt=20230101.done"); + long snapshotId = table.snapshotManager().latestSnapshotId(); + + // A real STRING partition ending in .done must still support ordinary data deletion. + catalog.dropPartitions(identifier, partitionSpecs("20230101.done")); + + assertHmsDts(databaseName, tableName, "20990101"); + assertPhysicalDts(identifier, "20990101"); + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId + 1); + } + + @Test + public void testRealDoneSuffixPartitionKeepsBranchProtection() throws Exception { + String databaseName = "test_real_done_suffix_partition"; + String tableName = "t"; + catalog.createDatabase(databaseName, false); + Identifier identifier = Identifier.create(databaseName, tableName); + catalog.createTable( + identifier, + Schema.newBuilder() + .option(METASTORE_PARTITIONED_TABLE.key(), "true") + .column("col", DataTypes.INT()) + .column("dt", DataTypes.STRING()) + .partitionKeys("dt") + .build(), + false); + writeAppendPartition(identifier, "20230101.done"); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + table.createBranch("other"); + Identifier otherId = new Identifier(databaseName, tableName, "other"); + writeAppendPartition(otherId, "20230101.done"); + AddDonePartitionAction action = + new AddDonePartitionAction(table.catalogEnvironment().partitionModification()); + action.markDone("dt=20230101.done"); + long snapshotId = table.snapshotManager().latestSnapshotId(); + + catalog.dropPartitions(identifier, partitionSpecs("20230101.done")); + assertPhysicalDts(identifier); + assertPhysicalDts(otherId, "20230101.done"); + assertHmsDts(databaseName, tableName, "20230101.done", "20230101.done.done"); + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId + 1); + + catalog.dropPartitions(otherId, partitionSpecs("20230101.done")); + assertPhysicalDts(otherId); + assertHmsDts(databaseName, tableName); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testTtlKeepsOtherBranchRealDoneSuffixPartition(boolean markDone) throws Exception { + String databaseName = "test_ttl_done_name_collision"; + String tableName = "t"; + catalog.createDatabase(databaseName, false); + Identifier identifier = Identifier.create(databaseName, tableName); + catalog.createTable( + identifier, + Schema.newBuilder() + .option(METASTORE_PARTITIONED_TABLE.key(), "true") + .option(PARTITION_TIMESTAMP_FORMATTER.key(), "yyyyMMdd") + .option(PARTITION_TIMESTAMP_PATTERN.key(), "$dt") + .column("col", DataTypes.INT()) + .column("dt", DataTypes.STRING()) + .partitionKeys("dt") + .build(), + false); + writeAppendPartition(identifier, "20230101"); + writeAppendPartition(identifier, "20990101"); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + table.createBranch("other"); + Identifier otherId = new Identifier(databaseName, tableName, "other"); + writeAppendPartition(otherId, "20230101.done"); + FileStoreTable other = (FileStoreTable) catalog.getTable(otherId); + long otherSnapshot = other.snapshotManager().latestSnapshotId(); + if (markDone) { + AddDonePartitionAction action = + new AddDonePartitionAction(table.catalogEnvironment().partitionModification()); + action.markDone("dt=20230101"); + } + Map expirationOptions = new HashMap<>(); + expirationOptions.put(CoreOptions.PARTITION_EXPIRATION_TIME.key(), "1 d"); + expirationOptions.put(CoreOptions.PARTITION_EXPIRATION_CHECK_INTERVAL.key(), "0 s"); + table = table.copy(expirationOptions); + table.store().newPartitionExpire("expire-main", table).expire(Long.MAX_VALUE); + + assertPhysicalDts(identifier, "20990101"); + assertPhysicalDts(otherId, "20230101.done"); + assertThat(other.snapshotManager().latestSnapshotId()).isEqualTo(otherSnapshot); + assertHmsDts(databaseName, tableName, "20990101", "20230101.done"); + } + + @ParameterizedTest + @CsvSource({"0,false", "0,true", "1,false", "1,true"}) + public void testExpireIntegerChainPartitionsWithDoneMarkers(int batchSize, boolean markDone) + throws Exception { + String databaseName = "test_integer_chain_done_expire"; + String tableName = "t"; + catalog.createDatabase(databaseName, false); + Identifier identifier = Identifier.create(databaseName, tableName); + catalog.createTable( + identifier, + Schema.newBuilder() + .option(METASTORE_PARTITIONED_TABLE.key(), "true") + .option(BUCKET.key(), "1") + .option(SEQUENCE_FIELD.key(), "v") + .option(FILE_FORMAT.key(), "avro") + .option(FILE_COMPRESSION.key(), "snappy") + .option(MANIFEST_COMPRESSION.key(), "snappy") + .column("dt", DataTypes.INT()) + .column("pk", DataTypes.STRING()) + .column("v", DataTypes.STRING()) + .partitionKeys("dt") + .primaryKey("pk", "dt") + .build(), + false); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + table.createBranch("snapshot"); + table.createBranch("delta"); + Identifier snapshotId = new Identifier(databaseName, tableName, "snapshot"); + Identifier deltaId = new Identifier(databaseName, tableName, "delta"); + List chainOptions = + Arrays.asList( + SchemaChange.setOption(CHAIN_TABLE_ENABLED.key(), "true"), + SchemaChange.setOption(SCAN_FALLBACK_SNAPSHOT_BRANCH.key(), "snapshot"), + SchemaChange.setOption(SCAN_FALLBACK_DELTA_BRANCH.key(), "delta"), + SchemaChange.setOption(PARTITION_TIMESTAMP_PATTERN.key(), "$dt"), + SchemaChange.setOption(PARTITION_TIMESTAMP_FORMATTER.key(), "yyyyMMdd")); + for (Identifier branchId : Arrays.asList(identifier, snapshotId, deltaId)) { + catalog.alterTable(branchId, chainOptions, false); + } + writeIntegerChainPartitions(snapshotId, 20230101, 20230201, 20230301); + writeIntegerChainPartitions(deltaId, 20230110, 20230210, 20230310); + if (markDone) { + AddDonePartitionAction action = + new AddDonePartitionAction( + ((FileStoreTable) catalog.getTable(identifier)) + .catalogEnvironment() + .partitionModification()); + for (int dt : new int[] {20230101, 20230201, 20230301, 20230110, 20230210, 20230310}) { + action.markDone("dt=" + dt); + } + } + table = (FileStoreTable) catalog.getTable(identifier); + Map expirationOptions = new HashMap<>(); + expirationOptions.put(CoreOptions.PARTITION_EXPIRATION_TIME.key(), "1 d"); + expirationOptions.put(CoreOptions.PARTITION_EXPIRATION_CHECK_INTERVAL.key(), "0 s"); + if (batchSize > 0) { + expirationOptions.put( + CoreOptions.PARTITION_EXPIRATION_BATCH_SIZE.key(), String.valueOf(batchSize)); + } + table = table.copy(expirationOptions); + table.store().newPartitionExpire("expire-chain", table).expire(Long.MAX_VALUE); + + assertThat( + ((FileStoreTable) catalog.getTable(snapshotId)) + .newSnapshotReader() + .partitionEntries()) + .extracting(entry -> entry.partition().getInt(0)) + .containsExactly(20230301); + assertThat( + ((FileStoreTable) catalog.getTable(deltaId)) + .newSnapshotReader() + .partitionEntries()) + .extracting(entry -> entry.partition().getInt(0)) + .containsExactly(20230310); + if (markDone) { + assertHmsDts( + databaseName, + tableName, + "20230301", + "20230301.done", + "20230310", + "20230310.done"); + } else { + assertHmsDts(databaseName, tableName, "20230301", "20230310"); + } + } + + @Test + public void testMarkerCleanupKeepsCurrentBranchRealDoneSuffixPartition() throws Exception { + String databaseName = "test_current_branch_done_collision"; + String tableName = "t"; + catalog.createDatabase(databaseName, false); + Identifier identifier = Identifier.create(databaseName, tableName); + catalog.createTable( + identifier, + Schema.newBuilder() + .option(METASTORE_PARTITIONED_TABLE.key(), "true") + .column("col", DataTypes.INT()) + .column("dt", DataTypes.STRING()) + .partitionKeys("dt") + .build(), + false); + writeAppendPartition(identifier, "20230101"); + writeAppendPartition(identifier, "20230101.done"); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + long snapshotId = table.snapshotManager().latestSnapshotId(); + + catalog.dropPartitions(identifier, partitionSpecs("20230101")); + + assertHmsDts(databaseName, tableName, "20230101.done"); + assertPhysicalDts(identifier, "20230101.done"); + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId + 1); + } + + @ParameterizedTest + @CsvSource({ + "scan.snapshot-id,false", + "scan.snapshot-id,true", + "scan.tag-name,false", + "scan.tag-name,true" + }) + public void testMarkerCleanupIgnoresHistoricalReads(String selector, boolean otherBranch) + throws Exception { + String databaseName = "test_marker_cleanup_historical_reads"; + String tableName = "t"; + catalog.createDatabase(databaseName, false); + Identifier identifier = Identifier.create(databaseName, tableName); + catalog.createTable( + identifier, + Schema.newBuilder() + .option(METASTORE_PARTITIONED_TABLE.key(), "true") + .column("col", DataTypes.INT()) + .column("dt", DataTypes.STRING()) + .partitionKeys("dt") + .build(), + false); + writeAppendPartition(identifier, "keep"); + writeAppendPartition(identifier, "20230101"); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + Identifier owner = identifier; + if (otherBranch) { + table.createBranch("other"); + owner = new Identifier(databaseName, tableName, "other"); + writeAppendPartition(owner, "keep"); + } + writeAppendPartition(owner, "20230101.done"); + setHistoricalReadSelector(owner, selector); + assertPhysicalDts(owner, "keep"); + long snapshotId = table.snapshotManager().latestSnapshotId(); + + catalog.dropPartitions(identifier, partitionSpecs("20230101")); + + // Ownership checks must neither follow nor remove the persisted query selector. + assertPhysicalDts(owner, "keep"); + catalog.alterTable( + owner, Collections.singletonList(SchemaChange.removeOption(selector)), false); + assertPhysicalDts(owner, "keep", "20230101.done"); + if (otherBranch) { + assertPhysicalDts(identifier, "keep"); + } + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId + 1); + assertHmsDts(databaseName, tableName, "keep", "20230101.done"); + } + + @ParameterizedTest + @ValueSource(strings = {"scan.snapshot-id", "scan.tag-name"}) + public void testDropPartitionsKeepsSharedPartitionWithHistoricalReads(String selector) + throws Exception { + String databaseName = "test_shared_partition_historical_reads"; + String tableName = "t"; + catalog.createDatabase(databaseName, false); + Identifier identifier = Identifier.create(databaseName, tableName); + catalog.createTable( + identifier, + Schema.newBuilder() + .option(METASTORE_PARTITIONED_TABLE.key(), "true") + .column("col", DataTypes.INT()) + .column("dt", DataTypes.STRING()) + .partitionKeys("dt") + .build(), + false); + writeAppendPartition(identifier, "keep"); + writeAppendPartition(identifier, "20230101"); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + table.createBranch("other"); + Identifier other = new Identifier(databaseName, tableName, "other"); + writeAppendPartition(other, "keep"); + writeAppendPartition(other, "20230101"); + AddDonePartitionAction action = + new AddDonePartitionAction(table.catalogEnvironment().partitionModification()); + action.markDone("dt=20230101"); + setHistoricalReadSelector(other, selector); + assertPhysicalDts(other, "keep"); + + catalog.dropPartitions(identifier, partitionSpecs("20230101")); + + assertPhysicalDts(other, "keep"); + catalog.alterTable( + other, Collections.singletonList(SchemaChange.removeOption(selector)), false); + assertPhysicalDts(identifier, "keep"); + assertPhysicalDts(other, "keep", "20230101"); + assertHmsDts(databaseName, tableName, "keep", "20230101", "20230101.done"); + } + + private void setHistoricalReadSelector(Identifier identifier, String selector) + throws Exception { + String value = "1"; + if (selector.equals("scan.tag-name")) { + catalog.getTable(identifier).createTag("before", 1L); + value = "before"; + } + catalog.alterTable( + identifier, + Collections.singletonList(SchemaChange.setOption(selector, value)), + false); + } + + private void writeIntegerChainPartitions(Identifier identifier, int... partitions) + throws Exception { + BatchWriteBuilder writeBuilder = catalog.getTable(identifier).newBatchWriteBuilder(); + try (BatchTableWrite write = writeBuilder.newWrite(); + BatchTableCommit commit = writeBuilder.newCommit()) { + for (int dt : partitions) { + write.write( + GenericRow.of( + dt, BinaryString.fromString("1"), BinaryString.fromString("1"))); + } + commit.commit(write.prepareCommit()); + } + } + private static Map orderedSpec( String key1, String value1, String key2, String value2) { Map spec = new LinkedHashMap<>();