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/12/25 07:10:25 UTC
(incubator-paimon) branch master updated: [core] Introduce Table.createTag(tag) from latest snapshot
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 cd07c35e1 [core] Introduce Table.createTag(tag) from latest snapshot
cd07c35e1 is described below
commit cd07c35e1e463b1eabb846f70f488921b2aeb51f
Author: Jingsong <ji...@gmail.com>
AuthorDate: Mon Dec 25 15:07:09 2023 +0800
[core] Introduce Table.createTag(tag) from latest snapshot
---
.../paimon/table/AbstractFileStoreTable.java | 14 +++++++++++--
.../org/apache/paimon/table/ReadonlyTable.java | 8 ++++++++
.../main/java/org/apache/paimon/table/Table.java | 4 ++++
.../paimon/flink/action/CreateTagAction.java | 12 +++++------
.../paimon/flink/procedure/CreateTagProcedure.java | 23 +++++++++++-----------
.../paimon/spark/procedure/CreateTagProcedure.java | 19 ++++++------------
6 files changed, 47 insertions(+), 33 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
index 41f0b447f..d7f468ed1 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
@@ -65,6 +65,7 @@ import java.util.function.BiConsumer;
import static org.apache.paimon.CoreOptions.PATH;
import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
/** Abstract {@link FileStoreTable}. */
public abstract class AbstractFileStoreTable implements FileStoreTable {
@@ -359,9 +360,18 @@ public abstract class AbstractFileStoreTable implements FileStoreTable {
snapshotManager.snapshotExists(fromSnapshotId),
"Cannot create tag because given snapshot #%s doesn't exist.",
fromSnapshotId);
+ createTag(tagName, snapshotManager.snapshot(fromSnapshotId));
+ }
+
+ @Override
+ public void createTag(String tagName) {
+ Snapshot latestSnapshot = snapshotManager().latestSnapshot();
+ checkNotNull(latestSnapshot, "Cannot create tag because latest snapshot doesn't exist.");
+ createTag(tagName, latestSnapshot);
+ }
- Snapshot snapshot = snapshotManager.snapshot(fromSnapshotId);
- tagManager().createTag(snapshot, tagName, store().createTagCallbacks());
+ private void createTag(String tagName, Snapshot fromSnapshot) {
+ tagManager().createTag(fromSnapshot, tagName, store().createTagCallbacks());
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/ReadonlyTable.java b/paimon-core/src/main/java/org/apache/paimon/table/ReadonlyTable.java
index 86dd1f97d..ca2dce0bd 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/ReadonlyTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/ReadonlyTable.java
@@ -103,6 +103,14 @@ public interface ReadonlyTable extends InnerTable {
this.getClass().getSimpleName()));
}
+ @Override
+ default void createTag(String tagName) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Readonly Table %s does not support createTag.",
+ this.getClass().getSimpleName()));
+ }
+
@Override
default void deleteTag(String tagName) {
throw new UnsupportedOperationException(
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/Table.java b/paimon-core/src/main/java/org/apache/paimon/table/Table.java
index c1b4b7882..3e6261b06 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/Table.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/Table.java
@@ -71,6 +71,10 @@ public interface Table extends Serializable {
@Experimental
void createTag(String tagName, long fromSnapshotId);
+ /** Create a tag from latest snapshot. */
+ @Experimental
+ void createTag(String tagName);
+
/** Delete a tag by name. */
@Experimental
void deleteTag(String tagName);
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CreateTagAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CreateTagAction.java
index 41b35c598..0e500c3e7 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CreateTagAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CreateTagAction.java
@@ -18,11 +18,7 @@
package org.apache.paimon.flink.action;
-import org.apache.paimon.table.AbstractFileStoreTable;
-import org.apache.paimon.utils.SnapshotManager;
-
import java.util.Map;
-import java.util.Objects;
/** Create tag action for Flink. */
public class CreateTagAction extends TableActionBase {
@@ -44,8 +40,10 @@ public class CreateTagAction extends TableActionBase {
@Override
public void run() throws Exception {
- SnapshotManager snapshotManager = ((AbstractFileStoreTable) table).snapshotManager();
- Long idToUse = snapshotId == null ? snapshotManager.latestSnapshotId() : snapshotId;
- table.createTag(tagName, Objects.requireNonNull(idToUse));
+ if (snapshotId == null) {
+ table.createTag(tagName);
+ } else {
+ table.createTag(tagName, snapshotId);
+ }
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateTagProcedure.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateTagProcedure.java
index 9ebbccfa6..9999fdf11 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateTagProcedure.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateTagProcedure.java
@@ -20,13 +20,11 @@ package org.apache.paimon.flink.procedure;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.Identifier;
-import org.apache.paimon.table.AbstractFileStoreTable;
import org.apache.paimon.table.Table;
-import org.apache.paimon.utils.SnapshotManager;
import org.apache.flink.table.procedure.ProcedureContext;
-import java.util.Objects;
+import javax.annotation.Nullable;
/**
* Create tag procedure. Usage:
@@ -42,19 +40,22 @@ public class CreateTagProcedure extends ProcedureBase {
public String[] call(
ProcedureContext procedureContext, String tableId, String tagName, long snapshotId)
throws Catalog.TableNotExistException {
- Table table = catalog.getTable(Identifier.fromString(tableId));
- table.createTag(tagName, snapshotId);
-
- return new String[] {"Success"};
+ return innerCall(tableId, tagName, snapshotId);
}
public String[] call(ProcedureContext procedureContext, String tableId, String tagName)
throws Catalog.TableNotExistException {
- Table table = catalog.getTable(Identifier.fromString(tableId));
- SnapshotManager snapshotManager = ((AbstractFileStoreTable) table).snapshotManager();
- long latestSnapshotId = Objects.requireNonNull(snapshotManager.latestSnapshotId());
+ return innerCall(tableId, tagName, null);
+ }
- table.createTag(tagName, latestSnapshotId);
+ private String[] innerCall(String tableId, String tagName, @Nullable Long snapshotId)
+ throws Catalog.TableNotExistException {
+ Table table = catalog.getTable(Identifier.fromString(tableId));
+ if (snapshotId == null) {
+ table.createTag(tagName);
+ } else {
+ table.createTag(tagName, snapshotId);
+ }
return new String[] {"Success"};
}
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreateTagProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreateTagProcedure.java
index 1389dbc01..4639dddb9 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreateTagProcedure.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreateTagProcedure.java
@@ -18,8 +18,6 @@
package org.apache.paimon.spark.procedure;
-import org.apache.paimon.table.AbstractFileStoreTable;
-
import org.apache.spark.sql.catalyst.InternalRow;
import org.apache.spark.sql.connector.catalog.Identifier;
import org.apache.spark.sql.connector.catalog.TableCatalog;
@@ -28,8 +26,6 @@ import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
-import java.util.Objects;
-
import static org.apache.spark.sql.types.DataTypes.LongType;
import static org.apache.spark.sql.types.DataTypes.StringType;
@@ -67,19 +63,16 @@ public class CreateTagProcedure extends BaseProcedure {
public InternalRow[] call(InternalRow args) {
Identifier tableIdent = toIdentifier(args.getString(0), PARAMETERS[0].name());
String tag = args.getString(1);
- long snapshot = args.getLong(2);
+ Long snapshot = args.isNullAt(2) ? null : args.getLong(2);
return modifyPaimonTable(
tableIdent,
table -> {
- long snapshotToUse =
- (snapshot == 0)
- ? Objects.requireNonNull(
- ((AbstractFileStoreTable) table)
- .snapshotManager()
- .latestSnapshotId())
- : snapshot;
- table.createTag(tag, snapshotToUse);
+ if (snapshot == null) {
+ table.createTag(tag);
+ } else {
+ table.createTag(tag, snapshot);
+ }
InternalRow outputRow = newInternalRow(true);
return new InternalRow[] {outputRow};
});