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 2023/06/21 08:16:38 UTC
[incubator-paimon] branch master updated: [core] Introduce tags metadata table (#1417)
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 66e2780bf [core] Introduce tags metadata table (#1417)
66e2780bf is described below
commit 66e2780bfc511c23909c81afd3e4266c8e322f68
Author: yuzelin <33...@users.noreply.github.com>
AuthorDate: Wed Jun 21 16:16:34 2023 +0800
[core] Introduce tags metadata table (#1417)
---
.../paimon/table/system/SystemTableLoader.java | 3 +
.../org/apache/paimon/table/system/TagsTable.java | 209 +++++++++++++++++++++
.../java/org/apache/paimon/utils/TagManager.java | 43 +++--
3 files changed, 241 insertions(+), 14 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java b/paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java
index 97c117617..01fa2432d 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java
@@ -30,6 +30,7 @@ import static org.apache.paimon.table.system.FilesTable.FILES;
import static org.apache.paimon.table.system.OptionsTable.OPTIONS;
import static org.apache.paimon.table.system.SchemasTable.SCHEMAS;
import static org.apache.paimon.table.system.SnapshotsTable.SNAPSHOTS;
+import static org.apache.paimon.table.system.TagsTable.TAGS;
/** Loader to load system {@link Table}s. */
public class SystemTableLoader {
@@ -48,6 +49,8 @@ public class SystemTableLoader {
return new AuditLogTable(dataTable);
case FILES:
return new FilesTable(dataTable);
+ case TAGS:
+ return new TagsTable(fileIO, location);
default:
return null;
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/TagsTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/TagsTable.java
new file mode 100644
index 000000000..a7b6ba447
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/TagsTable.java
@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.system;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.ReadonlyTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.source.InnerTableRead;
+import org.apache.paimon.table.source.InnerTableScan;
+import org.apache.paimon.table.source.ReadOnceTableScan;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.types.BigIntType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.TimestampType;
+import org.apache.paimon.utils.DateTimeUtils;
+import org.apache.paimon.utils.IteratorRecordReader;
+import org.apache.paimon.utils.ProjectedRow;
+import org.apache.paimon.utils.SerializationUtils;
+import org.apache.paimon.utils.TagManager;
+
+import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.SortedMap;
+
+import static org.apache.paimon.catalog.Catalog.SYSTEM_TABLE_SPLITTER;
+
+/** A {@link Table} for showing tags of table. */
+public class TagsTable implements ReadonlyTable {
+
+ private static final long serialVersionUID = 1L;
+
+ public static final String TAGS = "tags";
+
+ public static final RowType TABLE_TYPE =
+ new RowType(
+ Arrays.asList(
+ new DataField(0, "tag_name", SerializationUtils.newStringType(false)),
+ new DataField(1, "snapshot_id", new BigIntType(false)),
+ new DataField(2, "schema_id", new BigIntType(false)),
+ new DataField(3, "commit_time", new TimestampType(false, 3)),
+ new DataField(4, "record_count", new BigIntType(true))));
+
+ private final FileIO fileIO;
+ private final Path location;
+
+ public TagsTable(FileIO fileIO, Path location) {
+ this.fileIO = fileIO;
+ this.location = location;
+ }
+
+ @Override
+ public String name() {
+ return location.getName() + SYSTEM_TABLE_SPLITTER + TAGS;
+ }
+
+ @Override
+ public RowType rowType() {
+ return TABLE_TYPE;
+ }
+
+ @Override
+ public List<String> primaryKeys() {
+ return Collections.singletonList("tag_name");
+ }
+
+ @Override
+ public InnerTableScan newScan() {
+ return new TagsScan();
+ }
+
+ @Override
+ public InnerTableRead newRead() {
+ return new TagsRead(fileIO);
+ }
+
+ @Override
+ public Table copy(Map<String, String> dynamicOptions) {
+ return new TagsTable(fileIO, location);
+ }
+
+ private class TagsScan extends ReadOnceTableScan {
+
+ @Override
+ public InnerTableScan withFilter(Predicate predicate) {
+ // TODO
+ return this;
+ }
+
+ @Override
+ public Plan innerPlan() {
+ return () -> Collections.singletonList(new TagsSplit(fileIO, location));
+ }
+ }
+
+ private static class TagsSplit implements Split {
+ private static final long serialVersionUID = 1L;
+
+ private final FileIO fileIO;
+ private final Path location;
+
+ private TagsSplit(FileIO fileIO, Path location) {
+ this.fileIO = fileIO;
+ this.location = location;
+ }
+
+ @Override
+ public long rowCount() {
+ return new TagManager(fileIO, location).tagCount();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ TagsSplit that = (TagsSplit) o;
+ return Objects.equals(location, that.location);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(location);
+ }
+ }
+
+ private static class TagsRead implements InnerTableRead {
+
+ private final FileIO fileIO;
+ private int[][] projection;
+
+ public TagsRead(FileIO fileIO) {
+ this.fileIO = fileIO;
+ }
+
+ @Override
+ public InnerTableRead withFilter(Predicate predicate) {
+ // TODO
+ return this;
+ }
+
+ @Override
+ public InnerTableRead withProjection(int[][] projection) {
+ this.projection = projection;
+ return this;
+ }
+
+ @Override
+ public RecordReader<InternalRow> createReader(Split split) {
+ if (!(split instanceof TagsSplit)) {
+ throw new IllegalArgumentException("Unsupported split: " + split.getClass());
+ }
+ Path location = ((TagsSplit) split).location;
+ SortedMap<Snapshot, String> tags = new TagManager(fileIO, location).tags();
+ Iterator<InternalRow> rows =
+ Iterators.transform(tags.entrySet().iterator(), this::toRow);
+ if (projection != null) {
+ rows =
+ Iterators.transform(
+ rows, row -> ProjectedRow.from(projection).replaceRow(row));
+ }
+ return new IteratorRecordReader<>(rows);
+ }
+
+ private InternalRow toRow(Map.Entry<Snapshot, String> tag) {
+ Snapshot snapshot = tag.getKey();
+ return GenericRow.of(
+ BinaryString.fromString(tag.getValue()),
+ snapshot.id(),
+ snapshot.schemaId(),
+ Timestamp.fromLocalDateTime(
+ DateTimeUtils.toLocalDateTime(snapshot.timeMillis())),
+ snapshot.totalRecordCount());
+ }
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/TagManager.java b/paimon-core/src/main/java/org/apache/paimon/utils/TagManager.java
index ea3dce938..f16871d43 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/TagManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/TagManager.java
@@ -24,11 +24,11 @@ import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.Path;
import java.io.IOException;
-import java.util.Arrays;
-import java.util.Collections;
+import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
-import java.util.stream.Collectors;
+import java.util.SortedMap;
+import java.util.TreeMap;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -96,12 +96,34 @@ public class TagManager {
return Snapshot.fromPath(fileIO, tagPath(tagName));
}
- /** Get all tagged snapshots sorted by commit time. */
+ public int tagCount() {
+ return listStatus().length;
+ }
+
+ /** Get all tagged snapshots sorted by snapshot id. */
public List<Snapshot> taggedSnapshots() {
+ return new ArrayList<>(tags().keySet());
+ }
+
+ /** Get all tagged snapshots with names sorted by snapshot id. */
+ public SortedMap<Snapshot, String> tags() {
+ FileStatus[] statuses = listStatus();
+ TreeMap<Snapshot, String> tags = new TreeMap<>(Comparator.comparingLong(Snapshot::id));
+
+ for (FileStatus status : statuses) {
+ Path path = status.getPath();
+ if (path.getName().startsWith(TAG_PREFIX)) {
+ tags.put(Snapshot.fromPath(fileIO, path), path.getName());
+ }
+ }
+ return tags;
+ }
+
+ private FileStatus[] listStatus() {
Path tagDirectory = tagDirectory();
try {
if (!fileIO.exists(tagDirectory)) {
- return Collections.emptyList();
+ return new FileStatus[0];
}
FileStatus[] statuses = fileIO.listStatus(tagDirectory);
@@ -113,17 +135,10 @@ public class TagManager {
tagDirectory));
}
- return Arrays.stream(statuses)
- .map(FileStatus::getPath)
- .filter(path -> path.getName().startsWith(TAG_PREFIX))
- .map(path -> Snapshot.fromPath(fileIO, path))
- .sorted(Comparator.comparingLong(Snapshot::id))
- .collect(Collectors.toList());
+ return statuses;
} catch (IOException e) {
throw new RuntimeException(
- String.format(
- "Failed to get tagged snapshots in tag directory '%s'.", tagDirectory),
- e);
+ String.format("Failed to list status in the '%s' directory.", tagDirectory), e);
}
}
}