This is an automated email from the ASF dual-hosted git repository.
JingsongLi 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 52b8e8adac [spark] Support create_tag_from_watermark procedure (#10420)
52b8e8adac is described below
commit 52b8e8adac8972a4be1bc5ecb7f08b929f57d4d8
Author: Wenchao Wu <[email protected]>
AuthorDate: Tue Oct 13 13:31:27 2026 +0800
[spark] Support create_tag_from_watermark procedure (#10420)
---
docs/docs/spark/procedures.md | 1 +
docs/docs/spark/procedures/versions.md | 20 +++
.../org/apache/paimon/utils/SnapshotManager.java | 7 +-
.../apache/paimon/utils/SnapshotManagerTest.java | 54 ++++++
.../org/apache/paimon/spark/SparkProcedures.java | 3 +
.../procedure/CreateTagFromWatermarkProcedure.java | 139 ++++++++++++++++
.../CreateTagFromWatermarkProcedureTest.scala | 184 +++++++++++++++++++++
7 files changed, 406 insertions(+), 2 deletions(-)
diff --git a/docs/docs/spark/procedures.md b/docs/docs/spark/procedures.md
index 70f329b3a4..8e209a484f 100644
--- a/docs/docs/spark/procedures.md
+++ b/docs/docs/spark/procedures.md
@@ -79,6 +79,7 @@ Choose a group, then use the page contents to jump to a
procedure:
[`create_tag`](./procedures/versions#create_tag),
[`create_tag_from_timestamp`](./procedures/versions#create_tag_from_timestamp),
+[`create_tag_from_watermark`](./procedures/versions#create_tag_from_watermark),
[`replace_tag`](./procedures/versions#replace_tag),
[`rename_tag`](./procedures/versions#rename_tag),
[`delete_tag`](./procedures/versions#delete_tag),
diff --git a/docs/docs/spark/procedures/versions.md
b/docs/docs/spark/procedures/versions.md
index f4d4a76f08..7b007c527b 100644
--- a/docs/docs/spark/procedures/versions.md
+++ b/docs/docs/spark/procedures/versions.md
@@ -69,6 +69,26 @@ CALL sys.create_tag_from_timestamp(
);
```
+## create_tag_from_watermark
+
+Create a tag from the first retained snapshot (including tagged snapshots)
whose watermark is at or after the given value.
+
+**Arguments**
+
+- `table` (`STRING`, required): the target table identifier.
+- `tag` (`STRING`, required): name of the new tag.
+- `watermark` (`BIGINT`, required): find the first snapshot whose watermark is
later than or equal to this value.
+- `time_retained` (`STRING`, optional): The maximum time retained for newly
created tags.
+
+```sql
+CALL sys.create_tag_from_watermark(
+ `table` => 'default.T',
+ `tag` => 'my_tag',
+ `watermark` => 1724404318750,
+ time_retained => '1 d'
+);
+```
+
## replace_tag
Replace an existing tag with new tag info.
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 5f90d2db91..df627955e5 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
@@ -558,8 +558,11 @@ public class SnapshotManager implements Serializable {
} else if (commitWatermark < watermark) {
earliest = mid + 1; // Search in the right half
} else {
- finalSnapshot = snapshot; // Found the exact match
- break;
+ // Lower bound: an exact match is a candidate, but an
earlier snapshot may
+ // carry the same watermark. Watermarks repeat because the
commit watermark
+ // takes the maximum, so stopping here would return the
middle of such a run.
+ finalSnapshot = snapshot;
+ latest = pos - 1; // Keep searching the left half
}
}
}
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 fb09951952..9bd9444bd5 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
@@ -855,6 +855,60 @@ public class SnapshotManagerTest {
assertThat(snapshotManager.laterOrEqualWatermark(250)).isNull();
}
+ @Test
+ public void testLaterOrEqualWatermarkWithRepeatedWatermarks() throws
IOException {
+ FileIO localFileIO = LocalFileIO.create();
+ SnapshotManager snapshotManager =
+ newSnapshotManager(localFileIO, new Path(tempDir.toString()));
+ // a plateau of identical watermarks: snapshots 1, 2, 3 all carry
1000. Repeated
+ // watermarks are normal because the commit watermark takes the
maximum.
+ Map<Long, Long> watermarks = watermarkMap(0, 500, 1, 1000, 2, 1000, 3,
1000, 4, 2000);
+ for (long i = 0; i < 5; i++) {
+ Snapshot snapshot = createSnapshotWithMillis(i, 1000 + i * 1000,
watermarks.get(i));
+ localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i),
snapshot.toJson());
+ }
+
+ // lower bound: the first snapshot of the plateau, not the
binary-search midpoint
+
assertThat(snapshotManager.laterOrEqualWatermark(1000).id()).isEqualTo(1);
+ // a value inside the gap still resolves to the start of the next
plateau
+
assertThat(snapshotManager.laterOrEqualWatermark(501).id()).isEqualTo(1);
+ // non-repeated watermarks are unaffected
+
assertThat(snapshotManager.laterOrEqualWatermark(2000).id()).isEqualTo(4);
+
assertThat(snapshotManager.laterOrEqualWatermark(500).id()).isEqualTo(0);
+ }
+
+ @Test
+ public void testLaterOrEqualWatermarkWithPlateauAtTheEarliestSnapshot()
throws IOException {
+ FileIO localFileIO = LocalFileIO.create();
+ SnapshotManager snapshotManager =
+ newSnapshotManager(localFileIO, new Path(tempDir.toString()));
+ // the plateau starts at the earliest snapshot, so the left walk must
stop there
+ Map<Long, Long> watermarks = watermarkMap(0, 1000, 1, 1000, 2, 1000,
3, 2000);
+ for (long i = 0; i < 4; i++) {
+ Snapshot snapshot = createSnapshotWithMillis(i, 1000 + i * 1000,
watermarks.get(i));
+ localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i),
snapshot.toJson());
+ }
+
+
assertThat(snapshotManager.laterOrEqualWatermark(1000).id()).isEqualTo(0);
+ }
+
+ @Test
+ public void
testLaterOrEqualWatermarkWithRepeatedWatermarksAndNullWatermarks()
+ throws IOException {
+ FileIO localFileIO = LocalFileIO.create();
+ SnapshotManager snapshotManager =
+ newSnapshotManager(localFileIO, new Path(tempDir.toString()));
+ // snapshots 2 and 4 carry no watermark and sit inside a plateau of
1000s
+ Map<Long, Long> watermarks = watermarkMap(0, 500, 1, 1000, 3, 1000, 5,
1000, 6, 2000);
+ for (long i = 0; i < 7; i++) {
+ Snapshot snapshot = createSnapshotWithMillis(i, 1000 + i * 1000,
watermarks.get(i));
+ localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i),
snapshot.toJson());
+ }
+
+ // null-watermark snapshots must not shadow the real start of the
plateau
+
assertThat(snapshotManager.laterOrEqualWatermark(1000).id()).isEqualTo(1);
+ }
+
@Test
public void testEarlierOrEqualWatermarkBelowMinimum() throws IOException {
FileIO localFileIO = LocalFileIO.create();
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 61d318a5d9..3ecd59c8bf 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
@@ -31,6 +31,7 @@ import
org.apache.paimon.spark.procedure.CreateFunctionProcedure;
import org.apache.paimon.spark.procedure.CreateGlobalIndexProcedure;
import org.apache.paimon.spark.procedure.CreatePolicyProcedure;
import org.apache.paimon.spark.procedure.CreateTagFromTimestampProcedure;
+import org.apache.paimon.spark.procedure.CreateTagFromWatermarkProcedure;
import org.apache.paimon.spark.procedure.CreateTagProcedure;
import org.apache.paimon.spark.procedure.DeleteBranchProcedure;
import org.apache.paimon.spark.procedure.DeleteTagProcedure;
@@ -110,6 +111,8 @@ public class SparkProcedures {
procedureBuilders.put("rename_tag", RenameTagProcedure::builder);
procedureBuilders.put(
"create_tag_from_timestamp",
CreateTagFromTimestampProcedure::builder);
+ procedureBuilders.put(
+ "create_tag_from_watermark",
CreateTagFromWatermarkProcedure::builder);
procedureBuilders.put("delete_tag", DeleteTagProcedure::builder);
procedureBuilders.put("expire_tags", ExpireTagsProcedure::builder);
procedureBuilders.put("create_branch", CreateBranchProcedure::builder);
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreateTagFromWatermarkProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreateTagFromWatermarkProcedure.java
new file mode 100644
index 0000000000..214787f20e
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreateTagFromWatermarkProcedure.java
@@ -0,0 +1,139 @@
+/*
+ * 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 snapshot watermarks. */
+public class CreateTagFromWatermarkProcedure extends BaseProcedure {
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("table", StringType),
+ ProcedureParameter.required("tag", StringType),
+ ProcedureParameter.required("watermark", 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())
+ });
+
+ private CreateTagFromWatermarkProcedure(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 watermark = 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.laterOrEqualWatermark(watermark);
+
+ Set<Snapshot> sortedTagsSnapshots =
fileStoreTable.tagManager().tags().keySet();
+ for (Snapshot tagSnapshot : sortedTagsSnapshots) {
+ if (tagSnapshot.watermark() != null
+ && watermark <= tagSnapshot.watermark()) {
+ // Equal watermarks keep the earlier snapshot.
Comparing with <=
+ // would replace an earlier live snapshot with a
later tag.
+ if (snapshot == null
+ || snapshot.watermark() == null
+ || tagSnapshot.watermark() <
snapshot.watermark()
+ ||
(tagSnapshot.watermark().equals(snapshot.watermark())
+ && tagSnapshot.id() <
snapshot.id())) {
+ snapshot = tagSnapshot;
+ }
+ break;
+ }
+ }
+
+ SnapshotNotExistException.checkNotNull(
+ snapshot,
+ String.format(
+ "Could not find any snapshot whose
watermark later than %s.",
+ watermark));
+
+ 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<CreateTagFromWatermarkProcedure>() {
+ @Override
+ public CreateTagFromWatermarkProcedure doBuild() {
+ return new CreateTagFromWatermarkProcedure(tableCatalog());
+ }
+ };
+ }
+
+ @Override
+ public String description() {
+ return "CreateTagFromWatermarkProcedure";
+ }
+}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CreateTagFromWatermarkProcedureTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CreateTagFromWatermarkProcedureTest.scala
new file mode 100644
index 0000000000..90f01f93dd
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CreateTagFromWatermarkProcedureTest.scala
@@ -0,0 +1,184 @@
+/*
+ * 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.data.{BinaryString, GenericRow}
+import org.apache.paimon.manifest.ManifestCommittable
+import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.table.sink.TableCommitImpl
+import org.apache.paimon.utils.SnapshotNotExistException
+
+import org.apache.spark.sql.Row
+
+class CreateTagFromWatermarkProcedureTest extends PaimonSparkTestBase {
+
+ test("Paimon Procedure: create tags from snapshot watermarks") {
+ spark.sql("CREATE TABLE T (k INT, v STRING)")
+
+ writeWithWatermark("T", null, GenericRow.of(Integer.valueOf(1),
BinaryString.fromString("a")))
+
+ intercept[SnapshotNotExistException] {
+ spark.sql(
+ "CALL sys.create_tag_from_watermark(table => 'test.T', tag => 'tag1',
watermark => 1000)")
+ }
+
+ writeWithWatermark("T", 1000L, GenericRow.of(Integer.valueOf(2),
BinaryString.fromString("b")))
+ writeWithWatermark("T", 2000L, GenericRow.of(Integer.valueOf(3),
BinaryString.fromString("c")))
+
+ val table = loadTable("T")
+ val snapshot2 = table.snapshotManager.snapshot(2)
+ val snapshot3 = table.snapshotManager.snapshot(3)
+
+ checkAnswer(
+ spark.sql(
+ s"CALL sys.create_tag_from_watermark(table => 'test.T', tag => 'tag2',
watermark => ${1000L - 1})"),
+ Row("tag2", 2L, snapshot2.timeMillis,
String.valueOf(snapshot2.watermark)) :: Nil
+ )
+
+ checkAnswer(
+ spark.sql(
+ s"CALL sys.create_tag_from_watermark(table => 'test.T', tag => 'tag3',
watermark => ${1000L + 1})"),
+ Row("tag3", 3L, snapshot3.timeMillis,
String.valueOf(snapshot3.watermark)) :: Nil
+ )
+
+ intercept[SnapshotNotExistException] {
+ spark.sql(
+ s"CALL sys.create_tag_from_watermark(table => 'test.T', tag => 'tag4',
watermark => ${2000L + 1})")
+ }
+ }
+
+ test("Paimon Procedure: create tags from tagged snapshot watermarks after
expire") {
+ spark.sql("CREATE TABLE T (k INT, v STRING)")
+
+ writeWithWatermark("T", 1000L, GenericRow.of(Integer.valueOf(1),
BinaryString.fromString("a")))
+ spark.sql("CALL sys.create_tag(table => 'test.T', tag => 'tag1', snapshot
=> 1)")
+ writeWithWatermark("T", 2000L, GenericRow.of(Integer.valueOf(2),
BinaryString.fromString("b")))
+
+ spark.sql("CALL sys.expire_snapshots(table => 'test.T', retain_max => 1,
retain_min => 1)")
+
+ val table = loadTable("T")
+ assert(!table.snapshotManager.snapshotExists(1))
+
+ val tagSnapshot = table.tagManager.getOrThrow("tag1")
+ val snapshot2 = table.snapshotManager.snapshot(2)
+
+ checkAnswer(
+ spark.sql(
+ s"CALL sys.create_tag_from_watermark(table => 'test.T', tag => 'tag2',
watermark => ${tagSnapshot.watermark - 1})"),
+ Row("tag2", 1L, tagSnapshot.timeMillis,
String.valueOf(tagSnapshot.watermark)) :: Nil
+ )
+
+ checkAnswer(
+ spark.sql(
+ s"CALL sys.create_tag_from_watermark(table => 'test.T', tag => 'tag3',
watermark => ${snapshot2.watermark - 1})"),
+ Row("tag3", 2L, snapshot2.timeMillis,
String.valueOf(snapshot2.watermark)) :: Nil
+ )
+ }
+
+ test("Paimon Procedure: prefer earlier tagged snapshot when watermarks are
equal") {
+ spark.sql("CREATE TABLE T (k INT, v STRING)")
+
+ writeWithWatermark("T", 1000L, GenericRow.of(Integer.valueOf(1),
BinaryString.fromString("a")))
+ spark.sql("CALL sys.create_tag(table => 'test.T', tag => 'keep1', snapshot
=> 1)")
+ writeWithWatermark("T", 1000L, GenericRow.of(Integer.valueOf(2),
BinaryString.fromString("b")))
+
+ spark.sql("CALL sys.expire_snapshots(table => 'test.T', retain_max => 1,
retain_min => 1)")
+
+ val table = loadTable("T")
+ assert(!table.snapshotManager.snapshotExists(1))
+ val tagged = table.tagManager.getOrThrow("keep1")
+
+ checkAnswer(
+ spark.sql(
+ "CALL sys.create_tag_from_watermark(table => 'test.T', tag =>
'tag_from_wm', watermark => 1000)"),
+ Row("tag_from_wm", 1L, tagged.timeMillis,
String.valueOf(tagged.watermark)) :: Nil
+ )
+ }
+
+ test("Paimon Procedure: select the first live snapshot in an equal-watermark
interval") {
+ spark.sql("CREATE TABLE T (k INT, v STRING)")
+
+ writeWithWatermark("T", 500L, GenericRow.of(Integer.valueOf(1),
BinaryString.fromString("a")))
+ writeWithWatermark("T", 1000L, GenericRow.of(Integer.valueOf(2),
BinaryString.fromString("b")))
+ writeWithWatermark("T", 1000L, GenericRow.of(Integer.valueOf(3),
BinaryString.fromString("c")))
+ writeWithWatermark("T", 1000L, GenericRow.of(Integer.valueOf(4),
BinaryString.fromString("d")))
+ writeWithWatermark("T", 2000L, GenericRow.of(Integer.valueOf(5),
BinaryString.fromString("e")))
+
+ val table = loadTable("T")
+ val snapshot2 = table.snapshotManager.snapshot(2)
+ val snapshot5 = table.snapshotManager.snapshot(5)
+
+ checkAnswer(
+ spark.sql(
+ "CALL sys.create_tag_from_watermark(table => 'test.T', tag =>
'plateau', watermark => 1000)"),
+ Row("plateau", 2L, snapshot2.timeMillis,
String.valueOf(snapshot2.watermark)) :: Nil
+ )
+
+ checkAnswer(
+ spark.sql(
+ "CALL sys.create_tag_from_watermark(table => 'test.T', tag => 'later',
watermark => 2000)"),
+ Row("later", 5L, snapshot5.timeMillis,
String.valueOf(snapshot5.watermark)) :: Nil
+ )
+ }
+
+ test("Paimon Procedure: keep earlier live snapshot when a later tag has the
same watermark") {
+ spark.sql("CREATE TABLE T (k INT, v STRING)")
+
+ writeWithWatermark("T", 1000L, GenericRow.of(Integer.valueOf(1),
BinaryString.fromString("a")))
+ writeWithWatermark("T", 1000L, GenericRow.of(Integer.valueOf(2),
BinaryString.fromString("b")))
+ spark.sql("CALL sys.create_tag(table => 'test.T', tag => 'keep2', snapshot
=> 2)")
+
+ val table = loadTable("T")
+ val snapshot1 = table.snapshotManager.snapshot(1)
+ assert(table.tagManager.getOrThrow("keep2").id == 2)
+
+ checkAnswer(
+ spark.sql(
+ "CALL sys.create_tag_from_watermark(table => 'test.T', tag =>
'tag_from_wm', watermark => 1000)"),
+ Row("tag_from_wm", 1L, snapshot1.timeMillis,
String.valueOf(snapshot1.watermark)) :: Nil
+ )
+ }
+
+ private var commitIdentifier = 0L
+
+ private def writeWithWatermark(
+ tableName: String,
+ watermark: java.lang.Long,
+ rows: GenericRow*): Unit = {
+ val table = loadTable(tableName)
+ val writeBuilder = table.newStreamWriteBuilder
+ val writer = writeBuilder.newWrite
+ val commit = writeBuilder.newCommit.asInstanceOf[TableCommitImpl]
+ try {
+ rows.foreach(writer.write)
+ commitIdentifier += 1
+ val committable = new ManifestCommittable(commitIdentifier, watermark)
+ val messages = writer.prepareCommit(true, commitIdentifier)
+ var i = 0
+ while (i < messages.size()) {
+ committable.addFileCommittable(messages.get(i))
+ i += 1
+ }
+ commit.commit(committable)
+ } finally {
+ writer.close()
+ commit.close()
+ }
+ }
+}
