You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2024/01/09 07:45:12 UTC
(incubator-paimon) branch master updated: [core] Add partition modify timestamp to partition system table (#2579)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 860e4cd78 [core] Add partition modify timestamp to partition system table (#2579)
860e4cd78 is described below
commit 860e4cd78d5971c216f765fb5fb208261d88c3f5
Author: Aitozi <10...@qq.com>
AuthorDate: Tue Jan 9 15:45:07 2024 +0800
[core] Add partition modify timestamp to partition system table (#2579)
---
.../paimon/table/system/PartitionsTable.java | 50 +++++++++++++++++++---
.../paimon/table/system/PartitionsTableTest.java | 12 +++++-
2 files changed, 56 insertions(+), 6 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/PartitionsTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/PartitionsTable.java
index 9dacf9d5f..877f9b7ef 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/PartitionsTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/PartitionsTable.java
@@ -22,6 +22,7 @@ import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.LazyGenericRow;
+import org.apache.paimon.data.Timestamp;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.predicate.Predicate;
@@ -38,6 +39,7 @@ import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.table.source.TableScan;
import org.apache.paimon.types.BigIntType;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.IteratorRecordReader;
import org.apache.paimon.utils.ProjectedRow;
@@ -47,6 +49,9 @@ import org.apache.paimon.utils.SerializationUtils;
import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
import java.io.IOException;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -73,7 +78,9 @@ public class PartitionsTable implements ReadonlyTable {
Arrays.asList(
new DataField(0, "partition", SerializationUtils.newStringType(true)),
new DataField(1, "record_count", new BigIntType(false)),
- new DataField(2, "file_size_in_bytes", new BigIntType(false))));
+ new DataField(2, "file_size_in_bytes", new BigIntType(false)),
+ new DataField(3, "file_count", new BigIntType(false)),
+ new DataField(4, "last_update_time", DataTypes.TIMESTAMP_MILLIS())));
private final FileStoreTable storeTable;
@@ -241,7 +248,10 @@ public class PartitionsTable implements ReadonlyTable {
@SuppressWarnings("unchecked")
Supplier<Object>[] fields =
new Supplier[] {
- () -> partitionId, dataFileMeta::rowCount, dataFileMeta::fileSize
+ () -> partitionId,
+ dataFileMeta::rowCount,
+ dataFileMeta::fileSize,
+ dataFileMeta::creationTimeEpochMillis
};
return new LazyGenericRow(fields);
@@ -270,13 +280,17 @@ public class PartitionsTable implements ReadonlyTable {
BinaryString partitionId = row.getString(0);
long recordCount = row.getLong(1);
long fileSizeInBytes = row.getLong(2);
+ long lastFileCreationTime = row.getLong(3);
// Grouping and summing
Partition rowData =
groupedData.computeIfAbsent(
- partitionId, key -> new Partition(partitionId, 0, 0));
+ partitionId, key -> new Partition(partitionId, 0, 0, 0, -1));
rowData.recordCount += recordCount;
rowData.fileSizeInBytes += fileSizeInBytes;
+ rowData.fileCount++;
+ rowData.lastFileCreationTime =
+ Math.max(rowData.lastFileCreationTime, lastFileCreationTime);
}
resultIterator = groupedData.values().iterator();
}
@@ -291,7 +305,14 @@ public class PartitionsTable implements ReadonlyTable {
if (hasNext()) {
Partition partition = resultIterator.next();
return GenericRow.of(
- partition.partition, partition.recordCount, partition.fileSizeInBytes);
+ partition.partition,
+ partition.recordCount,
+ partition.fileSizeInBytes,
+ partition.fileCount,
+ Timestamp.fromLocalDateTime(
+ LocalDateTime.ofInstant(
+ Instant.ofEpochMilli(partition.lastFileCreationTime),
+ ZoneId.systemDefault())));
} else {
throw new NoSuchElementException("No more elements in the iterator.");
}
@@ -303,10 +324,21 @@ public class PartitionsTable implements ReadonlyTable {
private long recordCount;
private long fileSizeInBytes;
- Partition(BinaryString partition, long recordCount, long fileSizeInBytes) {
+ private long fileCount;
+
+ private long lastFileCreationTime;
+
+ Partition(
+ BinaryString partition,
+ long recordCount,
+ long fileSizeInBytes,
+ long fileCount,
+ long lastFileCreationTime) {
this.partition = partition;
this.recordCount = recordCount;
this.fileSizeInBytes = fileSizeInBytes;
+ this.lastFileCreationTime = lastFileCreationTime;
+ this.fileCount = fileCount;
}
public long recordCount() {
@@ -316,5 +348,13 @@ public class PartitionsTable implements ReadonlyTable {
public long fileSize() {
return fileSizeInBytes;
}
+
+ public long getFileCount() {
+ return fileCount;
+ }
+
+ public long getLastFileCreationTime() {
+ return lastFileCreationTime;
+ }
}
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/system/PartitionsTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/system/PartitionsTableTest.java
index 0a9a0fe52..27591b9ca 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/system/PartitionsTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/system/PartitionsTableTest.java
@@ -67,7 +67,6 @@ public class PartitionsTableTest extends TableTestBase {
.partitionKeys("pt")
.primaryKey("pk", "pt")
.option(CoreOptions.CHANGELOG_PRODUCER.key(), "input")
- .option(CoreOptions.BUCKET.key(), "2")
.build();
TableSchema tableSchema =
SchemaUtils.forceCommit(new SchemaManager(fileIO, tablePath), schema);
@@ -93,4 +92,15 @@ public class PartitionsTableTest extends TableTestBase {
List<InternalRow> result = read(partitionsTable, new int[][] {{0}, {1}});
assertThat(result).containsExactlyInAnyOrderElementsOf(expectedRow);
}
+
+ @Test
+ public void testPartitionValue() throws Exception {
+ write(table, GenericRow.of(2, 1, 3), GenericRow.of(3, 1, 4));
+ List<InternalRow> expectedRow = new ArrayList<>();
+ expectedRow.add(GenericRow.of(BinaryString.fromString("[1]"), 4L, 3L));
+ expectedRow.add(GenericRow.of(BinaryString.fromString("[2]"), 2L, 2L));
+
+ List<InternalRow> result = read(partitionsTable, new int[][] {{0}, {1}, {3}});
+ assertThat(result).containsExactlyInAnyOrderElementsOf(expectedRow);
+ }
}