From ee50ad66db0727d590282a34217591016673528b Mon Sep 17 00:00:00 2001 From: Kerwin Zhang Date: Tue, 3 Nov 2026 11:43:53 +0800 Subject: [PATCH 1/5] [core] Drop expired done partitions only from the metastore --- .../apache/paimon/catalog/CachingCatalog.java | 10 +++ .../org/apache/paimon/catalog/Catalog.java | 10 +++ .../paimon/catalog/DelegateCatalog.java | 7 ++ .../operation/NormalPartitionExpire.java | 3 +- .../paimon/table/PartitionModification.java | 10 +++ .../paimon/catalog/CachingCatalogTest.java | 18 +++++ .../paimon/catalog/DelegateCatalogTest.java | 13 +++ .../paimon/operation/PartitionExpireTest.java | 17 ++-- .../org/apache/paimon/hive/HiveCatalog.java | 22 +++-- .../apache/paimon/hive/HiveCatalogTest.java | 81 +++++++++++++++++++ 10 files changed, 179 insertions(+), 12 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java index 037f7ea2bfba..3fa572b92d36 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java @@ -390,6 +390,16 @@ public void dropPartitions(Identifier identifier, List> part } } + @Override + public void dropPartitionsFromMetastore( + Identifier identifier, List> partitions) + throws TableNotExistException { + wrapped.dropPartitionsFromMetastore(identifier, partitions); + if (partitionCache != null) { + partitionCache.invalidate(identifier); + } + } + @Override public void alterPartitions(Identifier identifier, List partitions) throws TableNotExistException { diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java index f3882b5b66bb..6e9679063dc3 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java @@ -1151,6 +1151,16 @@ default void dropPartitions(Identifier identifier, List> par } } + /** + * Remove partition metadata from the metastore without modifying table data or snapshots. + * + *

This is also used for synthetic partitions such as done markers, whose values need not + * conform to the table's partition types. Catalogs without partition metadata do nothing. + */ + default void dropPartitionsFromMetastore( + Identifier identifier, List> partitions) + throws TableNotExistException {} + /** * Alter partitions of the specify table. For non-existent partitions, partitions will be * created directly. diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java index 8b0aa817d430..1ebc10ced816 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java @@ -378,6 +378,13 @@ public void dropPartitions(Identifier identifier, List> part wrapped.dropPartitions(identifier, partitions); } + @Override + public void dropPartitionsFromMetastore( + Identifier identifier, List> partitions) + throws TableNotExistException { + wrapped.dropPartitionsFromMetastore(identifier, partitions); + } + @Override public void alterPartitions(Identifier identifier, List partitions) throws TableNotExistException { 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..c8dcedacfff8 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 @@ -199,7 +199,8 @@ private void doBatchExpire( try { partitionModification.dropPartitions(expiredBatchPartitions); // also drop corresponding .done partitions - partitionModification.dropPartitions(toDonePartitions(expiredBatchPartitions)); + partitionModification.dropPartitionsFromMetastore( + toDonePartitions(expiredBatchPartitions)); } catch (Catalog.TableNotExistException e) { throw new RuntimeException(e); } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/PartitionModification.java b/paimon-core/src/main/java/org/apache/paimon/table/PartitionModification.java index b3850bf9bef7..7430d5fe19ce 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/PartitionModification.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/PartitionModification.java @@ -33,6 +33,10 @@ void createPartitions(List> partitions) void dropPartitions(List> partitions) throws Catalog.TableNotExistException; + /** Remove only metastore metadata, including synthetic done partitions. */ + default void dropPartitionsFromMetastore(List> partitions) + throws Catalog.TableNotExistException {} + void alterPartitions(List partitions) throws Catalog.TableNotExistException; @@ -51,6 +55,12 @@ public void dropPartitions(List> partitions) catalog.dropPartitions(identifier, partitions); } + @Override + public void dropPartitionsFromMetastore(List> partitions) + throws Catalog.TableNotExistException { + catalog.dropPartitionsFromMetastore(identifier, partitions); + } + @Override public void alterPartitions(List partitions) throws Catalog.TableNotExistException { diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java index 71427641b79f..ca546c265480 100644 --- a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java @@ -345,6 +345,24 @@ public void testPartitionCache() throws Exception { assertThat(partitionEntryListFromCache).containsAll(partitionEntryList); } + @Test + public void testDropPartitionsFromMetastoreInvalidatesPartitionCache() throws Exception { + Catalog wrapped = Mockito.mock(Catalog.class); + TestableCachingCatalog catalog = + new TestableCachingCatalog(wrapped, EXPIRATION_TTL, ticker); + Identifier identifier = new Identifier("db", "tbl"); + Map spec = singletonMap("dt", "20230101.done"); + Partition partition = new Partition(spec, 0, 0, 0, 0, -1, true); + when(wrapped.listPartitions(identifier)).thenReturn(singletonList(partition), emptyList()); + + assertThat(catalog.listPartitions(identifier)).containsExactly(partition); + catalog.dropPartitionsFromMetastore(identifier, singletonList(spec)); + + Mockito.verify(wrapped).dropPartitionsFromMetastore(identifier, singletonList(spec)); + Mockito.verify(wrapped, Mockito.never()).dropPartitions(identifier, singletonList(spec)); + assertThat(catalog.listPartitions(identifier)).isEmpty(); + } + @Test public void testCreatePartitionsInvalidatesPartitionCache() throws Exception { Catalog wrapped = Mockito.mock(Catalog.class); diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/DelegateCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/DelegateCatalogTest.java index 63fa0b478181..0d80dd8c2ff1 100644 --- a/paimon-core/src/test/java/org/apache/paimon/catalog/DelegateCatalogTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/catalog/DelegateCatalogTest.java @@ -91,6 +91,19 @@ void testCreatePartitionsCarriesOptionsToTheWrappedCatalog() throws Exception { verify(wrapped, never()).createPartitions(IDENTIFIER, specs, true, null, false, null); } + @Test + void testDropPartitionsFromMetastoreDoesNotDropData() throws Exception { + Catalog wrapped = mock(Catalog.class); + Catalog delegating = new TestDelegateCatalog(wrapped); + List> specs = + Collections.singletonList(Collections.singletonMap("dt", "20230101.done")); + + delegating.dropPartitionsFromMetastore(IDENTIFIER, specs); + + verify(wrapped).dropPartitionsFromMetastore(IDENTIFIER, specs); + verify(wrapped, never()).dropPartitions(any(), anyList()); + } + /** {@link DelegateCatalog} forwards every operation; these tests never rebuild one. */ private static class TestDelegateCatalog extends DelegateCatalog { 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..4ca0d904938d 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); - } - } + dropPartitionsFromMetastore(partitions); try (FileStoreCommit commit = table.store() .newCommit( @@ -155,6 +150,16 @@ public void dropPartitions(List> partitions) } } + @Override + public void dropPartitionsFromMetastore(List> partitions) { + for (Map partition : partitions) { + // only record partitions that were created + if (createdPartitions.remove(partition)) { + deletedPartitions.add(partition); + } + } + } + @Override public void alterPartitions(List partitions) throws Catalog.TableNotExistException {} 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..3552bb29172c 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 @@ -488,8 +488,23 @@ 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; + dropPartitionsFromMetastore(identifier, partitions, schema); + if (CoreOptions.fromMap(schema.options()).tagToPartitionField() == null) { + super.dropPartitions(identifier, partitions); + } + } + + @Override + public void dropPartitionsFromMetastore( + Identifier identifier, List> partitions) + throws TableNotExistException { + dropPartitionsFromMetastore(identifier, partitions, loadTableSchema(identifier)); + } + + private void dropPartitionsFromMetastore( + Identifier identifier, List> partitions, TableSchema schema) + throws TableNotExistException { + boolean tagToPart = CoreOptions.fromMap(schema.options()).tagToPartitionField() != null; if (metastorePartitioned(schema)) { List> metaPartitions = tagToPart @@ -515,9 +530,6 @@ public void dropPartitions(Identifier identifier, List> part } } } - if (!tagToPart) { - super.dropPartitions(identifier, partitions); - } } private String getDataFilePath(Identifier tableIdentifier, Table hmsTable) { 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..5adbe20ba1fd 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; @@ -833,6 +834,86 @@ 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 testDropPartitionsFromMetastorePreservesData() 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"); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + long snapshotId = table.snapshotManager().latestSnapshotId(); + + catalog.dropPartitionsFromMetastore(identifier, partitionSpecs("20230101.done")); + + assertHmsDts(databaseName, tableName); + assertPhysicalDts(identifier, "20230101.done"); + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId); + + // A real STRING partition ending in .done must still support ordinary data deletion. + catalog.dropPartitions(identifier, partitionSpecs("20230101.done")); + assertPhysicalDts(identifier); + } + private static Map orderedSpec( String key1, String value1, String key2, String value2) { Map spec = new LinkedHashMap<>(); From 08b5cfb575e8f2ec2661ebc23c6e09300bc7d518 Mon Sep 17 00:00:00 2001 From: Kerwin Zhang Date: Tue, 3 Nov 2026 14:20:56 +0800 Subject: [PATCH 2/5] [hive] Check done partition branch ownership using ordinary values --- .../apache/paimon/catalog/CachingCatalog.java | 12 ++ .../org/apache/paimon/catalog/Catalog.java | 15 ++ .../paimon/catalog/DelegateCatalog.java | 9 + .../operation/NormalPartitionExpire.java | 2 +- .../paimon/table/PartitionModification.java | 14 ++ .../paimon/catalog/CachingCatalogTest.java | 23 +++ .../paimon/catalog/DelegateCatalogTest.java | 16 ++ .../org/apache/paimon/hive/HiveCatalog.java | 39 +++- .../apache/paimon/hive/HiveCatalogTest.java | 170 ++++++++++++++++++ 9 files changed, 292 insertions(+), 8 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java index 3fa572b92d36..f37ebd52fae3 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java @@ -400,6 +400,18 @@ public void dropPartitionsFromMetastore( } } + @Override + public void dropPartitionsFromMetastore( + Identifier identifier, + List> partitions, + List> dataPartitions) + throws TableNotExistException { + wrapped.dropPartitionsFromMetastore(identifier, partitions, dataPartitions); + if (partitionCache != null) { + partitionCache.invalidate(identifier); + } + } + @Override public void alterPartitions(Identifier identifier, List partitions) throws TableNotExistException { diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java index 6e9679063dc3..3120704eea3a 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java @@ -1161,6 +1161,21 @@ default void dropPartitionsFromMetastore( Identifier identifier, List> partitions) throws TableNotExistException {} + /** + * Remove partition metadata using the corresponding data partitions to check branch ownership. + * + * @param partitions metastore partition specs to remove, including synthetic markers + * @param dataPartitions underlying data partition specs, in the same order as {@code + * partitions} + */ + default void dropPartitionsFromMetastore( + Identifier identifier, + List> partitions, + List> dataPartitions) + throws TableNotExistException { + dropPartitionsFromMetastore(identifier, partitions); + } + /** * Alter partitions of the specify table. For non-existent partitions, partitions will be * created directly. diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java index 1ebc10ced816..6231ce211b1f 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java @@ -385,6 +385,15 @@ public void dropPartitionsFromMetastore( wrapped.dropPartitionsFromMetastore(identifier, partitions); } + @Override + public void dropPartitionsFromMetastore( + Identifier identifier, + List> partitions, + List> dataPartitions) + throws TableNotExistException { + wrapped.dropPartitionsFromMetastore(identifier, partitions, dataPartitions); + } + @Override public void alterPartitions(Identifier identifier, List partitions) throws TableNotExistException { 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 c8dcedacfff8..75aef5001563 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 @@ -200,7 +200,7 @@ private void doBatchExpire( partitionModification.dropPartitions(expiredBatchPartitions); // also drop corresponding .done partitions partitionModification.dropPartitionsFromMetastore( - toDonePartitions(expiredBatchPartitions)); + toDonePartitions(expiredBatchPartitions), expiredBatchPartitions); } catch (Catalog.TableNotExistException e) { throw new RuntimeException(e); } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/PartitionModification.java b/paimon-core/src/main/java/org/apache/paimon/table/PartitionModification.java index 7430d5fe19ce..62825bb1393e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/PartitionModification.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/PartitionModification.java @@ -37,6 +37,13 @@ void createPartitions(List> partitions) default void dropPartitionsFromMetastore(List> partitions) throws Catalog.TableNotExistException {} + /** Remove metadata while checking ownership using the corresponding data partition specs. */ + default void dropPartitionsFromMetastore( + List> partitions, List> dataPartitions) + throws Catalog.TableNotExistException { + dropPartitionsFromMetastore(partitions); + } + void alterPartitions(List partitions) throws Catalog.TableNotExistException; @@ -61,6 +68,13 @@ public void dropPartitionsFromMetastore(List> partitions) catalog.dropPartitionsFromMetastore(identifier, partitions); } + @Override + public void dropPartitionsFromMetastore( + List> partitions, List> dataPartitions) + throws Catalog.TableNotExistException { + catalog.dropPartitionsFromMetastore(identifier, partitions, dataPartitions); + } + @Override public void alterPartitions(List partitions) throws Catalog.TableNotExistException { diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java index ca546c265480..1c0ed34487be 100644 --- a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java @@ -363,6 +363,29 @@ public void testDropPartitionsFromMetastoreInvalidatesPartitionCache() throws Ex assertThat(catalog.listPartitions(identifier)).isEmpty(); } + @Test + public void testDropMarkerMetadataForwardsDataPartitionsAndInvalidatesCache() throws Exception { + Catalog wrapped = Mockito.mock(Catalog.class); + TestableCachingCatalog catalog = + new TestableCachingCatalog(wrapped, EXPIRATION_TTL, ticker); + Identifier identifier = new Identifier("db", "tbl"); + Map marker = singletonMap("dt", "20230101.done"); + Map dataPartition = singletonMap("dt", "20230101"); + Partition partition = new Partition(marker, 0, 0, 0, 0, -1, true); + when(wrapped.listPartitions(identifier)).thenReturn(singletonList(partition), emptyList()); + + assertThat(catalog.listPartitions(identifier)).containsExactly(partition); + catalog.dropPartitionsFromMetastore( + identifier, singletonList(marker), singletonList(dataPartition)); + + Mockito.verify(wrapped) + .dropPartitionsFromMetastore( + identifier, singletonList(marker), singletonList(dataPartition)); + Mockito.verify(wrapped, Mockito.never()) + .dropPartitionsFromMetastore(identifier, singletonList(marker)); + assertThat(catalog.listPartitions(identifier)).isEmpty(); + } + @Test public void testCreatePartitionsInvalidatesPartitionCache() throws Exception { Catalog wrapped = Mockito.mock(Catalog.class); diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/DelegateCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/DelegateCatalogTest.java index 0d80dd8c2ff1..e08cd7359d9c 100644 --- a/paimon-core/src/test/java/org/apache/paimon/catalog/DelegateCatalogTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/catalog/DelegateCatalogTest.java @@ -104,6 +104,22 @@ void testDropPartitionsFromMetastoreDoesNotDropData() throws Exception { verify(wrapped, never()).dropPartitions(any(), anyList()); } + @Test + void testDropMarkerMetadataForwardsDataPartitionSpecs() throws Exception { + Catalog wrapped = mock(Catalog.class); + Catalog delegating = new TestDelegateCatalog(wrapped); + List> markers = + Collections.singletonList(Collections.singletonMap("dt", "20230101.done")); + List> dataPartitions = + Collections.singletonList(Collections.singletonMap("dt", "20230101")); + + delegating.dropPartitionsFromMetastore(IDENTIFIER, markers, dataPartitions); + + verify(wrapped).dropPartitionsFromMetastore(IDENTIFIER, markers, dataPartitions); + verify(wrapped, never()).dropPartitionsFromMetastore(IDENTIFIER, markers); + verify(wrapped, never()).dropPartitions(any(), anyList()); + } + /** {@link DelegateCatalog} forwards every operation; these tests never rebuild one. */ private static class TestDelegateCatalog extends DelegateCatalog { 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 3552bb29172c..aee9d387fb61 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 @@ -488,7 +488,7 @@ private void createPartitionsUtil( public void dropPartitions(Identifier identifier, List> partitions) throws TableNotExistException { TableSchema schema = this.loadTableSchema(identifier); - dropPartitionsFromMetastore(identifier, partitions, schema); + dropPartitionsFromMetastore(identifier, partitions, partitions, schema); if (CoreOptions.fromMap(schema.options()).tagToPartitionField() == null) { super.dropPartitions(identifier, partitions); } @@ -498,18 +498,43 @@ public void dropPartitions(Identifier identifier, List> part public void dropPartitionsFromMetastore( Identifier identifier, List> partitions) throws TableNotExistException { - dropPartitionsFromMetastore(identifier, partitions, loadTableSchema(identifier)); + dropPartitionsFromMetastore( + identifier, partitions, partitions, loadTableSchema(identifier)); + } + + @Override + public void dropPartitionsFromMetastore( + Identifier identifier, + List> partitions, + List> dataPartitions) + throws TableNotExistException { + dropPartitionsFromMetastore( + identifier, partitions, dataPartitions, loadTableSchema(identifier)); } private void dropPartitionsFromMetastore( - Identifier identifier, List> partitions, TableSchema schema) + Identifier identifier, + List> partitions, + List> dataPartitions, + TableSchema schema) throws TableNotExistException { + checkArgument( + partitions.size() == dataPartitions.size(), + "Metastore and data partition specs must have the same size."); boolean tagToPart = CoreOptions.fromMap(schema.options()).tagToPartitionField() != null; if (metastorePartitioned(schema)) { - List> metaPartitions = - tagToPart - ? partitions - : removePartitionsExistsInOtherBranches(identifier, partitions); + List> metaPartitions = partitions; + if (!tagToPart) { + Set> removableDataPartitions = + new HashSet<>( + removePartitionsExistsInOtherBranches(identifier, dataPartitions)); + metaPartitions = new ArrayList<>(); + for (int i = 0; i < partitions.size(); i++) { + if (removableDataPartitions.contains(dataPartitions.get(i))) { + metaPartitions.add(partitions.get(i)); + } + } + } for (Map part : metaPartitions) { List partitionValues = new ArrayList<>( 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 5adbe20ba1fd..f615decd100b 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 @@ -884,6 +884,139 @@ public void testExpireIntegerPartitionsWithDoneMarkers(boolean markDone) throws } } + @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 testDropPartitionsFromMetastorePreservesData() throws Exception { String databaseName = "test_drop_partition_metadata"; @@ -914,6 +1047,43 @@ public void testDropPartitionsFromMetastorePreservesData() throws Exception { assertPhysicalDts(identifier); } + @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"); + long snapshotId = table.snapshotManager().latestSnapshotId(); + + catalog.dropPartitionsFromMetastore(identifier, partitionSpecs("20230101.done")); + assertHmsDts(databaseName, tableName, "20230101.done"); + assertPhysicalDts(identifier, "20230101.done"); + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId); + + catalog.dropPartitions(identifier, partitionSpecs("20230101.done")); + assertPhysicalDts(identifier); + assertPhysicalDts(otherId, "20230101.done"); + assertHmsDts(databaseName, tableName, "20230101.done"); + + catalog.dropPartitions(otherId, partitionSpecs("20230101.done")); + assertPhysicalDts(otherId); + assertHmsDts(databaseName, tableName); + } + private static Map orderedSpec( String key1, String value1, String key2, String value2) { Map spec = new LinkedHashMap<>(); From bea7c77251d090e947fd8c2aec441e735dd69099 Mon Sep 17 00:00:00 2001 From: Kerwin Zhang Date: Tue, 3 Nov 2026 14:48:52 +0800 Subject: [PATCH 3/5] [core][hive] Preserve marker name collisions and fix chain cleanup --- .../org/apache/paimon/catalog/Catalog.java | 3 + .../operation/ChainTablePartitionExpire.java | 6 +- .../ChainTablePartitionExpireTest.java | 6 + .../org/apache/paimon/hive/HiveCatalog.java | 45 +++-- .../apache/paimon/hive/HiveCatalogTest.java | 174 ++++++++++++++++++ 5 files changed, 218 insertions(+), 16 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java index 3120704eea3a..f0353fa61396 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java @@ -1164,6 +1164,9 @@ default void dropPartitionsFromMetastore( /** * Remove partition metadata using the corresponding data partitions to check branch ownership. * + *

A synthetic spec must also be retained when it names a real data partition still in use. + * Synthetic specs need not conform to the data partition types. + * * @param partitions metastore partition specs to remove, including synthetic markers * @param dataPartitions underlying data partition specs, in the same order as {@code * partitions} 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..e422ea922815 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,13 @@ private void batchDropPartitions( for (List> batch : Lists.partition(partitionSpecs, expireBatchSize)) { partitionModification.dropPartitions(batch); - partitionModification.dropPartitions(toDonePartitions(batch)); + partitionModification.dropPartitionsFromMetastore( + toDonePartitions(batch), batch); } } else { partitionModification.dropPartitions(partitionSpecs); - partitionModification.dropPartitions(toDonePartitions(partitionSpecs)); + partitionModification.dropPartitionsFromMetastore( + toDonePartitions(partitionSpecs), partitionSpecs); } } catch (Catalog.TableNotExistException e) { throw new RuntimeException(e); 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..532f39f74f87 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 @@ -892,6 +892,12 @@ public void dropPartitions(List> partitions) } } + @Override + public void dropPartitionsFromMetastore(List> partitions) + throws Catalog.TableNotExistException { + dropPartitions(partitions); + } + @Override public void alterPartitions(List partitions) throws Catalog.TableNotExistException {} 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 aee9d387fb61..399395bdea45 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 @@ -49,6 +49,7 @@ import org.apache.paimon.table.FallbackReadFileStoreTable; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.FormatTable; +import org.apache.paimon.table.source.DataTableScan; import org.apache.paimon.types.DataField; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowType; @@ -525,12 +526,19 @@ private void dropPartitionsFromMetastore( if (metastorePartitioned(schema)) { List> metaPartitions = partitions; if (!tagToPart) { - Set> removableDataPartitions = + boolean syntheticPartitions = !partitions.equals(dataPartitions); + Set> specsToCheck = new HashSet<>(dataPartitions); + specsToCheck.addAll(partitions); + Set> removablePartitions = new HashSet<>( - removePartitionsExistsInOtherBranches(identifier, dataPartitions)); + removePartitionsStillInUse( + identifier, + new ArrayList<>(specsToCheck), + syntheticPartitions)); metaPartitions = new ArrayList<>(); for (int i = 0; i < partitions.size(); i++) { - if (removableDataPartitions.contains(dataPartitions.get(i))) { + if (removablePartitions.contains(dataPartitions.get(i)) + && removablePartitions.contains(partitions.get(i))) { metaPartitions.add(partitions.get(i)); } } @@ -790,26 +798,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; } @@ -819,8 +824,20 @@ 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()); + DataTableScan scan = branchTable.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 f615decd100b..ff8a6f8bbf3b 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 @@ -60,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; @@ -1084,6 +1085,179 @@ public void testRealDoneSuffixPartitionKeepsBranchProtection() throws Exception 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"); + catalog.dropPartitions(identifier, partitionSpecs("20230101")); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + long snapshotId = table.snapshotManager().latestSnapshotId(); + + catalog.dropPartitionsFromMetastore( + identifier, partitionSpecs("20230101.done"), partitionSpecs("20230101")); + + assertHmsDts(databaseName, tableName, "20230101.done"); + assertPhysicalDts(identifier, "20230101.done"); + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId); + } + + 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<>(); From ec75005b046c9ec536427542e382ba2b31c9e9d4 Mon Sep 17 00:00:00 2001 From: Kerwin Zhang Date: Tue, 3 Nov 2026 19:26:25 +0800 Subject: [PATCH 4/5] [hive] Keep done marker cleanup inside partition deletion --- .../apache/paimon/catalog/CachingCatalog.java | 22 ---- .../org/apache/paimon/catalog/Catalog.java | 28 ---- .../paimon/catalog/DelegateCatalog.java | 16 --- .../operation/ChainTablePartitionExpire.java | 21 --- .../operation/NormalPartitionExpire.java | 24 ---- .../paimon/table/PartitionModification.java | 24 ---- .../paimon/catalog/CachingCatalogTest.java | 41 ------ .../paimon/catalog/DelegateCatalogTest.java | 29 ----- .../ChainTablePartitionExpireTest.java | 16 ++- .../paimon/operation/PartitionExpireTest.java | 17 ++- .../org/apache/paimon/hive/HiveCatalog.java | 121 +++++++++--------- .../apache/paimon/hive/HiveCatalogTest.java | 79 +++++++++--- 12 files changed, 142 insertions(+), 296 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java index f37ebd52fae3..037f7ea2bfba 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java @@ -390,28 +390,6 @@ public void dropPartitions(Identifier identifier, List> part } } - @Override - public void dropPartitionsFromMetastore( - Identifier identifier, List> partitions) - throws TableNotExistException { - wrapped.dropPartitionsFromMetastore(identifier, partitions); - if (partitionCache != null) { - partitionCache.invalidate(identifier); - } - } - - @Override - public void dropPartitionsFromMetastore( - Identifier identifier, - List> partitions, - List> dataPartitions) - throws TableNotExistException { - wrapped.dropPartitionsFromMetastore(identifier, partitions, dataPartitions); - if (partitionCache != null) { - partitionCache.invalidate(identifier); - } - } - @Override public void alterPartitions(Identifier identifier, List partitions) throws TableNotExistException { diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java index f0353fa61396..f3882b5b66bb 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java @@ -1151,34 +1151,6 @@ default void dropPartitions(Identifier identifier, List> par } } - /** - * Remove partition metadata from the metastore without modifying table data or snapshots. - * - *

This is also used for synthetic partitions such as done markers, whose values need not - * conform to the table's partition types. Catalogs without partition metadata do nothing. - */ - default void dropPartitionsFromMetastore( - Identifier identifier, List> partitions) - throws TableNotExistException {} - - /** - * Remove partition metadata using the corresponding data partitions to check branch ownership. - * - *

A synthetic spec must also be retained when it names a real data partition still in use. - * Synthetic specs need not conform to the data partition types. - * - * @param partitions metastore partition specs to remove, including synthetic markers - * @param dataPartitions underlying data partition specs, in the same order as {@code - * partitions} - */ - default void dropPartitionsFromMetastore( - Identifier identifier, - List> partitions, - List> dataPartitions) - throws TableNotExistException { - dropPartitionsFromMetastore(identifier, partitions); - } - /** * Alter partitions of the specify table. For non-existent partitions, partitions will be * created directly. diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java index 6231ce211b1f..8b0aa817d430 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java @@ -378,22 +378,6 @@ public void dropPartitions(Identifier identifier, List> part wrapped.dropPartitions(identifier, partitions); } - @Override - public void dropPartitionsFromMetastore( - Identifier identifier, List> partitions) - throws TableNotExistException { - wrapped.dropPartitionsFromMetastore(identifier, partitions); - } - - @Override - public void dropPartitionsFromMetastore( - Identifier identifier, - List> partitions, - List> dataPartitions) - throws TableNotExistException { - wrapped.dropPartitionsFromMetastore(identifier, partitions, dataPartitions); - } - @Override public void alterPartitions(Identifier identifier, List partitions) throws TableNotExistException { 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 e422ea922815..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,13 +352,9 @@ private void batchDropPartitions( for (List> batch : Lists.partition(partitionSpecs, expireBatchSize)) { partitionModification.dropPartitions(batch); - partitionModification.dropPartitionsFromMetastore( - toDonePartitions(batch), batch); } } else { partitionModification.dropPartitions(partitionSpecs); - partitionModification.dropPartitionsFromMetastore( - toDonePartitions(partitionSpecs), partitionSpecs); } } catch (Catalog.TableNotExistException e) { throw new RuntimeException(e); @@ -375,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 75aef5001563..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,37 +197,14 @@ private void doBatchExpire( if (partitionModification != null) { try { partitionModification.dropPartitions(expiredBatchPartitions); - // also drop corresponding .done partitions - partitionModification.dropPartitionsFromMetastore( - toDonePartitions(expiredBatchPartitions), 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/main/java/org/apache/paimon/table/PartitionModification.java b/paimon-core/src/main/java/org/apache/paimon/table/PartitionModification.java index 62825bb1393e..b3850bf9bef7 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/PartitionModification.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/PartitionModification.java @@ -33,17 +33,6 @@ void createPartitions(List> partitions) void dropPartitions(List> partitions) throws Catalog.TableNotExistException; - /** Remove only metastore metadata, including synthetic done partitions. */ - default void dropPartitionsFromMetastore(List> partitions) - throws Catalog.TableNotExistException {} - - /** Remove metadata while checking ownership using the corresponding data partition specs. */ - default void dropPartitionsFromMetastore( - List> partitions, List> dataPartitions) - throws Catalog.TableNotExistException { - dropPartitionsFromMetastore(partitions); - } - void alterPartitions(List partitions) throws Catalog.TableNotExistException; @@ -62,19 +51,6 @@ public void dropPartitions(List> partitions) catalog.dropPartitions(identifier, partitions); } - @Override - public void dropPartitionsFromMetastore(List> partitions) - throws Catalog.TableNotExistException { - catalog.dropPartitionsFromMetastore(identifier, partitions); - } - - @Override - public void dropPartitionsFromMetastore( - List> partitions, List> dataPartitions) - throws Catalog.TableNotExistException { - catalog.dropPartitionsFromMetastore(identifier, partitions, dataPartitions); - } - @Override public void alterPartitions(List partitions) throws Catalog.TableNotExistException { diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java index 1c0ed34487be..71427641b79f 100644 --- a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java @@ -345,47 +345,6 @@ public void testPartitionCache() throws Exception { assertThat(partitionEntryListFromCache).containsAll(partitionEntryList); } - @Test - public void testDropPartitionsFromMetastoreInvalidatesPartitionCache() throws Exception { - Catalog wrapped = Mockito.mock(Catalog.class); - TestableCachingCatalog catalog = - new TestableCachingCatalog(wrapped, EXPIRATION_TTL, ticker); - Identifier identifier = new Identifier("db", "tbl"); - Map spec = singletonMap("dt", "20230101.done"); - Partition partition = new Partition(spec, 0, 0, 0, 0, -1, true); - when(wrapped.listPartitions(identifier)).thenReturn(singletonList(partition), emptyList()); - - assertThat(catalog.listPartitions(identifier)).containsExactly(partition); - catalog.dropPartitionsFromMetastore(identifier, singletonList(spec)); - - Mockito.verify(wrapped).dropPartitionsFromMetastore(identifier, singletonList(spec)); - Mockito.verify(wrapped, Mockito.never()).dropPartitions(identifier, singletonList(spec)); - assertThat(catalog.listPartitions(identifier)).isEmpty(); - } - - @Test - public void testDropMarkerMetadataForwardsDataPartitionsAndInvalidatesCache() throws Exception { - Catalog wrapped = Mockito.mock(Catalog.class); - TestableCachingCatalog catalog = - new TestableCachingCatalog(wrapped, EXPIRATION_TTL, ticker); - Identifier identifier = new Identifier("db", "tbl"); - Map marker = singletonMap("dt", "20230101.done"); - Map dataPartition = singletonMap("dt", "20230101"); - Partition partition = new Partition(marker, 0, 0, 0, 0, -1, true); - when(wrapped.listPartitions(identifier)).thenReturn(singletonList(partition), emptyList()); - - assertThat(catalog.listPartitions(identifier)).containsExactly(partition); - catalog.dropPartitionsFromMetastore( - identifier, singletonList(marker), singletonList(dataPartition)); - - Mockito.verify(wrapped) - .dropPartitionsFromMetastore( - identifier, singletonList(marker), singletonList(dataPartition)); - Mockito.verify(wrapped, Mockito.never()) - .dropPartitionsFromMetastore(identifier, singletonList(marker)); - assertThat(catalog.listPartitions(identifier)).isEmpty(); - } - @Test public void testCreatePartitionsInvalidatesPartitionCache() throws Exception { Catalog wrapped = Mockito.mock(Catalog.class); diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/DelegateCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/DelegateCatalogTest.java index e08cd7359d9c..63fa0b478181 100644 --- a/paimon-core/src/test/java/org/apache/paimon/catalog/DelegateCatalogTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/catalog/DelegateCatalogTest.java @@ -91,35 +91,6 @@ void testCreatePartitionsCarriesOptionsToTheWrappedCatalog() throws Exception { verify(wrapped, never()).createPartitions(IDENTIFIER, specs, true, null, false, null); } - @Test - void testDropPartitionsFromMetastoreDoesNotDropData() throws Exception { - Catalog wrapped = mock(Catalog.class); - Catalog delegating = new TestDelegateCatalog(wrapped); - List> specs = - Collections.singletonList(Collections.singletonMap("dt", "20230101.done")); - - delegating.dropPartitionsFromMetastore(IDENTIFIER, specs); - - verify(wrapped).dropPartitionsFromMetastore(IDENTIFIER, specs); - verify(wrapped, never()).dropPartitions(any(), anyList()); - } - - @Test - void testDropMarkerMetadataForwardsDataPartitionSpecs() throws Exception { - Catalog wrapped = mock(Catalog.class); - Catalog delegating = new TestDelegateCatalog(wrapped); - List> markers = - Collections.singletonList(Collections.singletonMap("dt", "20230101.done")); - List> dataPartitions = - Collections.singletonList(Collections.singletonMap("dt", "20230101")); - - delegating.dropPartitionsFromMetastore(IDENTIFIER, markers, dataPartitions); - - verify(wrapped).dropPartitionsFromMetastore(IDENTIFIER, markers, dataPartitions); - verify(wrapped, never()).dropPartitionsFromMetastore(IDENTIFIER, markers); - verify(wrapped, never()).dropPartitions(any(), anyList()); - } - /** {@link DelegateCatalog} forwards every operation; these tests never rebuild one. */ private static class TestDelegateCatalog extends DelegateCatalog { 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 532f39f74f87..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,15 +889,19 @@ 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); + } } } - @Override - public void dropPartitionsFromMetastore(List> partitions) - throws Catalog.TableNotExistException { - dropPartitions(partitions); - } - @Override public void alterPartitions(List partitions) throws Catalog.TableNotExistException {} 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 4ca0d904938d..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,7 +139,7 @@ public void createPartitions(List> partitions) @Override public void dropPartitions(List> partitions) throws Catalog.TableNotExistException { - dropPartitionsFromMetastore(partitions); + dropPartitionMetadata(partitions); try (FileStoreCommit commit = table.store() .newCommit( @@ -148,10 +148,21 @@ 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); } - @Override - public void dropPartitionsFromMetastore(List> partitions) { + private void dropPartitionMetadata(List> partitions) { for (Map partition : partitions) { // only record partitions that were created if (createdPartitions.remove(partition)) { 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 399395bdea45..b4221c97ed22 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 @@ -489,78 +489,71 @@ private void createPartitionsUtil( public void dropPartitions(Identifier identifier, List> partitions) throws TableNotExistException { TableSchema schema = this.loadTableSchema(identifier); - dropPartitionsFromMetastore(identifier, partitions, partitions, schema); - if (CoreOptions.fromMap(schema.options()).tagToPartitionField() == null) { + boolean tagToPart = CoreOptions.fromMap(schema.options()).tagToPartitionField() != null; + boolean metastorePartitioned = metastorePartitioned(schema); + if (metastorePartitioned) { + dropPartitionsFromMetastore( + identifier, + tagToPart + ? partitions + : removePartitionsStillInUse(identifier, partitions, false), + schema); + } + if (!tagToPart) { super.dropPartitions(identifier, partitions); + if (metastorePartitioned) { + dropDonePartitions(identifier, partitions, schema); + } } } - @Override - public void dropPartitionsFromMetastore( - Identifier identifier, List> partitions) - throws TableNotExistException { - dropPartitionsFromMetastore( - identifier, partitions, partitions, loadTableSchema(identifier)); - } - - @Override - public void dropPartitionsFromMetastore( - Identifier identifier, - List> partitions, - List> dataPartitions) + private void dropDonePartitions( + Identifier identifier, List> partitions, TableSchema schema) throws TableNotExistException { - dropPartitionsFromMetastore( - identifier, partitions, dataPartitions, loadTableSchema(identifier)); + 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, - List> dataPartitions, - TableSchema schema) - throws TableNotExistException { - checkArgument( - partitions.size() == dataPartitions.size(), - "Metastore and data partition specs must have the same size."); - boolean tagToPart = CoreOptions.fromMap(schema.options()).tagToPartitionField() != null; - if (metastorePartitioned(schema)) { - List> metaPartitions = partitions; - if (!tagToPart) { - boolean syntheticPartitions = !partitions.equals(dataPartitions); - Set> specsToCheck = new HashSet<>(dataPartitions); - specsToCheck.addAll(partitions); - Set> removablePartitions = - new HashSet<>( - removePartitionsStillInUse( - identifier, - new ArrayList<>(specsToCheck), - syntheticPartitions)); - metaPartitions = new ArrayList<>(); - for (int i = 0; i < partitions.size(); i++) { - if (removablePartitions.contains(dataPartitions.get(i)) - && removablePartitions.contains(partitions.get(i))) { - metaPartitions.add(partitions.get(i)); - } - } - } - 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); - } + 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); } } } 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 ff8a6f8bbf3b..948a87af3c84 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 @@ -885,6 +885,51 @@ public void testExpireIntegerPartitionsWithDoneMarkers(boolean markDone) throws } } + @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 { @@ -1019,7 +1064,7 @@ public void testExpireIntegerPartitionsSharedWithOtherBranch(boolean markDone) } @Test - public void testDropPartitionsFromMetastorePreservesData() throws Exception { + public void testDropRealDoneSuffixPartitionAndItsMarker() throws Exception { String databaseName = "test_drop_partition_metadata"; String tableName = "t"; catalog.createDatabase(databaseName, false); @@ -1034,18 +1079,19 @@ public void testDropPartitionsFromMetastorePreservesData() throws Exception { .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(); - catalog.dropPartitionsFromMetastore(identifier, partitionSpecs("20230101.done")); - - assertHmsDts(databaseName, tableName); - assertPhysicalDts(identifier, "20230101.done"); - assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId); - // A real STRING partition ending in .done must still support ordinary data deletion. catalog.dropPartitions(identifier, partitionSpecs("20230101.done")); - assertPhysicalDts(identifier); + + assertHmsDts(databaseName, tableName, "20990101"); + assertPhysicalDts(identifier, "20990101"); + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId + 1); } @Test @@ -1068,17 +1114,16 @@ public void testRealDoneSuffixPartitionKeepsBranchProtection() throws Exception 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.dropPartitionsFromMetastore(identifier, partitionSpecs("20230101.done")); - assertHmsDts(databaseName, tableName, "20230101.done"); - assertPhysicalDts(identifier, "20230101.done"); - assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId); - catalog.dropPartitions(identifier, partitionSpecs("20230101.done")); assertPhysicalDts(identifier); assertPhysicalDts(otherId, "20230101.done"); - assertHmsDts(databaseName, tableName, "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); @@ -1232,16 +1277,14 @@ public void testMarkerCleanupKeepsCurrentBranchRealDoneSuffixPartition() throws false); writeAppendPartition(identifier, "20230101"); writeAppendPartition(identifier, "20230101.done"); - catalog.dropPartitions(identifier, partitionSpecs("20230101")); FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); long snapshotId = table.snapshotManager().latestSnapshotId(); - catalog.dropPartitionsFromMetastore( - identifier, partitionSpecs("20230101.done"), partitionSpecs("20230101")); + catalog.dropPartitions(identifier, partitionSpecs("20230101")); assertHmsDts(databaseName, tableName, "20230101.done"); assertPhysicalDts(identifier, "20230101.done"); - assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId); + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId + 1); } private void writeIntegerChainPartitions(Identifier identifier, int... partitions) From e68d1d51cf789dd37b26538ac1fac1d8623ffd5a Mon Sep 17 00:00:00 2001 From: Kerwin Zhang Date: Wed, 4 Nov 2026 00:28:00 +0800 Subject: [PATCH 5/5] [hive] Use latest physical metadata for partition ownership --- .../org/apache/paimon/hive/HiveCatalog.java | 6 +- .../apache/paimon/hive/HiveCatalogTest.java | 103 ++++++++++++++++++ 2 files changed, 107 insertions(+), 2 deletions(-) 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 b4221c97ed22..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; @@ -49,7 +50,6 @@ import org.apache.paimon.table.FallbackReadFileStoreTable; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.FormatTable; -import org.apache.paimon.table.source.DataTableScan; import org.apache.paimon.types.DataField; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowType; @@ -824,7 +824,9 @@ private List> removePartitionsStillInUse( branchTable.rowType().project(branchTable.partitionKeys()), branchTable.partitionKeys().toArray(new String[0]), branchTable.coreOptions().legacyPartitionName()); - DataTableScan scan = branchTable.newScan(); + // 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)); } 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 948a87af3c84..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 @@ -1287,6 +1287,109 @@ public void testMarkerCleanupKeepsCurrentBranchRealDoneSuffixPartition() throws 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();