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:06:46 UTC
(incubator-paimon) branch master updated: [flink] Support the creation of tags based on the latest snapshot (#2533)
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 ea37f62c7 [flink] Support the creation of tags based on the latest snapshot (#2533)
ea37f62c7 is described below
commit ea37f62c7b19fc2c68ba72367bba6e1be5b583c0
Author: monster <60...@users.noreply.github.com>
AuthorDate: Mon Dec 25 15:06:42 2023 +0800
[flink] Support the creation of tags based on the latest snapshot (#2533)
---
docs/content/maintenance/manage-tags.md | 9 ++++-
.../paimon/flink/action/CreateTagAction.java | 12 ++++--
.../flink/action/CreateTagActionFactory.java | 9 +++--
.../paimon/flink/procedure/CreateTagProcedure.java | 14 +++++++
.../paimon/flink/action/TagActionITCase.java | 46 ++++++++++++++++++++++
.../sql/CreateAndDeleteTagProcedureTest.scala | 12 ++++++
.../paimon/spark/procedure/CreateTagProcedure.java | 15 ++++++-
.../CreateAndDeleteTagProcedureTest.scala | 12 ++++++
8 files changed, 120 insertions(+), 9 deletions(-)
diff --git a/docs/content/maintenance/manage-tags.md b/docs/content/maintenance/manage-tags.md
index 865e68b5e..cbc2d859a 100644
--- a/docs/content/maintenance/manage-tags.md
+++ b/docs/content/maintenance/manage-tags.md
@@ -108,10 +108,12 @@ You can create a tag with given name (cannot be number) and snapshot ID.
--database <database-name> \
--table <table-name> \
--tag_name <tag-name> \
- --snapshot <snapshot-id> \
+ [--snapshot <snapshot_id>] \
[--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]]
```
+If `snapshot` unset, snapshot_id defaults to the latest.
+
{{< /tab >}}
{{< tab "Java API" >}}
@@ -136,6 +138,11 @@ Run the following sql:
CALL create_tag(table => 'test.T', tag => 'test_tag', snapshot => 2);
```
+To create a tag based on the latest snapshot id, run the following sql:
+```sql
+CALL create_tag(table => 'test.T', tag => 'test_tag');
+```
+
{{< /tab >}}
{{< /tabs >}}
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 948cec406..41b35c598 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,13 +18,17 @@
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 {
private final String tagName;
- private final long snapshotId;
+ private final Long snapshotId;
public CreateTagAction(
String warehouse,
@@ -32,7 +36,7 @@ public class CreateTagAction extends TableActionBase {
String tableName,
Map<String, String> catalogConfig,
String tagName,
- long snapshotId) {
+ Long snapshotId) {
super(warehouse, databaseName, tableName, catalogConfig);
this.tagName = tagName;
this.snapshotId = snapshotId;
@@ -40,6 +44,8 @@ public class CreateTagAction extends TableActionBase {
@Override
public void run() throws Exception {
- table.createTag(tagName, snapshotId);
+ SnapshotManager snapshotManager = ((AbstractFileStoreTable) table).snapshotManager();
+ Long idToUse = snapshotId == null ? snapshotManager.latestSnapshotId() : snapshotId;
+ table.createTag(tagName, Objects.requireNonNull(idToUse));
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CreateTagActionFactory.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CreateTagActionFactory.java
index bfc9510d7..01bbdba42 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CreateTagActionFactory.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CreateTagActionFactory.java
@@ -39,12 +39,15 @@ public class CreateTagActionFactory implements ActionFactory {
@Override
public Optional<Action> create(MultipleParameterToolAdapter params) {
checkRequiredArgument(params, TAG_NAME);
- checkRequiredArgument(params, SNAPSHOT);
Tuple3<String, String, String> tablePath = getTablePath(params);
Map<String, String> catalogConfig = optionalConfigMap(params, CATALOG_CONF);
String tagName = params.get(TAG_NAME);
- long snapshot = Long.parseLong(params.get(SNAPSHOT));
+
+ Long snapshot = null;
+ if (params.has(SNAPSHOT)) {
+ snapshot = Long.parseLong(params.get(SNAPSHOT));
+ }
CreateTagAction action =
new CreateTagAction(
@@ -60,7 +63,7 @@ public class CreateTagActionFactory implements ActionFactory {
System.out.println("Syntax:");
System.out.println(
" create_tag --warehouse <warehouse_path> --database <database_name> "
- + "--table <table_name> --tag_name <tag_name> --snapshot <snapshot_id>");
+ + "--table <table_name> --tag_name <tag_name> [--snapshot <snapshot_id>]");
System.out.println();
}
}
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 9cd4067bc..9ebbccfa6 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,10 +20,14 @@ 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;
+
/**
* Create tag procedure. Usage:
*
@@ -44,6 +48,16 @@ public class CreateTagProcedure extends ProcedureBase {
return new String[] {"Success"};
}
+ 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());
+
+ table.createTag(tagName, latestSnapshotId);
+ return new String[] {"Success"};
+ }
+
@Override
public String identifier() {
return IDENTIFIER;
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/TagActionITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/TagActionITCase.java
index b5c616707..a01849ed9 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/TagActionITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/TagActionITCase.java
@@ -111,4 +111,50 @@ public class TagActionITCase extends ActionITCaseBase {
}
assertThat(tagManager.tagExists("tag2")).isFalse();
}
+
+ @Test
+ public void testCreateLatestTag() throws Exception {
+ init(warehouse);
+
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.BIGINT(), DataTypes.STRING()},
+ new String[] {"k", "v"});
+ FileStoreTable table =
+ createFileStoreTable(
+ rowType,
+ Collections.emptyList(),
+ Collections.singletonList("k"),
+ Collections.emptyMap());
+
+ StreamWriteBuilder writeBuilder = table.newStreamWriteBuilder().withCommitUser(commitUser);
+ write = writeBuilder.newWrite();
+ commit = writeBuilder.newCommit();
+
+ // 3 snapshots
+ writeData(rowData(1L, BinaryString.fromString("Hi")));
+ writeData(rowData(2L, BinaryString.fromString("Hello")));
+ writeData(rowData(3L, BinaryString.fromString("Paimon")));
+
+ TagManager tagManager = new TagManager(table.fileIO(), table.location());
+
+ if (ThreadLocalRandom.current().nextBoolean()) {
+ createAction(
+ CreateTagAction.class,
+ "create_tag",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName,
+ "--tag_name",
+ "tag2")
+ .run();
+ } else {
+ callProcedure(
+ String.format("CALL sys.create_tag('%s.%s', 'tag2', 2)", database, tableName));
+ }
+ assertThat(tagManager.tagExists("tag2")).isTrue();
+ }
}
diff --git a/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/CreateAndDeleteTagProcedureTest.scala b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/CreateAndDeleteTagProcedureTest.scala
index 30e112ab5..b1c469b8c 100644
--- a/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/CreateAndDeleteTagProcedureTest.scala
+++ b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/CreateAndDeleteTagProcedureTest.scala
@@ -78,6 +78,18 @@ class CreateAndDeleteTagProcedureTest extends PaimonSparkTestBase with StreamTes
spark.sql("CALL paimon.sys.delete_tag(table => 'test.T', tag => 'test_tag')"),
Row(true) :: Nil)
checkAnswer(spark.sql("SELECT tag_name FROM paimon.test.`T$tags`"), Nil)
+ checkAnswer(
+ spark.sql(
+ "CALL paimon.sys.create_tag(table => 'test.T', tag => 'test_latestSnapshot_tag')"),
+ Row(true) :: Nil)
+ checkAnswer(
+ spark.sql("SELECT tag_name FROM paimon.test.`T$tags`"),
+ Row("test_latestSnapshot_tag") :: Nil)
+ checkAnswer(
+ spark.sql(
+ "CALL paimon.sys.delete_tag(table => 'test.T', tag => 'test_latestSnapshot_tag')"),
+ Row(true) :: Nil)
+ checkAnswer(spark.sql("SELECT tag_name FROM paimon.test.`T$tags`"), Nil)
} finally {
stream.stop()
}
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 c0c63b9b0..1389dbc01 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,6 +18,8 @@
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;
@@ -26,6 +28,8 @@ 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;
@@ -36,7 +40,7 @@ public class CreateTagProcedure extends BaseProcedure {
new ProcedureParameter[] {
ProcedureParameter.required("table", StringType),
ProcedureParameter.required("tag", StringType),
- ProcedureParameter.required("snapshot", LongType)
+ ProcedureParameter.optional("snapshot", LongType)
};
private static final StructType OUTPUT_TYPE =
@@ -68,7 +72,14 @@ public class CreateTagProcedure extends BaseProcedure {
return modifyPaimonTable(
tableIdent,
table -> {
- table.createTag(tag, snapshot);
+ long snapshotToUse =
+ (snapshot == 0)
+ ? Objects.requireNonNull(
+ ((AbstractFileStoreTable) table)
+ .snapshotManager()
+ .latestSnapshotId())
+ : snapshot;
+ table.createTag(tag, snapshotToUse);
InternalRow outputRow = newInternalRow(true);
return new InternalRow[] {outputRow};
});
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CreateAndDeleteTagProcedureTest.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CreateAndDeleteTagProcedureTest.scala
index f9f559273..0e642979a 100644
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CreateAndDeleteTagProcedureTest.scala
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CreateAndDeleteTagProcedureTest.scala
@@ -78,6 +78,18 @@ class CreateAndDeleteTagProcedureTest extends PaimonSparkTestBase with StreamTes
spark.sql("CALL paimon.sys.delete_tag(table => 'test.T', tag => 'test_tag')"),
Row(true) :: Nil)
checkAnswer(spark.sql("SELECT tag_name FROM paimon.test.`T$tags`"), Nil)
+ checkAnswer(
+ spark.sql(
+ "CALL paimon.sys.create_tag(table => 'test.T', tag => 'test_latestSnapshot_tag')"),
+ Row(true) :: Nil)
+ checkAnswer(
+ spark.sql("SELECT tag_name FROM paimon.test.`T$tags`"),
+ Row("test_latestSnapshot_tag") :: Nil)
+ checkAnswer(
+ spark.sql(
+ "CALL paimon.sys.delete_tag(table => 'test.T', tag => 'test_latestSnapshot_tag')"),
+ Row(true) :: Nil)
+ checkAnswer(spark.sql("SELECT tag_name FROM paimon.test.`T$tags`"), Nil)
} finally {
stream.stop()
}