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/08/27 02:39:58 UTC
(paimon) branch master updated: [core] Supports creating tags from snapshots commit-time. (#4051)
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/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new db77eec5b [core] Supports creating tags from snapshots commit-time. (#4051)
db77eec5b is described below
commit db77eec5b73aef22657a40eddf44804f8eb279c6
Author: HunterXHunter <cn...@gmail.com>
AuthorDate: Tue Aug 27 10:39:52 2024 +0800
[core] Supports creating tags from snapshots commit-time. (#4051)
---
docs/content/flink/procedures.md | 20 +++
docs/content/spark/procedures.md | 13 ++
.../org/apache/paimon/utils/SnapshotManager.java | 33 ++++
.../apache/paimon/utils/SnapshotManagerTest.java | 30 ++++
.../procedure/CreateTagFromTimestampProcedure.java | 107 ++++++++++++
.../services/org.apache.paimon.factories.Factory | 1 +
.../CreateTagFromTimestampProcedureITCase.java | 167 +++++++++++++++++++
.../org/apache/paimon/spark/SparkProcedures.java | 3 +
.../procedure/CreateTagFromTimestampProcedure.java | 136 ++++++++++++++++
.../CreateTagFromTimestampProcedureTest.scala | 179 +++++++++++++++++++++
10 files changed, 689 insertions(+)
diff --git a/docs/content/flink/procedures.md b/docs/content/flink/procedures.md
index f028659a5..c8c8b5c48 100644
--- a/docs/content/flink/procedures.md
+++ b/docs/content/flink/procedures.md
@@ -121,6 +121,26 @@ All available procedures are listed below.
<td>
CALL sys.create_tag('default.T', 'my_tag', 10, '1 d')
</td>
+ </tr>
+ <tr>
+ <td>create_tag_from_timestamp</td>
+ <td>
+ -- Create a tag from the first snapshot whose commit-time greater than the specified timestamp. <br/>
+ CALL [catalog.]sys.create_tag_from_timestamp('identifier', 'tagName', timestamp, time_retained)
+ </td>
+ <td>
+ To create a tag based on given timestamp. Arguments:
+ <li>identifier: the target table identifier. Cannot be empty.</li>
+ <li>tag: name of the new tag.</li>
+ <li>timestamp (Long): Find the first snapshot whose commit-time greater than this timestamp.</li>
+ <li>time_retained : The maximum time retained for newly created tags.</li>
+ </td>
+ <td>
+ -- for Flink 1.18<br/>
+ CALL sys.create_tag_from_timestamp('default.T', 'my_tag', 1724404318750, '1 d')
+ -- for Flink 1.19 and later<br/>
+ CALL sys.create_tag_from_timestamp(`table` => 'default.T', `tag` => 'my_tag', `timestamp` => 1724404318750, time_retained => '1 d')
+ </td>
</tr>
<tr>
<td>delete_tag</td>
diff --git a/docs/content/spark/procedures.md b/docs/content/spark/procedures.md
index cba00296e..249ede0e8 100644
--- a/docs/content/spark/procedures.md
+++ b/docs/content/spark/procedures.md
@@ -96,6 +96,19 @@ This section introduce all available spark procedures about paimon.
CALL sys.create_tag(table => 'default.T', tag => 'my_tag')
</td>
</tr>
+ <tr>
+ <td>create_tag_from_timestamp</td>
+ <td>
+ To create a tag based on given timestamp. Arguments:
+ <li>identifier: the target table identifier. Cannot be empty.</li>
+ <li>tag: name of the new tag.</li>
+ <li>timestamp (Long): Find the first snapshot whose commit-time is greater than this timestamp.</li>
+ <li>time_retained : The maximum time retained for newly created tags.</li>
+ </td>
+ <td>
+ CALL sys.create_tag_from_timestamp(`table` => 'default.T', `tag` => 'my_tag', `timestamp` => 1724404318750, time_retained => '1 d')
+ </td>
+ </tr>
<tr>
<td>delete_tag</td>
<td>
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java b/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
index d64164d53..7a1508544 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
@@ -292,6 +292,39 @@ public class SnapshotManager implements Serializable {
return finalSnapshot;
}
+ /**
+ * Returns a {@link Snapshot} whoes commit time is later than or equal to given timestamp mills.
+ * If there is no such a snapshot, returns null.
+ */
+ public @Nullable Snapshot laterOrEqualTimeMills(long timestampMills) {
+ Long earliest = earliestSnapshotId();
+ Long latest = latestSnapshotId();
+ if (earliest == null || latest == null) {
+ return null;
+ }
+
+ Snapshot latestSnapShot = snapshot(latest);
+ if (latestSnapShot.timeMillis() < timestampMills) {
+ return null;
+ }
+ Snapshot finalSnapshot = null;
+ while (earliest <= latest) {
+ long mid = earliest + (latest - earliest) / 2; // Avoid overflow
+ Snapshot snapshot = snapshot(mid);
+ long commitTime = snapshot.timeMillis();
+ if (commitTime > timestampMills) {
+ latest = mid - 1; // Search in the left half
+ finalSnapshot = snapshot;
+ } else if (commitTime < timestampMills) {
+ earliest = mid + 1; // Search in the right half
+ } else {
+ finalSnapshot = snapshot; // Found the exact match
+ break;
+ }
+ }
+ return finalSnapshot;
+ }
+
public @Nullable Snapshot laterOrEqualWatermark(long watermark) {
Long earliest = earliestSnapshotId();
Long latest = latestSnapshotId();
diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java
index 4e0b18d47..544638d4b 100644
--- a/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java
@@ -112,6 +112,11 @@ public class SnapshotManagerTest {
Snapshot snapshot = createSnapshotWithMillis(i, millis + i * 1000);
localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i), snapshot.toJson());
}
+
+ // there is no snapshot smaller than "millis - 1L" return the earliest snapshot
+ assertThat(snapshotManager.earlierOrEqualTimeMills(millis - 1L).timeMillis())
+ .isEqualTo(millis);
+
// smaller than the second snapshot return the first snapshot
assertThat(snapshotManager.earlierOrEqualTimeMills(millis + 999).timeMillis())
.isEqualTo(millis);
@@ -123,6 +128,31 @@ public class SnapshotManagerTest {
.isEqualTo(millis + 1000);
}
+ @Test
+ public void testLaterOrEqualTimeMills() throws IOException {
+ long millis = 1684726826L;
+ FileIO localFileIO = LocalFileIO.create();
+ SnapshotManager snapshotManager =
+ new SnapshotManager(localFileIO, new Path(tempDir.toString()));
+ // create 10 snapshots
+ for (long i = 0; i < 10; i++) {
+ Snapshot snapshot = createSnapshotWithMillis(i, millis + i * 1000);
+ localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i), snapshot.toJson());
+ }
+ // smaller than the second snapshot return the second snapshot
+ assertThat(snapshotManager.laterOrEqualTimeMills(millis + 999).timeMillis())
+ .isEqualTo(millis + 1000);
+ // equal to the second snapshot return the second snapshot
+ assertThat(snapshotManager.laterOrEqualTimeMills(millis + 1000).timeMillis())
+ .isEqualTo(millis + 1000);
+ // larger than the second snapshot return the third snapshot
+ assertThat(snapshotManager.laterOrEqualTimeMills(millis + 1001).timeMillis())
+ .isEqualTo(millis + 2000);
+
+ // larger than the latest snapshot return null
+ assertThat(snapshotManager.laterOrEqualTimeMills(millis + 10001)).isNull();
+ }
+
@Test
public void testlaterOrEqualWatermark() throws IOException {
long millis = Long.MIN_VALUE;
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateTagFromTimestampProcedure.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateTagFromTimestampProcedure.java
new file mode 100644
index 000000000..1a7a0de3b
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateTagFromTimestampProcedure.java
@@ -0,0 +1,107 @@
+/*
+ * 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.flink.procedure;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.SnapshotNotExistException;
+import org.apache.paimon.utils.TimeUtils;
+
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+import org.apache.flink.types.Row;
+
+import javax.annotation.Nullable;
+
+import java.time.Duration;
+import java.util.Set;
+
+/** The procedure supports creating tags from snapshots commit-time. */
+public class CreateTagFromTimestampProcedure extends ProcedureBase {
+
+ public static final String IDENTIFIER = "create_tag_from_timestamp";
+
+ @ProcedureHint(
+ argument = {
+ @ArgumentHint(name = "table", type = @DataTypeHint("STRING")),
+ @ArgumentHint(name = "tag", type = @DataTypeHint(value = "STRING")),
+ @ArgumentHint(name = "timestamp", type = @DataTypeHint("bigint")),
+ @ArgumentHint(
+ name = "time_retained",
+ type = @DataTypeHint("STRING"),
+ isOptional = true),
+ })
+ @DataTypeHint("ROW< tagName STRING, snapshot BIGINT, `commit_time` BIGINT, `watermark` STRING>")
+ public Row[] call(
+ ProcedureContext procedureContext,
+ String tableId,
+ String tagName,
+ Long timestamp,
+ @Nullable String timeRetained)
+ throws Catalog.TableNotExistException {
+ FileStoreTable fileStoreTable = (FileStoreTable) table(tableId);
+ SnapshotManager snapshotManager = fileStoreTable.snapshotManager();
+
+ Snapshot snapshot = snapshotManager.laterOrEqualTimeMills(timestamp);
+
+ Set<Snapshot> sortedTagsSnapshots = fileStoreTable.tagManager().tags().keySet();
+
+ for (Snapshot tagSnapshot : sortedTagsSnapshots) {
+ if (timestamp <= tagSnapshot.timeMillis()) {
+ if (snapshot == null || tagSnapshot.timeMillis() < snapshot.timeMillis()) {
+ snapshot = tagSnapshot;
+ }
+ break;
+ }
+ }
+
+ SnapshotNotExistException.checkNotNull(
+ snapshot,
+ String.format(
+ "Could not find any snapshot whose commit-time later than %s.", timestamp));
+
+ fileStoreTable.createTag(tagName, snapshot.id(), toDuration(timeRetained));
+
+ return new Row[] {
+ Row.of(
+ tagName,
+ snapshot.id(),
+ snapshot.timeMillis(),
+ String.valueOf(snapshot.watermark()))
+ };
+ }
+
+ @Nullable
+ private static Duration toDuration(@Nullable String s) {
+ if (s == null) {
+ return null;
+ }
+
+ return TimeUtils.parseDuration(s);
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
index 22a0b564e..e689d71ab 100644
--- a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
+++ b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -42,6 +42,7 @@ org.apache.paimon.flink.procedure.CompactDatabaseProcedure
org.apache.paimon.flink.procedure.CompactProcedure
org.apache.paimon.flink.procedure.RewriteFileIndexProcedure
org.apache.paimon.flink.procedure.CreateTagProcedure
+org.apache.paimon.flink.procedure.CreateTagFromTimestampProcedure
org.apache.paimon.flink.procedure.DeleteTagProcedure
org.apache.paimon.flink.procedure.CreateBranchProcedure
org.apache.paimon.flink.procedure.DeleteBranchProcedure
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CreateTagFromTimestampProcedureITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CreateTagFromTimestampProcedureITCase.java
new file mode 100644
index 000000000..9e7aeb5de
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CreateTagFromTimestampProcedureITCase.java
@@ -0,0 +1,167 @@
+/*
+ * 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.flink.procedure;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotNotExistException;
+
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatException;
+
+/** IT Case for {@link CreateTagFromTimestampProcedure}. */
+public class CreateTagFromTimestampProcedureITCase extends CatalogITCaseBase {
+
+ @Test
+ public void testCreateTagsFromSnapshotsCommitTime() throws Exception {
+ sql(
+ "CREATE TABLE T ("
+ + " k STRING,"
+ + " dt STRING,"
+ + " PRIMARY KEY (k, dt) NOT ENFORCED"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'bucket' = '1'"
+ + ")");
+
+ for (int i = 1; i <= 4; i++) {
+ sql("insert into T values('%s', '2024-01-01')", i);
+ Thread.sleep(100L);
+ }
+
+ FileStoreTable table = paimonTable("T");
+ long earliestCommitTime = table.snapshotManager().earliestSnapshot().timeMillis();
+ long commitTime3 = table.snapshotManager().snapshot(3).timeMillis();
+ long commitTime4 = table.snapshotManager().snapshot(4).timeMillis();
+
+ // create tag from timestamp that earlier than the earliest snapshot commit time.
+ assertThat(
+ sql(
+ "CALL sys.create_tag_from_timestamp("
+ + "`table` => 'default.T',"
+ + "`tag` => 'tag1',"
+ + "`timestamp` => %s)",
+ earliestCommitTime - 1)
+ .stream()
+ .map(Row::toString))
+ .containsExactlyInAnyOrder(
+ String.format("+I[tag1, 1, %s, %s]", earliestCommitTime, Long.MIN_VALUE));
+
+ // create tag from timestamp that equals to snapshot-3 commit time.
+ assertThat(
+ sql(
+ "CALL sys.create_tag_from_timestamp("
+ + "`table` => 'default.T',"
+ + "`tag` => 'tag2',"
+ + "`timestamp` => %s)",
+ commitTime3)
+ .stream()
+ .map(Row::toString))
+ .containsExactlyInAnyOrder(
+ String.format("+I[tag2, 3, %s, %s]", commitTime3, Long.MIN_VALUE));
+
+ // create tag from timestamp that later than snapshot-3 commit time.
+ assertThat(
+ sql(
+ "CALL sys.create_tag_from_timestamp("
+ + "`table` => 'default.T',"
+ + "`tag` => 'tag3',"
+ + "`timestamp` => %s)",
+ commitTime3 + 1)
+ .stream()
+ .map(Row::toString))
+ .containsExactlyInAnyOrder(
+ String.format("+I[tag3, 4, %s, %s]", commitTime4, Long.MIN_VALUE));
+
+ // create tag from timestamp that later than the latest snapshot commit time and throw
+ // SnapshotNotExistException.
+ assertThatException()
+ .isThrownBy(
+ () ->
+ sql(
+ "CALL sys.create_tag_from_timestamp("
+ + "`table` => 'default.T',"
+ + "`tag` => 'tag4',"
+ + "`timestamp` => %s)",
+ Long.MAX_VALUE))
+ .withRootCauseInstanceOf(SnapshotNotExistException.class)
+ .withMessageContaining(
+ "Could not find any snapshot whose commit-time later than %s.",
+ Long.MAX_VALUE);
+ }
+
+ @Test
+ public void testCreateTagsFromTagsCommitTime() throws Exception {
+ sql(
+ "CREATE TABLE T ("
+ + " k STRING,"
+ + " dt STRING,"
+ + " PRIMARY KEY (k, dt) NOT ENFORCED"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'bucket' = '1'"
+ + ")");
+
+ sql("insert into T values('1', '2024-01-01')");
+ Thread.sleep(100L);
+
+ sql("CALL sys.create_tag('default.T', 'tag1', 1)");
+
+ // make snapshot-1 expire.
+ sql(
+ "insert into T/*+ OPTIONS("
+ + " 'snapshot.num-retained.max' = '1',"
+ + " 'snapshot.num-retained.min' = '1') */"
+ + " values('2', '2024-01-01')");
+
+ FileStoreTable table = paimonTable("T");
+ long earliestCommitTime = table.snapshotManager().earliestSnapshot().timeMillis();
+ long tagSnapshotCommitTime = table.tagManager().taggedSnapshot("tag1").timeMillis();
+
+ assertThat(tagSnapshotCommitTime < earliestCommitTime).isTrue();
+
+ // create tag from timestamp that earlier than the expired snapshot 1.
+ assertThat(
+ sql(
+ "CALL sys.create_tag_from_timestamp("
+ + "`table` => 'default.T',"
+ + "`tag` => 'tag2',"
+ + "`timestamp` => %s)",
+ tagSnapshotCommitTime - 1)
+ .stream()
+ .map(Row::toString))
+ .containsExactlyInAnyOrder(
+ String.format(
+ "+I[tag2, 1, %s, %s]", tagSnapshotCommitTime, Long.MIN_VALUE));
+
+ // create tag from timestamp that later than the expired snapshot 1.
+ assertThat(
+ sql(
+ "CALL sys.create_tag_from_timestamp("
+ + "`table` => 'default.T',"
+ + "`tag` => 'tag3',"
+ + "`timestamp` => %s)",
+ earliestCommitTime - 1)
+ .stream()
+ .map(Row::toString))
+ .containsExactlyInAnyOrder(
+ String.format("+I[tag3, 2, %s, %s]", earliestCommitTime, Long.MIN_VALUE));
+ }
+}
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
index f143cf7b6..0a9076247 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
@@ -20,6 +20,7 @@ package org.apache.paimon.spark;
import org.apache.paimon.spark.procedure.CompactProcedure;
import org.apache.paimon.spark.procedure.CreateBranchProcedure;
+import org.apache.paimon.spark.procedure.CreateTagFromTimestampProcedure;
import org.apache.paimon.spark.procedure.CreateTagProcedure;
import org.apache.paimon.spark.procedure.DeleteBranchProcedure;
import org.apache.paimon.spark.procedure.DeleteTagProcedure;
@@ -59,6 +60,8 @@ public class SparkProcedures {
ImmutableMap.builder();
procedureBuilders.put("rollback", RollbackProcedure::builder);
procedureBuilders.put("create_tag", CreateTagProcedure::builder);
+ procedureBuilders.put(
+ "create_tag_from_timestamp", CreateTagFromTimestampProcedure::builder);
procedureBuilders.put("delete_tag", DeleteTagProcedure::builder);
procedureBuilders.put("create_branch", CreateBranchProcedure::builder);
procedureBuilders.put("delete_branch", DeleteBranchProcedure::builder);
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreateTagFromTimestampProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreateTagFromTimestampProcedure.java
new file mode 100644
index 000000000..e62aacc21
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreateTagFromTimestampProcedure.java
@@ -0,0 +1,136 @@
+/*
+ * 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.spark.procedure;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.SnapshotNotExistException;
+import org.apache.paimon.utils.TimeUtils;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+
+import java.time.Duration;
+import java.util.Set;
+
+import static org.apache.spark.sql.types.DataTypes.LongType;
+import static org.apache.spark.sql.types.DataTypes.StringType;
+
+/** The procedure supports creating tags from snapshots commit-time or watermark. */
+public class CreateTagFromTimestampProcedure extends BaseProcedure {
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("table", StringType),
+ ProcedureParameter.required("tag", StringType),
+ ProcedureParameter.optional("timestamp", LongType),
+ ProcedureParameter.optional("time_retained", StringType)
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("tagName", DataTypes.StringType, true, Metadata.empty()),
+ new StructField("snapshot", DataTypes.LongType, true, Metadata.empty()),
+ new StructField("commit_time", DataTypes.LongType, true, Metadata.empty()),
+ new StructField("watermark", DataTypes.StringType, true, Metadata.empty())
+ });
+
+ protected CreateTagFromTimestampProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ Identifier tableIdent = toIdentifier(args.getString(0), PARAMETERS[0].name());
+ String tag = args.getString(1);
+ Long timestamp = args.getLong(2);
+ Duration timeRetained =
+ args.isNullAt(3) ? null : TimeUtils.parseDuration(args.getString(3));
+
+ return modifyPaimonTable(
+ tableIdent,
+ table -> {
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ SnapshotManager snapshotManager = fileStoreTable.snapshotManager();
+ Snapshot snapshot = snapshotManager.laterOrEqualTimeMills(timestamp);
+
+ Set<Snapshot> sortedTagsSnapshots = fileStoreTable.tagManager().tags().keySet();
+
+ // compare to tagsSnapshot.
+ for (Snapshot tagSnapshot : sortedTagsSnapshots) {
+ if (timestamp <= tagSnapshot.timeMillis()) {
+ if (snapshot == null
+ || tagSnapshot.timeMillis() < snapshot.timeMillis()) {
+ snapshot = tagSnapshot;
+ }
+ break;
+ }
+ }
+
+ SnapshotNotExistException.checkNotNull(
+ snapshot,
+ String.format(
+ "Could not find any snapshot whose commit-time later than %s.",
+ timestamp));
+
+ fileStoreTable.createTag(tag, snapshot.id(), timeRetained);
+
+ InternalRow outputRow =
+ newInternalRow(
+ UTF8String.fromString(tag),
+ snapshot.id(),
+ snapshot.timeMillis(),
+ UTF8String.fromString(String.valueOf(snapshot.watermark())));
+
+ return new InternalRow[] {outputRow};
+ });
+ }
+
+ public static ProcedureBuilder builder() {
+ return new BaseProcedure.Builder<CreateTagFromTimestampProcedure>() {
+ @Override
+ public CreateTagFromTimestampProcedure doBuild() {
+ return new CreateTagFromTimestampProcedure(tableCatalog());
+ }
+ };
+ }
+
+ @Override
+ public String description() {
+ return "CreateTagFromTimestampProcedure";
+ }
+}
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CreateTagFromTimestampProcedureTest.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CreateTagFromTimestampProcedureTest.scala
new file mode 100644
index 000000000..1da05843d
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CreateTagFromTimestampProcedureTest.scala
@@ -0,0 +1,179 @@
+/*
+ * 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.spark.procedure
+
+import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.utils.SnapshotNotExistException
+
+import org.apache.spark.sql.{Dataset, Row}
+import org.apache.spark.sql.execution.streaming.MemoryStream
+import org.apache.spark.sql.streaming.StreamTest
+
+class CreateTagFromTimestampProcedureTest extends PaimonSparkTestBase with StreamTest {
+
+ import testImplicits._
+
+ test("Paimon Procedure: Create tags from snapshots commit-time ") {
+ failAfter(streamingTimeout) {
+ withTempDir {
+ checkpointDir =>
+ spark.sql(s"""
+ |CREATE TABLE T (a INT, b STRING)
+ |TBLPROPERTIES ('primary-key'='a', 'bucket'='3')
+ |""".stripMargin)
+ val location = loadTable("T").location().toString
+
+ val inputData = MemoryStream[(Int, String)]
+ val stream = inputData
+ .toDS()
+ .toDF("a", "b")
+ .writeStream
+ .option("checkpointLocation", checkpointDir.getCanonicalPath)
+ .foreachBatch {
+ (batch: Dataset[Row], _: Long) =>
+ batch.write.format("paimon").mode("append").save(location)
+ }
+ .start()
+
+ try {
+
+ for (i <- 1 to 4) {
+ inputData.addData((i, "a"))
+ stream.processAllAvailable()
+ Thread.sleep(500L)
+ }
+
+ val table = loadTable("T")
+ val earliestCommitTime = table.snapshotManager.earliestSnapshot.timeMillis
+ val commitTime3 = table.snapshotManager.snapshot(3).timeMillis
+ val commitTime4 = table.snapshotManager.snapshot(4).timeMillis
+
+ // create tag from timestamp that earlier than the earliest snapshot commit time.
+ checkAnswer(
+ spark.sql(s"""CALL paimon.sys.create_tag_from_timestamp(
+ |table => 'test.T',
+ | tag => 'test_tag',
+ | timestamp => ${earliestCommitTime - 1})""".stripMargin),
+ Row("test_tag", 1, earliestCommitTime, "null") :: Nil
+ )
+
+ // create tag from timestamp that equals to snapshot-3 commit time.
+ checkAnswer(
+ spark.sql(s"""CALL paimon.sys.create_tag_from_timestamp(
+ |table => 'test.T',
+ | tag => 'test_tag2',
+ | timestamp => $commitTime3)""".stripMargin),
+ Row("test_tag2", 3, commitTime3, "null") :: Nil
+ )
+
+ // create tag from timestamp that later than snapshot-3 commit time.
+ checkAnswer(
+ spark.sql(s"""CALL paimon.sys.create_tag_from_timestamp(
+ |table => 'test.T',
+ |tag => 'test_tag3',
+ |timestamp => ${commitTime3 + 1})""".stripMargin),
+ Row("test_tag3", 4, commitTime4, "null") :: Nil
+ )
+
+ // create tag from timestamp that later than the latest snapshot commit time and throw SnapshotNotExistException.
+ assertThrows[SnapshotNotExistException] {
+ spark.sql(s"""CALL paimon.sys.create_tag_from_timestamp(
+ |table => 'test.T',
+ |tag => 'test_tag3',
+ |timestamp => ${Long.MaxValue})""".stripMargin)
+ }
+
+ } finally {
+ stream.stop()
+ }
+ }
+ }
+ }
+
+ test("Paimon Procedure: Create tags from tags commit-time") {
+ failAfter(streamingTimeout) {
+ withTempDir {
+ checkpointDir =>
+ spark.sql(s"""
+ |CREATE TABLE T (a INT, b STRING)
+ |TBLPROPERTIES ('primary-key'='a', 'bucket'='3')
+ |""".stripMargin)
+ val location = loadTable("T").location().toString
+
+ val inputData = MemoryStream[(Int, String)]
+ val stream = inputData
+ .toDS()
+ .toDF("a", "b")
+ .writeStream
+ .option("checkpointLocation", checkpointDir.getCanonicalPath)
+ .foreachBatch {
+ (batch: Dataset[Row], _: Long) =>
+ batch.write.format("paimon").mode("append").save(location)
+ }
+ .start()
+
+ try {
+ for (i <- 1 to 2) {
+ inputData.addData((i, "a"))
+ stream.processAllAvailable()
+ Thread.sleep(500L)
+ }
+
+ checkAnswer(
+ spark.sql(
+ "CALL paimon.sys.create_tag(" +
+ "table => 'test.T', tag => 'test_tag', snapshot => 1)"),
+ Row(true) :: Nil)
+
+ val table = loadTable("T")
+ val latestCommitTime = table.snapshotManager.latestSnapshot().timeMillis
+ val tagsCommitTime = table.tagManager().taggedSnapshot("test_tag").timeMillis
+ assert(latestCommitTime > tagsCommitTime)
+
+ // make snapshot 1 expire.
+ checkAnswer(
+ spark.sql("CALL paimon.sys.expire_snapshots(table => 'test.T', retain_max => 1)"),
+ Row(1) :: Nil)
+
+ // create tag from timestamp that earlier than the expired snapshot 1.
+ checkAnswer(
+ spark.sql(s"""CALL paimon.sys.create_tag_from_timestamp(
+ |table => 'test.T',
+ | tag => 'test_tag1',
+ | timestamp => ${tagsCommitTime - 1})""".stripMargin),
+ Row("test_tag1", 1, tagsCommitTime, "null") :: Nil
+ )
+
+ // create tag from timestamp that later than the expired snapshot 1.
+ checkAnswer(
+ spark.sql(s"""CALL paimon.sys.create_tag_from_timestamp(
+ |table => 'test.T',
+ |tag => 'test_tag2',
+ |timestamp => ${tagsCommitTime + 1})""".stripMargin),
+ Row("test_tag2", 2, latestCommitTime, "null") :: Nil
+ )
+
+ } finally {
+ stream.stop()
+ }
+ }
+ }
+ }
+
+}