This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 0e3f4461cc [hive] Populate partition done status from metastore events
(#8866)
0e3f4461cc is described below
commit 0e3f4461cc0de6401548fa47bd90fe367d37ecff
Author: Zouxxyy <[email protected]>
AuthorDate: Tue Jul 28 10:29:04 2026 +0800
[hive] Populate partition done status from metastore events (#8866)
---
.../java/org/apache/paimon/hive/HiveCatalog.java | 132 +++++++++++++++------
.../org/apache/paimon/hive/HiveCatalogTest.java | 47 ++++++++
2 files changed, 144 insertions(+), 35 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 926fc024f2..5811796e5a 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
@@ -595,47 +595,109 @@ public class HiveCatalog extends AbstractCatalog {
public List<org.apache.paimon.partition.Partition>
listPartitions(Identifier identifier)
throws TableNotExistException {
FileStoreTable table = (FileStoreTable) getTable(identifier);
- String tagToPartitionField = table.coreOptions().tagToPartitionField();
+ CoreOptions coreOptions = table.coreOptions();
+ String tagToPartitionField = coreOptions.tagToPartitionField();
+ List<org.apache.paimon.partition.Partition> partitions;
if (tagToPartitionField != null) {
try {
- List<Partition> partitions = listPartitionsFromHms(identifier);
- return partitions.stream()
- .map(
- part -> {
- Map<String, String> parameters =
part.getParameters();
- long recordCount =
- Long.parseLong(
-
parameters.getOrDefault(NUM_ROWS_PROP, "1"));
- long fileSizeInBytes =
- Long.parseLong(
-
parameters.getOrDefault(TOTAL_SIZE_PROP, "1"));
- long fileCount =
- Long.parseLong(
-
parameters.getOrDefault(NUM_FILES_PROP, "1"));
- long lastFileCreationTime =
- Long.parseLong(
- parameters.getOrDefault(
-
LAST_UPDATE_TIME_PROP,
-
System.currentTimeMillis() + ""));
- int totalBuckets =
- Integer.parseInt(
-
parameters.getOrDefault(TOTAL_BUCKETS, "0"));
- return new
org.apache.paimon.partition.Partition(
- Collections.singletonMap(
- tagToPartitionField,
part.getValues().get(0)),
- recordCount,
- fileSizeInBytes,
- fileCount,
- lastFileCreationTime,
- totalBuckets,
- false);
- })
- .collect(Collectors.toList());
+ List<Partition> hivePartitions =
listPartitionsFromHms(identifier);
+ partitions =
+ hivePartitions.stream()
+ .map(
+ part -> {
+ Map<String, String> parameters =
part.getParameters();
+ long recordCount =
+ Long.parseLong(
+
parameters.getOrDefault(
+
NUM_ROWS_PROP, "1"));
+ long fileSizeInBytes =
+ Long.parseLong(
+
parameters.getOrDefault(
+
TOTAL_SIZE_PROP, "1"));
+ long fileCount =
+ Long.parseLong(
+
parameters.getOrDefault(
+
NUM_FILES_PROP, "1"));
+ long lastFileCreationTime =
+ Long.parseLong(
+
parameters.getOrDefault(
+
LAST_UPDATE_TIME_PROP,
+
System.currentTimeMillis()
+ +
""));
+ int totalBuckets =
+ Integer.parseInt(
+
parameters.getOrDefault(
+
TOTAL_BUCKETS, "0"));
+ return new
org.apache.paimon.partition.Partition(
+ Collections.singletonMap(
+
tagToPartitionField,
+
part.getValues().get(0)),
+ recordCount,
+ fileSizeInBytes,
+ fileCount,
+ lastFileCreationTime,
+ totalBuckets,
+ false);
+ })
+ .collect(Collectors.toList());
} catch (Exception e) {
throw new RuntimeException(e);
}
+ } else {
+ partitions = listPartitionsFromFileSystem(table);
+ }
+
+ if (coreOptions.partitionedTableInMetastore()
+ && coreOptions
+ .partitionMarkDoneActions()
+
.contains(CoreOptions.PartitionMarkDoneAction.MARK_EVENT)) {
+ return withDoneStatus(identifier, partitions);
}
- return listPartitionsFromFileSystem(table);
+ return partitions;
+ }
+
+ private List<org.apache.paimon.partition.Partition> withDoneStatus(
+ Identifier identifier, List<org.apache.paimon.partition.Partition>
partitions)
+ throws TableNotExistException {
+ try {
+ return clients()
+ .run(
+ client -> {
+ List<org.apache.paimon.partition.Partition>
result =
+ new ArrayList<>(partitions.size());
+ for (org.apache.paimon.partition.Partition
partition : partitions) {
+ boolean done =
+ client.isPartitionMarkedForEvent(
+
identifier.getDatabaseName(),
+ identifier.getTableName(),
+ partition.spec(),
+
PartitionEventType.LOAD_DONE);
+ result.add(copyWithDone(partition, done));
+ }
+ return result;
+ });
+ } catch (UnknownTableException e) {
+ throw new TableNotExistException(identifier);
+ } catch (TException | InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private org.apache.paimon.partition.Partition copyWithDone(
+ org.apache.paimon.partition.Partition partition, boolean done) {
+ return new org.apache.paimon.partition.Partition(
+ partition.spec(),
+ partition.recordCount(),
+ partition.fileSizeInBytes(),
+ partition.fileCount(),
+ partition.lastFileCreationTime(),
+ partition.totalBuckets(),
+ done,
+ partition.createdAt(),
+ partition.createdBy(),
+ partition.updatedAt(),
+ partition.updatedBy(),
+ partition.options());
}
@VisibleForTesting
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 e1053c4715..d158f0398c 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
@@ -25,6 +25,8 @@ import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.CatalogTestBase;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.client.ClientPool;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
import org.apache.paimon.fs.Path;
import org.apache.paimon.options.CatalogOptions;
import org.apache.paimon.options.Options;
@@ -33,6 +35,9 @@ import org.apache.paimon.partition.PartitionStatistics;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.table.object.ObjectTable;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.utils.CommonTestUtils;
@@ -579,6 +584,48 @@ public class HiveCatalogTest extends CatalogTestBase {
Collections.singletonMap("dt", "20250101"));
}
+ @Test
+ public void testListPartitionsWithDoneStatus() throws Exception {
+ String databaseName = "testListPartitionsWithDoneStatus";
+ catalog.createDatabase(databaseName, false);
+ Identifier identifier = Identifier.create(databaseName, "table");
+ catalog.createTable(
+ identifier,
+ Schema.newBuilder()
+ .option(METASTORE_PARTITIONED_TABLE.key(), "true")
+ .option(CoreOptions.PARTITION_MARK_DONE_ACTION.key(),
"mark-event")
+ .column("col", DataTypes.INT())
+ .column("dt", DataTypes.STRING())
+ .partitionKeys("dt")
+ .build(),
+ false);
+
+ List<Map<String, String>> partitionSpecs =
+ Arrays.asList(
+ Collections.singletonMap("dt", "20250101"),
+ Collections.singletonMap("dt", "20250102"));
+ BatchWriteBuilder writeBuilder =
catalog.getTable(identifier).newBatchWriteBuilder();
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ for (Map<String, String> partitionSpec : partitionSpecs) {
+ write.write(GenericRow.of(0,
BinaryString.fromString(partitionSpec.get("dt"))));
+ }
+ commit.commit(write.prepareCommit());
+ }
+
+ assertThat(catalog.listPartitions(identifier)).allMatch(partition ->
!partition.done());
+
+ catalog.markDonePartitions(identifier,
Collections.singletonList(partitionSpecs.get(0)));
+
+ Map<Map<String, String>, Boolean> doneByPartition = new HashMap<>();
+ for (Partition partition : catalog.listPartitions(identifier)) {
+ doneByPartition.put(partition.spec(), partition.done());
+ }
+ assertThat(doneByPartition)
+ .containsEntry(partitionSpecs.get(0), true)
+ .containsEntry(partitionSpecs.get(1), false);
+ }
+
@Test
public void testCreateTableWithBlob() throws Exception {
String databaseName = "testCreateTableWithBlob";
