This is an automated email from the ASF dual-hosted git repository.
amogh-jahagirdar pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new c534345602 Spark: Rename streaming merge-append write option and port
to 3.5/4.0 (#17403)
c534345602 is described below
commit c53434560298176454501b28d12895becfb80e10
Author: Amogh Jahagirdar <[email protected]>
AuthorDate: Wed Jul 29 07:14:39 2026 -0600
Spark: Rename streaming merge-append write option and port to 3.5/4.0
(#17403)
Renames the streaming merge-append write config to
`use-merge-append-for-streaming` in Spark 4.1, and ports the option to
the Spark 3.5 and 4.0 modules.
When enabled, streaming appends use a merge append instead of a fast
append so manifests are merged on commit rather than accumulating as
small manifests.
---
docs/docs/spark-configuration.md | 2 +-
.../org/apache/iceberg/spark/SparkWriteConf.java | 8 +++
.../apache/iceberg/spark/SparkWriteOptions.java | 4 ++
.../apache/iceberg/spark/source/SparkWrite.java | 4 +-
.../spark/source/TestStructuredStreaming.java | 66 ++++++++++++++++++++++
.../org/apache/iceberg/spark/SparkWriteConf.java | 8 +++
.../apache/iceberg/spark/SparkWriteOptions.java | 4 ++
.../apache/iceberg/spark/source/SparkWrite.java | 4 +-
.../spark/source/TestStructuredStreaming.java | 66 ++++++++++++++++++++++
.../org/apache/iceberg/spark/SparkWriteConf.java | 6 +-
.../apache/iceberg/spark/SparkWriteOptions.java | 4 +-
.../apache/iceberg/spark/source/SparkWrite.java | 6 +-
.../spark/source/TestStructuredStreaming.java | 2 +-
13 files changed, 172 insertions(+), 12 deletions(-)
diff --git a/docs/docs/spark-configuration.md b/docs/docs/spark-configuration.md
index 4d91a50d7d..746a0d3080 100644
--- a/docs/docs/spark-configuration.md
+++ b/docs/docs/spark-configuration.md
@@ -269,7 +269,7 @@ df.writeTo("catalog.db.table")
| delete-granularity | file | Override this table's delete granularity for
this write |
| shred-variants | false | Overrides this table's write.parquet.shred-variants
for this write |
| variant-inference-buffer-size | 100 | Overrides this table's
write.parquet.variant-inference-buffer-size for this write |
-| streaming-merge-append-enabled | false | Use a merge append instead of a
fast append for streaming appends, so manifests are merged on commit rather
than accumulating as small manifests |
+| use-merge-append-for-streaming | false | Use a merge append instead of a
fast append for streaming appends, so manifests are merged on commit rather
than accumulating as small manifests |
CommitMetadata provides an interface to add custom metadata to a snapshot
summary during a SQL execution, which can be beneficial for purposes such as
auditing or change tracking. If properties start with `snapshot-property.`,
then that prefix will be removed from each property. Here is an example:
diff --git
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java
index 537a14e4ab..2f98aec4e7 100644
---
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java
+++
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java
@@ -233,6 +233,14 @@ public class SparkWriteConf {
.parse();
}
+ public boolean useMergeAppendForStreaming() {
+ return confParser
+ .booleanConf()
+ .option(SparkWriteOptions.USE_MERGE_APPEND_FOR_STREAMING)
+ .defaultValue(SparkWriteOptions.USE_MERGE_APPEND_FOR_STREAMING_DEFAULT)
+ .parse();
+ }
+
public FileFormat deleteFileFormat() {
if (!(table instanceof BaseMetadataTable) &&
TableUtil.formatVersion(table) >= 3) {
return FileFormat.PUFFIN;
diff --git
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java
index 1be02feaf0..09bfd762b7 100644
---
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java
+++
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java
@@ -86,4 +86,8 @@ public class SparkWriteOptions {
// Overrides the delete granularity
public static final String DELETE_GRANULARITY = "delete-granularity";
+
+ // Uses the merge append instead of fast append for streaming appends
+ public static final String USE_MERGE_APPEND_FOR_STREAMING =
"use-merge-append-for-streaming";
+ public static final boolean USE_MERGE_APPEND_FOR_STREAMING_DEFAULT = false;
}
diff --git
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java
index aff8864b6d..18ad6b5a7a 100644
---
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java
+++
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java
@@ -101,6 +101,7 @@ abstract class SparkWrite implements Write,
RequiresDistributionAndOrdering {
private final StructType dsSchema;
private final Map<String, String> extraSnapshotMetadata;
private final boolean useFanoutWriter;
+ private final boolean useMergeAppendForStreaming;
private final SparkWriteRequirements writeRequirements;
private final Map<String, String> writeProperties;
@@ -129,6 +130,7 @@ abstract class SparkWrite implements Write,
RequiresDistributionAndOrdering {
this.dsSchema = dsSchema;
this.extraSnapshotMetadata = writeConf.extraSnapshotMetadata();
this.useFanoutWriter = writeConf.useFanoutWriter(writeRequirements);
+ this.useMergeAppendForStreaming = writeConf.useMergeAppendForStreaming();
this.writeRequirements = writeRequirements;
this.outputSpecId = writeConf.outputSpecId();
this.writeProperties = writeConf.writeProperties();
@@ -598,7 +600,7 @@ abstract class SparkWrite implements Write,
RequiresDistributionAndOrdering {
@Override
protected void doCommit(long epochId, WriterCommitMessage[] messages) {
- AppendFiles append = table.newFastAppend();
+ AppendFiles append = useMergeAppendForStreaming ? table.newAppend() :
table.newFastAppend();
int numFiles = 0;
for (DataFile file : files(messages)) {
append.appendFile(file);
diff --git
a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java
b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java
index ab76001053..4e243da54f 100644
---
a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java
+++
b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java
@@ -34,10 +34,12 @@ import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SortOrder;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
import org.apache.iceberg.hadoop.HadoopTables;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.spark.SparkWriteOptions;
import org.apache.iceberg.spark.TestBase;
import org.apache.iceberg.types.Types;
import org.apache.spark.sql.Dataset;
@@ -147,6 +149,70 @@ public class TestStructuredStreaming {
}
}
+ @Test
+ public void testStreamingWriteAppendModeWithMergeAppend() throws Exception {
+ File parent = temp.resolve("parquet").toFile();
+ File location = new File(parent, "test-table");
+ File checkpoint = new File(parent, "checkpoint");
+
+ HadoopTables tables = new HadoopTables(CONF);
+ PartitionSpec spec =
PartitionSpec.builderFor(SCHEMA).identity("data").build();
+ // set a low min-count-to-merge so the merge append visibly consolidates
manifests
+ Table table =
+ tables.create(
+ SCHEMA,
+ spec,
+ ImmutableMap.of(TableProperties.MANIFEST_MIN_MERGE_COUNT, "2"),
+ location.toString());
+
+ List<SimpleRecord> expected =
+ Lists.newArrayList(
+ new SimpleRecord(1, "1"),
+ new SimpleRecord(2, "2"),
+ new SimpleRecord(3, "3"),
+ new SimpleRecord(4, "4"));
+
+ MemoryStream<Integer> inputStream = newMemoryStream(1, spark.sqlContext(),
Encoders.INT());
+ DataStreamWriter<Row> streamWriter =
+ inputStream
+ .toDF()
+ .selectExpr("value AS id", "CAST (value AS STRING) AS data")
+ .writeStream()
+ .outputMode("append")
+ .format("iceberg")
+ .option("checkpointLocation", checkpoint.toString())
+ .option("path", location.toString())
+ .option(SparkWriteOptions.USE_MERGE_APPEND_FOR_STREAMING, "true");
+
+ try {
+ StreamingQuery query = streamWriter.start();
+ List<Integer> batch1 = Lists.newArrayList(1, 2);
+ send(batch1, inputStream);
+ query.processAllAvailable();
+ List<Integer> batch2 = Lists.newArrayList(3, 4);
+ send(batch2, inputStream);
+ query.processAllAvailable();
+ query.stop();
+
+ Dataset<Row> result =
spark.read().format("iceberg").load(location.toString());
+ List<SimpleRecord> actual =
+
result.orderBy("id").as(Encoders.bean(SimpleRecord.class)).collectAsList();
+
+ assertThat(actual).hasSameSizeAs(expected).isEqualTo(expected);
+ assertThat(table.snapshots()).as("Number of snapshots should
match").hasSize(2);
+
+ // the second streaming append uses the regular append, which merges the
two data manifests
+ // into a single manifest; a fast append would have left two separate
manifests
+ assertThat(table.currentSnapshot().dataManifests(table.io()))
+ .as("Merge append should consolidate data manifests")
+ .hasSize(1);
+ } finally {
+ for (StreamingQuery query : spark.streams().active()) {
+ query.stop();
+ }
+ }
+ }
+
@Test
public void testStreamingWriteCompleteMode() throws Exception {
File parent = temp.resolve("parquet").toFile();
diff --git
a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java
b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java
index 71a4a7878f..3a34a1eb83 100644
---
a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java
+++
b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java
@@ -237,6 +237,14 @@ public class SparkWriteConf {
.parse();
}
+ public boolean useMergeAppendForStreaming() {
+ return confParser
+ .booleanConf()
+ .option(SparkWriteOptions.USE_MERGE_APPEND_FOR_STREAMING)
+ .defaultValue(SparkWriteOptions.USE_MERGE_APPEND_FOR_STREAMING_DEFAULT)
+ .parse();
+ }
+
public FileFormat deleteFileFormat() {
if (!(table instanceof BaseMetadataTable) &&
TableUtil.formatVersion(table) >= 3) {
return FileFormat.PUFFIN;
diff --git
a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java
b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java
index 6c76b5c873..b7a5597aea 100644
---
a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java
+++
b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java
@@ -92,4 +92,8 @@ public class SparkWriteOptions {
// Controls the buffer size for variant schema inference during writes
public static final String VARIANT_INFERENCE_BUFFER_SIZE =
"variant-inference-buffer-size";
+
+ // Uses the merge append instead of fast append for streaming appends
+ public static final String USE_MERGE_APPEND_FOR_STREAMING =
"use-merge-append-for-streaming";
+ public static final boolean USE_MERGE_APPEND_FOR_STREAMING_DEFAULT = false;
}
diff --git
a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java
b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java
index c73a37ba34..7db1c60488 100644
---
a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java
+++
b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java
@@ -109,6 +109,7 @@ abstract class SparkWrite implements Write,
RequiresDistributionAndOrdering {
private final StructType dsSchema;
private final Map<String, String> extraSnapshotMetadata;
private final boolean useFanoutWriter;
+ private final boolean useMergeAppendForStreaming;
private final SparkWriteRequirements writeRequirements;
private final Map<String, String> writeProperties;
@@ -138,6 +139,7 @@ abstract class SparkWrite implements Write,
RequiresDistributionAndOrdering {
this.dsSchema = dsSchema;
this.extraSnapshotMetadata = writeConf.extraSnapshotMetadata();
this.useFanoutWriter = writeConf.useFanoutWriter(writeRequirements);
+ this.useMergeAppendForStreaming = writeConf.useMergeAppendForStreaming();
this.writeRequirements = writeRequirements;
this.outputSpecId = writeConf.outputSpecId();
this.writeProperties = writeConf.writeProperties();
@@ -622,7 +624,7 @@ abstract class SparkWrite implements Write,
RequiresDistributionAndOrdering {
@Override
protected void doCommit(long epochId, WriterCommitMessage[] messages) {
- AppendFiles append = table.newFastAppend();
+ AppendFiles append = useMergeAppendForStreaming ? table.newAppend() :
table.newFastAppend();
int numFiles = 0;
for (DataFile file : files(messages)) {
append.appendFile(file);
diff --git
a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java
b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java
index ab76001053..4e243da54f 100644
---
a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java
+++
b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java
@@ -34,10 +34,12 @@ import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SortOrder;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
import org.apache.iceberg.hadoop.HadoopTables;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.spark.SparkWriteOptions;
import org.apache.iceberg.spark.TestBase;
import org.apache.iceberg.types.Types;
import org.apache.spark.sql.Dataset;
@@ -147,6 +149,70 @@ public class TestStructuredStreaming {
}
}
+ @Test
+ public void testStreamingWriteAppendModeWithMergeAppend() throws Exception {
+ File parent = temp.resolve("parquet").toFile();
+ File location = new File(parent, "test-table");
+ File checkpoint = new File(parent, "checkpoint");
+
+ HadoopTables tables = new HadoopTables(CONF);
+ PartitionSpec spec =
PartitionSpec.builderFor(SCHEMA).identity("data").build();
+ // set a low min-count-to-merge so the merge append visibly consolidates
manifests
+ Table table =
+ tables.create(
+ SCHEMA,
+ spec,
+ ImmutableMap.of(TableProperties.MANIFEST_MIN_MERGE_COUNT, "2"),
+ location.toString());
+
+ List<SimpleRecord> expected =
+ Lists.newArrayList(
+ new SimpleRecord(1, "1"),
+ new SimpleRecord(2, "2"),
+ new SimpleRecord(3, "3"),
+ new SimpleRecord(4, "4"));
+
+ MemoryStream<Integer> inputStream = newMemoryStream(1, spark.sqlContext(),
Encoders.INT());
+ DataStreamWriter<Row> streamWriter =
+ inputStream
+ .toDF()
+ .selectExpr("value AS id", "CAST (value AS STRING) AS data")
+ .writeStream()
+ .outputMode("append")
+ .format("iceberg")
+ .option("checkpointLocation", checkpoint.toString())
+ .option("path", location.toString())
+ .option(SparkWriteOptions.USE_MERGE_APPEND_FOR_STREAMING, "true");
+
+ try {
+ StreamingQuery query = streamWriter.start();
+ List<Integer> batch1 = Lists.newArrayList(1, 2);
+ send(batch1, inputStream);
+ query.processAllAvailable();
+ List<Integer> batch2 = Lists.newArrayList(3, 4);
+ send(batch2, inputStream);
+ query.processAllAvailable();
+ query.stop();
+
+ Dataset<Row> result =
spark.read().format("iceberg").load(location.toString());
+ List<SimpleRecord> actual =
+
result.orderBy("id").as(Encoders.bean(SimpleRecord.class)).collectAsList();
+
+ assertThat(actual).hasSameSizeAs(expected).isEqualTo(expected);
+ assertThat(table.snapshots()).as("Number of snapshots should
match").hasSize(2);
+
+ // the second streaming append uses the regular append, which merges the
two data manifests
+ // into a single manifest; a fast append would have left two separate
manifests
+ assertThat(table.currentSnapshot().dataManifests(table.io()))
+ .as("Merge append should consolidate data manifests")
+ .hasSize(1);
+ } finally {
+ for (StreamingQuery query : spark.streams().active()) {
+ query.stop();
+ }
+ }
+ }
+
@Test
public void testStreamingWriteCompleteMode() throws Exception {
File parent = temp.resolve("parquet").toFile();
diff --git
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java
index b5bea931cd..ddaeced8f5 100644
---
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java
+++
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java
@@ -244,11 +244,11 @@ public class SparkWriteConf {
.parse();
}
- public boolean streamingMergeAppendEnabled() {
+ public boolean useMergeAppendForStreaming() {
return confParser
.booleanConf()
- .option(SparkWriteOptions.STREAMING_MERGE_APPEND_ENABLED)
- .defaultValue(SparkWriteOptions.STREAMING_MERGE_APPEND_ENABLED_DEFAULT)
+ .option(SparkWriteOptions.USE_MERGE_APPEND_FOR_STREAMING)
+ .defaultValue(SparkWriteOptions.USE_MERGE_APPEND_FOR_STREAMING_DEFAULT)
.parse();
}
diff --git
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java
index 23a0ab8a06..d7a62e2b1e 100644
---
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java
+++
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkWriteOptions.java
@@ -94,6 +94,6 @@ public class SparkWriteOptions {
public static final String VARIANT_INFERENCE_BUFFER_SIZE =
"variant-inference-buffer-size";
// Uses the merge append instead of fast append for streaming appends
- public static final String STREAMING_MERGE_APPEND_ENABLED =
"streaming-merge-append-enabled";
- public static final boolean STREAMING_MERGE_APPEND_ENABLED_DEFAULT = false;
+ public static final String USE_MERGE_APPEND_FOR_STREAMING =
"use-merge-append-for-streaming";
+ public static final boolean USE_MERGE_APPEND_FOR_STREAMING_DEFAULT = false;
}
diff --git
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java
index 07b975a007..f589fdd8f4 100644
---
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java
+++
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java
@@ -110,7 +110,7 @@ abstract class SparkWrite extends BaseSparkWrite implements
Write, RequiresDistr
private final StructType dsSchema;
private final Map<String, String> extraSnapshotMetadata;
private final boolean useFanoutWriter;
- private final boolean streamingMergeAppendEnabled;
+ private final boolean useMergeAppendForStreaming;
private final SparkWriteRequirements writeRequirements;
private final Map<String, String> writeProperties;
@@ -141,7 +141,7 @@ abstract class SparkWrite extends BaseSparkWrite implements
Write, RequiresDistr
this.dsSchema = dsSchema;
this.extraSnapshotMetadata = writeConf.extraSnapshotMetadata();
this.useFanoutWriter = writeConf.useFanoutWriter(writeRequirements);
- this.streamingMergeAppendEnabled = writeConf.streamingMergeAppendEnabled();
+ this.useMergeAppendForStreaming = writeConf.useMergeAppendForStreaming();
this.writeRequirements = writeRequirements;
this.outputSpecId = writeConf.outputSpecId();
this.writeProperties = writeConf.writeProperties();
@@ -636,7 +636,7 @@ abstract class SparkWrite extends BaseSparkWrite implements
Write, RequiresDistr
@Override
protected void doCommit(long epochId, WriterCommitMessage[] messages) {
- AppendFiles append = streamingMergeAppendEnabled ? table.newAppend() :
table.newFastAppend();
+ AppendFiles append = useMergeAppendForStreaming ? table.newAppend() :
table.newFastAppend();
int numFiles = 0;
for (DataFile file : files(messages)) {
append.appendFile(file);
diff --git
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java
index d96de81f92..d9b8815ed8 100644
---
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java
+++
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java
@@ -181,7 +181,7 @@ public class TestStructuredStreaming {
.format("iceberg")
.option("checkpointLocation", checkpoint.toString())
.option("path", location.toString())
- .option(SparkWriteOptions.STREAMING_MERGE_APPEND_ENABLED, "true");
+ .option(SparkWriteOptions.USE_MERGE_APPEND_FOR_STREAMING, "true");
try {
StreamingQuery query = streamWriter.start();
