This API accumulates batches of file additions and deletions by order, produces a new {@link + * Snapshot} of the changes where each batch is added to a new data sequence number, and commits + * that snapshot as the current. + * + *
When committing, these changes will be applied to the latest table snapshot. Commit conflicts
+ * will be resolved by applying the changes to the new latest snapshot and reattempting the commit.
+ * If any of the deleted files are no longer in the latest snapshot when reattempting, the commit
+ * will throw a {@link ValidationException}.
+ */
+public interface StreamingUpdate extends SnapshotUpdate This rewrite operation may change the size or layout of the data files. When applicable, it
+ * is also recommended to discard already deleted records while rewriting data files. However, the
+ * set of live data records must never change.
+ *
+ * @param dataFile a rewritten data file
+ * @return this for method chaining
+ */
+ default StreamingUpdate deleteFile(DataFile dataFile) {
+ throw new UnsupportedOperationException(
+ this.getClass().getName() + " does not implement deleteFile");
+ }
+
+ /**
+ * Remove a delete file from the table state.
+ *
+ * This rewrite operation may change the size or layout of the delete files. When applicable,
+ * it is also recommended to discard delete records for files that are no longer part of the table
+ * state. However, the set of applicable delete records must never change.
+ *
+ * @param deleteFile a rewritten delete file
+ * @return this for method chaining
+ */
+ default StreamingUpdate deleteFile(DeleteFile deleteFile) {
+ throw new UnsupportedOperationException(
+ this.getClass().getName() + " does not implement deleteFile");
+ }
+
+ /**
+ * Add a new data file to a specific. All files in this batch will receive the same data sequence
+ * number.
+ *
+ * This rewrite operation may change the size or layout of the data files. When applicable, it
+ * is also recommended to discard already deleted records while rewriting data files. However, the
+ * set of live data records must never change.
+ *
+ * @param dataFile a new data file
+ * @param batchOrdinal The batch ordinal to associate with this data file
+ * @return this for method chaining
+ */
+ default StreamingUpdate addFile(DataFile dataFile, int batchOrdinal) {
+ throw new UnsupportedOperationException(
+ this.getClass().getName() + " does not implement addFile");
+ }
+
+ /**
+ * Add a new delete file to a specific batch. All files in this batch will receive the same data
+ * sequence number.
+ *
+ * This rewrite operation may change the size or layout of the delete files. When applicable,
+ * it is also recommended to discard delete records for files that are no longer part of the table
+ * state. However, the set of applicable delete records must never change.
+ *
+ * @param deleteFile a new delete file
+ * @param batchOrdinal The batch ordinal to associate with this data file
+ * @return this for method chaining
+ */
+ default StreamingUpdate addFile(DeleteFile deleteFile, int batchOrdinal) {
+ throw new UnsupportedOperationException(
+ this.getClass().getName() + " does not implement addFile");
+ }
+}
diff --git a/api/src/main/java/org/apache/iceberg/Table.java b/api/src/main/java/org/apache/iceberg/Table.java
index 5ab1def51ca0..7683f1d59d84 100644
--- a/api/src/main/java/org/apache/iceberg/Table.java
+++ b/api/src/main/java/org/apache/iceberg/Table.java
@@ -233,6 +233,17 @@ default AppendFiles newFastAppend() {
*/
RewriteFiles newRewrite();
+ /**
+ * Create a new {@link StreamingUpdate streaming update API} to append sequential upserts to the
+ * table.
+ *
+ * @return a new {@link StreamingUpdate}
+ */
+ default StreamingUpdate newStreamingUpdate() {
+ throw new UnsupportedOperationException(
+ this.getClass().getName() + " doesn't implement newStreamingUpdate()");
+ }
+
/**
* Create a new {@link RewriteManifests rewrite manifests API} to replace manifests for this table
* and commit.
diff --git a/api/src/main/java/org/apache/iceberg/Transaction.java b/api/src/main/java/org/apache/iceberg/Transaction.java
index aeec1f589d06..85d0267bd379 100644
--- a/api/src/main/java/org/apache/iceberg/Transaction.java
+++ b/api/src/main/java/org/apache/iceberg/Transaction.java
@@ -95,6 +95,17 @@ default AppendFiles newFastAppend() {
*/
RewriteFiles newRewrite();
+ /**
+ * Create a new {@link StreamingUpdate streaming update API} to append sequential upserts to the
+ * table.
+ *
+ * @return a new {@link StreamingUpdate}
+ */
+ default StreamingUpdate newStreamingUpdate() {
+ throw new UnsupportedOperationException(
+ this.getClass().getName() + " doesn't implement newStreamingUpdate()");
+ }
+
/**
* Create a new {@link RewriteManifests rewrite manifests API} to replace manifests for this
* table.
diff --git a/api/src/test/java/org/apache/iceberg/TestHelpers.java b/api/src/test/java/org/apache/iceberg/TestHelpers.java
index cefd765a91b5..fcb528caba96 100644
--- a/api/src/test/java/org/apache/iceberg/TestHelpers.java
+++ b/api/src/test/java/org/apache/iceberg/TestHelpers.java
@@ -360,6 +360,52 @@ public int hashCode() {
}
}
+ // similar to Row but has its own hashCode() and equals() implementations
+ // it is useful for testing custom collections that rely on wrappers
+ public static class CustomRow implements StructLike {
+ public static CustomRow of(Object... values) {
+ return new CustomRow(values);
+ }
+
+ private final Object[] values;
+
+ private CustomRow(Object... values) {
+ this.values = values;
+ }
+
+ @Override
+ public int size() {
+ return values.length;
+ }
+
+ @Override
+ public This implementation internally stores provided partition tuples in {@link StructLikeMap} for
+ * consistent hashing and equals behavior. This ensures that objects of different types that
+ * represent the same structs are treated as equal keys in the map.
+ *
+ * Note: This map is not designed for concurrent modification by multiple threads. However, it
+ * supports safe concurrent reads, assuming there are no concurrent writes.
+ *
+ * Note: This map does not support null pairs but supports null as partition tuples.
+ *
+ * @param