This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 59e7296e3c6 [fix](iceberg) Write v1 tables whose partition spec has
void fields (#69358)
59e7296e3c6 is described below
commit 59e7296e3c67f7e5296564149d9eb3846dbf46c0
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Fri Oct 23 08:45:44 2026 +0100
[fix](iceberg) Write v1 tables whose partition spec has void fields (#69358)
### What problem does this PR solve?
Issue Number: None
Related PR: #69229
Problem Summary:
**In short.** Doris rejects every write to a format-v1 Iceberg table
whose current partition spec holds a void field, the field Iceberg keeps
for each partition field dropped from a v1 table. This PR pins such a
spec as it is, skips void fields wherever the write path looks for a
partition source column, and lets BE write NULL for a void field without
resolving its source. The problem came up in the end-to-end checks for
#69229.
**Background.** Dropping a partition field from a v1 table does not
remove it from the spec: Iceberg replaces it with a void transform,
which yields NULL for every row (v2 removes the field). The column the
field came from can be dropped afterwards, which Iceberg allows because
a void field never reads it. Before planning a write, Doris pins the
schema and the partition spec in `IcebergWriteSchemaContext`, and at
commit time it checks that the table still has the pinned spec with the
same definition.
**The problem, and what it cost.**
| Current spec of the v1 table | Spark 4.0 with Iceberg 1.10 | Doris
master | Doris with this PR |
|---|---|---|---|
| `[void(region)]`, region kept or dropped | Writes | INSERT and INSERT
OVERWRITE fail: `pinned spec 1 is not available with the same
definition; retry the statement` | Written like an unpartitioned table
and committed under spec 1 |
| `[void(region), identity(category)]`, region dropped | Reads, but
cannot write: its partition key needs an accessor for every source field
| Planning fails: `Iceberg partition field 1000 references source field
2 outside pinned schema 1` | Written under spec 1 with region NULL,
routed by category |
| `[void(region), identity(category)]`, region kept | Writes | Works |
Works |
| The tables above after an upgrade to format v2 | As above | DELETE,
UPDATE and MERGE fail the same way | Work |
- `bindPartitionSpec` replaces every spec whose `isPartitioned()` is
false with an unpartitioned stand-in that has the same spec id. An
all-void spec is not partitioned in that sense but still has fields, so
the stand-in's JSON never equals the table's spec and the commit-time
check always fails.
- `validateWriterMetadataSources` requires the source column of every
partition field, void included, and BE
`VIcebergTableWriter::_to_iceberg_partition_columns` resolves the source
of every field as well.
**How this PR fixes it.**
- `IcebergWriteSchemaContext` gives the unpartitioned stand-in only to a
spec without fields and binds every other spec as it is, so the pinned
definition equals the table's. Its source check skips void fields.
- `IcebergWritePlanProvider` skips void fields in the INSERT routing
(`getWriteDistribution`) and in the MERGE insert partitioning
(`getWritePartitioning`). They are NULL for every row; without the skip,
a dropped source turns the write random, and a table with many
partitions can exceed
`table_sink_partition_write_max_partition_nums_per_writer` on one
writer.
- BE `VIcebergTableWriter` no longer resolves a source for a void field
and fills a NULL column for it, so the file lands under
`region=null/...` and commits NULL for the field, as Spark writes the
same spec while the column exists. Iceberg itself allows dropping that
column, because a void field never reads it.
- An all-void spec stays unpartitioned for the write: the sink carries
no spec, BE writes unpartitioned files, and the commit builds the data
files with `DataFiles.builder(spec)` and no partition values, the path
Spark's unpartitioned writer takes. A dynamic overwrite replaces the
whole table, since `ReplacePartitions` checks `isUnpartitioned()`.
- During a rolling upgrade, a new FE with an old BE still fails for a
void field whose source column was dropped, with the old BE's `outside
writer schema` error before any file is written, so no exec version gate
is needed.
Branch-4.1 and branch-4.2 have the same code in fe-core
`IcebergWriteSchemaContext` and in BE.
**Results.**
- Unit tests: the new
`IcebergWritePlanProviderTest#writeSchemaPinsV1VoidPartitionFields` and
`#writeRoutingSkipsV1VoidPartitionFields`,
`IcebergConnectorTransactionTest#insertCommitsNullForV1VoidPartitionFields`
(pins, commits and checks the committed data file for an all-void spec
and for a void field whose source was dropped) and
`VIcebergTableWriterTest.VoidPartitionFieldNeedsNoSourceColumn` pass,
with all 1625 tests of fe-connector-iceberg and 91 BE tests of the
Iceberg writer, partition, MERGE / DELETE sink and related suites.
Reverting each of the six changes (the stand-in, the source check, the
INSERT routing, the MERGE partitioning, the BE source resolution, the BE
NULL column) fails the tests that cover it.
- Regression test: `test_iceberg_write_v1_void_partition` creates three
v1 tables in Spark (void only, void only with the column dropped, void
plus identity with the column dropped), writes them with Doris INSERT
and INSERT OVERWRITE, upgrades them to merge-on-read v2 and runs DELETE,
UPDATE and MERGE, comparing every step with Spark, including Spark's
partition pruning on the identity field. With the master connector the
first Doris INSERT fails with the `pinned spec 1` error; with this PR
the suite passes, as do all 43 suites of
`external_table_p0/iceberg/write`. Spark only reads the void plus
identity table, which it cannot write.
- Manual test, on a local Hadoop catalog with the same three shapes:
INSERT, INSERT OVERWRITE, and DELETE / UPDATE / MERGE after a v2 upgrade
return the expected rows, and the Iceberg API shows every new data and
delete file under the current spec with a NULL void value
(`region=null/category=x/` for the identity table, the table root for
the void-only tables).
---
.../sink/writer/iceberg/viceberg_table_writer.cpp | 18 ++
.../writer/iceberg/viceberg_table_writer_test.cpp | 44 +++++
.../iceberg/IcebergWritePlanProvider.java | 8 +
.../iceberg/IcebergWriteSchemaContext.java | 8 +-
.../iceberg/IcebergConnectorTransactionTest.java | 42 +++++
.../iceberg/IcebergWritePlanProviderTest.java | 62 +++++++
.../write/test_iceberg_write_v1_void_partition.out | 70 ++++++++
.../test_iceberg_write_v1_void_partition.groovy | 187 +++++++++++++++++++++
8 files changed, 438 insertions(+), 1 deletion(-)
diff --git a/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp
b/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp
index 7aa81a7387b..95f78cc0678 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp
+++ b/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp
@@ -29,6 +29,7 @@
#include "core/column/column_struct.h"
#include "core/column/column_vector.h"
#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_string.h"
#include "core/data_type/data_type_struct.h"
#include "core/data_type_serde/data_type_serde.h"
#include "exec/sink/writer/external_file_writer_compatibility.h"
@@ -133,6 +134,14 @@ VIcebergTableWriter::_to_iceberg_partition_columns() {
std::vector<IcebergPartitionColumn> partition_columns;
for (const auto& partition_field : _partition_spec->fields()) {
+ if (partition_field.transform() == "void") {
+ // A void field is NULL for every row. A v1 table keeps one for
each dropped partition
+ // field and may drop its source column later, so the source is
not resolved.
+ partition_columns.emplace_back(partition_field, TYPE_NULL, -1,
std::vector<size_t> {},
+
std::make_unique<VoidPartitionColumnTransform>(
+
std::make_shared<DataTypeString>()));
+ continue;
+ }
const auto* field_path =
_schema->find_field_path(partition_field.source_id());
if (field_path == nullptr || field_path->empty()) {
throw Exception(
@@ -528,6 +537,15 @@ Status VIcebergTableWriter::_write_prepared_block(Block&
output_block) {
auto col = ColumnConst::create(std::move(data_col),
output_block.rows());
transformed_block.insert(
{std::move(col), result_type,
iceberg_partition_columns.field().name()});
+ } else if (iceberg_partition_columns.source_idx() < 0) {
+ // A void field has no source column; materialize its NULLs,
since partition values
+ // are read row by row.
+ auto result_type = make_nullable(
+
iceberg_partition_columns.partition_column_transform().get_result_type());
+ auto data_col = result_type->create_column();
+ data_col->insert_many_defaults(output_block.rows());
+ transformed_block.insert({std::move(data_col), result_type,
+
iceberg_partition_columns.field().name()});
} else {
Block source_block;
source_block.insert(
diff --git a/be/test/exec/sink/writer/iceberg/viceberg_table_writer_test.cpp
b/be/test/exec/sink/writer/iceberg/viceberg_table_writer_test.cpp
index c16ba9a2164..7ccb060ef5d 100644
--- a/be/test/exec/sink/writer/iceberg/viceberg_table_writer_test.cpp
+++ b/be/test/exec/sink/writer/iceberg/viceberg_table_writer_test.cpp
@@ -351,6 +351,50 @@ TEST(VIcebergTableWriterTest,
RejectMissingPartitionSource) {
}
}
+TEST(VIcebergTableWriterTest, VoidPartitionFieldNeedsNoSourceColumn) {
+ // A v1 table keeps void(region) for its dropped partition field after
region is dropped too.
+ std::vector<iceberg::NestedField> columns;
+ columns.emplace_back(true, 1, "note",
std::make_unique<iceberg::IntegerType>(), std::nullopt);
+ columns.emplace_back(true, 3, "category",
std::make_unique<iceberg::IntegerType>(),
+ std::nullopt);
+ auto schema = std::make_shared<iceberg::Schema>(std::move(columns));
+ const std::string spec_json =
+
R"({"spec-id":1,"fields":[{"name":"region","transform":"void","source-id":2,)"
+
R"("field-id":1000},{"name":"category","transform":"identity","source-id":3,)"
+ R"("field-id":1001}]})";
+ auto int_type = make_nullable(std::make_shared<DataTypeInt32>());
+ VExprContextSPtrs output_exprs {
+ VExprContext::create_shared(VSlotRef::create_shared(0, 0, -1,
int_type, "note")),
+ VExprContext::create_shared(VSlotRef::create_shared(1, 1, -1,
int_type, "category"))};
+
+ TIcebergTableSink iceberg_sink;
+ TDataSink data_sink;
+ data_sink.__set_iceberg_table_sink(iceberg_sink);
+ VIcebergTableWriter writer(data_sink, output_exprs, nullptr, nullptr);
+ writer._schema = schema;
+ writer._partition_spec = iceberg::PartitionSpecParser::from_json(schema,
spec_json);
+ writer._iceberg_partition_columns = writer._to_iceberg_partition_columns();
+ ASSERT_EQ(writer._iceberg_partition_columns.size(), 2);
+ EXPECT_EQ(writer._iceberg_partition_columns[0].source_idx(), -1);
+ EXPECT_EQ(writer._iceberg_partition_columns[1].source_idx(), 1);
+
+ IcebergPartitionData partition({std::any(), Int32(7)});
+ EXPECT_EQ("region=null/category=7", writer._partition_to_path(partition));
+ EXPECT_EQ(std::vector<std::string>({"null", "7"}),
writer._partition_values(partition));
+ auto recording_writer = std::make_shared<RecordingPartitionWriter>();
+ writer._partitions_to_writers[writer._partition_to_writer_key(partition)]
= recording_writer;
+ auto note = int_type->create_column();
+ note->insert_many_defaults(2);
+ auto category = int_type->create_column();
+ category->insert(Field::create_field<TYPE_INT>(7));
+ category->insert(Field::create_field<TYPE_INT>(7));
+ Block block({ColumnWithTypeAndName(note->get_ptr(), int_type, "note"),
+ ColumnWithTypeAndName(category->get_ptr(), int_type,
"category")});
+ // Both rows must reach the writer of the partition whose void value is
NULL.
+ ASSERT_TRUE(writer._write_prepared_block(block).ok());
+ EXPECT_EQ(2, recording_writer->rows);
+}
+
TEST(VIcebergTableWriterTest, ResolvesNestedPartitionSource) {
std::vector<iceberg::NestedField> children;
children.emplace_back(true, 2, "part",
std::make_unique<iceberg::IntegerType>(), std::nullopt);
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java
index 5c8d0dc7ca3..10a0ff9e6f7 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java
@@ -621,6 +621,10 @@ public class IcebergWritePlanProvider implements
ConnectorWritePlanProvider {
? active.get().getSchema() : table.schema();
List<ConnectorWritePartitionField> fields = new ArrayList<>();
for (PartitionField field : spec.fields()) {
+ if (field.transform().isVoid()) {
+ // NULL for every row, and a v1 table may have dropped its
source column.
+ continue;
+ }
// Bind a nested source to its top-level slot. fe-core recovers
child indexes from the stable Iceberg
// field ids already carried by the Doris Column tree, so the
public connector SPI stays unchanged.
String sourceColumnName = findPartitionSourceColumnName(schema,
field.sourceId());
@@ -664,6 +668,10 @@ public class IcebergWritePlanProvider implements
ConnectorWritePlanProvider {
List<String> routeColumns = new ArrayList<>();
List<String> transforms = new ArrayList<>();
for (PartitionField field : spec.fields()) {
+ if (field.transform().isVoid()) {
+ // NULL for every row, and a v1 table may have dropped its
source column.
+ continue;
+ }
String sourceColumnName = findTopLevelColumnName(schema,
field.sourceId());
if (sourceColumnName == null) {
return null;
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriteSchemaContext.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriteSchemaContext.java
index ad535befa34..bb6c4bb7963 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriteSchemaContext.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriteSchemaContext.java
@@ -222,7 +222,9 @@ final class IcebergWriteSchemaContext {
private static PartitionSpec bindPartitionSpec(
PartitionSpec partitionSpec, Schema schema, String tableName) {
- if (!partitionSpec.isPartitioned()) {
+ // A v1 table replaces each dropped partition field with a void one,
so a spec that is not partitioned can
+ // still have fields. Bind those as well: the pinned definition must
stay equal to the table's spec.
+ if (partitionSpec.fields().isEmpty()) {
return
PartitionSpec.builderFor(schema).withSpecId(partitionSpec.specId()).build();
}
try {
@@ -250,6 +252,10 @@ final class IcebergWriteSchemaContext {
private static void validateWriterMetadataSources(
Schema schema, PartitionSpec partitionSpec, SortOrder sortOrder,
String tableName) {
for (PartitionField field : partitionSpec.fields()) {
+ // A void field writes NULL without reading its source, which a v1
table may drop after the field.
+ if (field.transform().isVoid()) {
+ continue;
+ }
// Iceberg permits a nested primitive field as a partition source;
field IDs are schema-wide.
if (schema.findField(field.sourceId()) == null) {
throw new DorisConnectorException("Iceberg partition field " +
field.fieldId()
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java
index ddd1ac2c7bf..c0b42a634dc 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java
@@ -944,6 +944,48 @@ public class IcebergConnectorTransactionTest {
Assertions.assertThrows(DorisConnectorException.class, txn::commit);
}
+ @Test
+ public void insertCommitsNullForV1VoidPartitionFields() {
+ // A v1 table replaces a dropped partition field with void(region) and
may drop region afterwards; Spark
+ // keeps writing it. MUTATION: pinning an unpartitioned stand-in for
the void-only spec -> commit rejects the
+ // pinned spec; checking the source of the void field -> create
rejects the dropped region -> red.
+ for (boolean identityCategory : new boolean[] {false, true}) {
+ InMemoryCatalog catalog = freshCatalog();
+ TableIdentifier id = TableIdentifier.of("db1", "t1");
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id",
Types.IntegerType.get()),
+ Types.NestedField.optional(2, "region",
Types.StringType.get()),
+ Types.NestedField.optional(3, "category",
Types.StringType.get()));
+ PartitionSpec.Builder spec =
PartitionSpec.builderFor(schema).identity("region");
+ if (identityCategory) {
+ spec.identity("category");
+ }
+ Table table = catalog.createTable(id, schema, spec.build(),
props("format-version", "1"));
+ table.updateSpec().removeField("region").commit();
+ table.updateSchema().deleteColumn("region").commit();
+
+ IcebergWriteSchemaContext schemaContext =
+ IcebergWriteSchemaContext.create(table, table.name(),
Optional.empty(), false, false);
+ IcebergConnectorTransaction txn = txnFor(opsReturning(table), new
RecordingConnectorContext());
+ txn.beginWrite(SESSION, "db1", "t1", new IcebergWriteContext(
+ WriteOperation.INSERT, false, Collections.emptyMap(),
Optional.empty(), -1L, schemaContext));
+ // BE writes a void-only spec as unpartitioned and reports NULL
for a void field of a partitioned one.
+ txn.addCommitData(commitBytes(identityCategory
+ ?
dataFileItem("s3://b/db1/t1/region=null/category=x/f1.parquet", 2L, 1024L,
+ Arrays.asList("null", "x"))
+ : dataFileItem("s3://b/db1/t1/f1.parquet", 2L, 1024L)));
+ txn.commit();
+
+ List<DataFile> files = currentDataFiles(catalog.loadTable(id));
+ Assertions.assertEquals(1, files.size());
+ Assertions.assertEquals(table.spec().specId(),
files.get(0).specId());
+ Assertions.assertNull(files.get(0).partition().get(0,
Object.class));
+ if (identityCategory) {
+ Assertions.assertEquals("x", files.get(0).partition().get(1,
String.class));
+ }
+ }
+ }
+
@Test
public void overwriteStaticPartitionUsesRowFilter() {
InMemoryCatalog catalog = freshCatalog();
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
index 0518bc94e52..9193fbbc79b 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
@@ -504,6 +504,68 @@ public class IcebergWritePlanProviderTest {
table, "db1.nested_partition", Optional.empty(), false,
false));
}
+ /** A v1 table whose region partition field was dropped, which Iceberg
keeps as void(region). */
+ private static Table v1TableWithVoidRegion(InMemoryCatalog catalog, String
name, boolean identityCategory,
+ boolean dropRegion) {
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "region",
Types.StringType.get()),
+ Types.NestedField.optional(3, "category",
Types.StringType.get()));
+ PartitionSpec.Builder spec =
PartitionSpec.builderFor(schema).identity("region");
+ if (identityCategory) {
+ spec.identity("category");
+ }
+ Table table = catalog.createTable(TableIdentifier.of("db1", name),
schema, spec.build(),
+ Collections.singletonMap("format-version", "1"));
+ table.updateSpec().removeField("region").commit();
+ if (dropRegion) {
+ table.updateSchema().deleteColumn("region").commit();
+ }
+ return table;
+ }
+
+ @Test
+ public void writeSchemaPinsV1VoidPartitionFields() {
+ // Spark keeps writing a v1 table after a partition field becomes
void, also once its source is dropped.
+ // MUTATION: pinning an unpartitioned stand-in for a void-only spec ->
the revalidation rejects it; checking
+ // the source column of a void field -> create rejects the dropped
region -> red.
+ InMemoryCatalog catalog = freshCatalog();
+ List<Table> tables = Arrays.asList(
+ v1TableWithVoidRegion(catalog, "void_only", false, false),
+ v1TableWithVoidRegion(catalog, "void_only_dropped", false,
true),
+ v1TableWithVoidRegion(catalog, "void_and_identity_dropped",
true, true));
+ for (Table table : tables) {
+ IcebergWriteSchemaContext writeContext =
IcebergWriteSchemaContext.create(
+ table, table.name(), Optional.empty(), false, false);
+
+ Assertions.assertEquals(PartitionSpecParser.toJson(table.spec()),
writeContext.getPartitionSpecJson(),
+ table.name());
+ Assertions.assertDoesNotThrow(() ->
writeContext.validateCurrentSchema(table, true), table.name());
+ }
+ }
+
+ @Test
+ public void writeRoutingSkipsV1VoidPartitionFields() {
+ // A void field is NULL for every row, so routing by the other fields
still keeps one partition on one
+ // writer, and the void field's source may be gone. MUTATION: routing
the void field -> no route column for
+ // the dropped region -> random write -> red.
+ Table table = v1TableWithVoidRegion(freshCatalog(),
"void_and_identity", true, true);
+ IcebergTableHandle handle = new IcebergTableHandle("db1",
"void_and_identity");
+
+ ConnectorWriteDistribution distribution = providerFor(table,
contextWithStorage())
+ .getWriteDistribution(sessionFor(table, contextWithStorage()),
handle);
+ Assertions.assertNotNull(distribution);
+ Assertions.assertEquals(Collections.singletonList("category"),
distribution.getRouteColumns());
+ Assertions.assertEquals(Collections.singletonMap("transforms",
"identity"),
+ distribution.getPartitionFunctionOptions());
+
+ ConnectorWritePartitionSpec partitioning = providerFor(table,
contextWithStorage())
+ .getWritePartitioning(sessionFor(table, contextWithStorage()),
handle);
+ Assertions.assertEquals(table.spec().specId(),
partitioning.getSpecId());
+ Assertions.assertEquals(1, partitioning.getFields().size());
+ Assertions.assertEquals("category",
partitioning.getFields().get(0).getSourceColumnName());
+ }
+
@Test
public void branchWriteRejectsCurrentRequiredFieldAbsentWithoutDefault() {
InMemoryCatalog catalog = freshCatalog();
diff --git
a/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_v1_void_partition.out
b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_v1_void_partition.out
new file mode 100644
index 00000000000..e80dfdc4ce6
--- /dev/null
+++
b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_v1_void_partition.out
@@ -0,0 +1,70 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !void_only_insert --
+1 a 10
+2 b 20
+3 a 30
+4 c 40
+5 d 50
+
+-- !void_only_overwrite --
+1 a 11
+2 b 21
+3 a 31
+4 c 41
+5 d 51
+
+-- !void_only_overwrite_specs --
+1
+
+-- !void_only_dml --
+2 b 21
+3 a 131
+4 c 400
+5 d 51
+6 f 60
+
+-- !void_only_delete_specs --
+1
+
+-- !void_only_dropped_source_insert --
+1 10
+2 20
+3 30
+4 40
+
+-- !void_only_dropped_source_overwrite --
+1 11
+2 21
+3 31
+4 41
+
+-- !void_only_dropped_source_dml --
+2 21
+3 131
+4 400
+6 60
+
+-- !void_and_identity_insert --
+1 x 10
+2 y 20
+3 x 30
+4 z 40
+5 z 50
+
+-- !void_and_identity_overwrite --
+1 x 11
+2 y 21
+3 x 31
+4 z 41
+5 z 51
+
+-- !void_and_identity_dml --
+2 y 200
+3 x 31
+4 z 141
+5 z 151
+6 w 60
+
+-- !void_and_identity_delete_specs --
+1
+
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_v1_void_partition.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_v1_void_partition.groovy
new file mode 100644
index 00000000000..030fad224f6
--- /dev/null
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_v1_void_partition.groovy
@@ -0,0 +1,187 @@
+// 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.
+
+// A format-v1 Iceberg table replaces every partition field it drops with a
void transform, and
+// the source column of that field may be dropped afterwards. Doris must write
such tables under
+// the current spec, also after an upgrade to format v2 for row-level DML, and
Spark must read
+// what Doris wrote.
+suite("test_iceberg_write_v1_void_partition",
+ "p0,external,iceberg,external_docker,external_docker_iceberg") {
+ if
(!"true".equalsIgnoreCase(context.config.otherConfigs.get("enableIcebergTest")))
{
+ logger.info("disable iceberg test")
+ return
+ }
+
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String catalogName = "test_iceberg_write_v1_void_partition"
+ String dbName = "iceberg_write_v1_void_partition_db"
+ sql """drop catalog if exists ${catalogName}"""
+ sql """
+ create catalog ${catalogName} properties (
+ "type" = "iceberg",
+ "iceberg.catalog.type" = "rest",
+ "uri" = "http://${externalEnvIp}:${restPort}";,
+ "s3.access_key" = "admin",
+ "s3.secret_key" = "password",
+ "s3.endpoint" = "http://${externalEnvIp}:${minioPort}";,
+ "s3.region" = "us-east-1"
+ )
+ """
+ sql """switch ${catalogName}"""
+ sql """create database if not exists ${dbName}"""
+ sql """use ${dbName}"""
+
+ def checkRows = { String table, String columns, String predicate = "true"
->
+ sql """refresh table ${dbName}.${table}"""
+ spark_iceberg """refresh table demo.${dbName}.${table}"""
+ assertSparkDorisResultEquals(
+ spark_iceberg("""
+ select ${columns} from demo.${dbName}.${table} where
${predicate} order by id
+ """),
+ sql("""select ${columns} from ${table} where ${predicate}
order by id"""))
+ }
+
+ def upgradeToMergeOnReadV2 = { String table ->
+ spark_iceberg """
+ alter table demo.${dbName}.${table} set tblproperties (
+ 'format-version' = '2',
+ 'write.delete.mode' = 'merge-on-read',
+ 'write.update.mode' = 'merge-on-read',
+ 'write.merge.mode' = 'merge-on-read')
+ """
+ sql """refresh table ${dbName}.${table}"""
+ }
+
+ // Dropping the only partition field leaves spec 1 = [void(region)].
+ spark_iceberg """drop table if exists demo.${dbName}.void_only"""
+ spark_iceberg """
+ create table demo.${dbName}.void_only (id int, region string, metric
int)
+ using iceberg partitioned by (region)
+ tblproperties ('format-version' = '1', 'write.format.default' =
'parquet')
+ """
+ spark_iceberg """insert into demo.${dbName}.void_only values (1, 'a', 10),
(2, 'b', 20)"""
+ spark_iceberg """alter table demo.${dbName}.void_only drop partition field
region"""
+ sql """refresh database ${dbName}"""
+ sql """insert into void_only values (3, 'a', 30), (4, 'c', 40)"""
+ spark_iceberg """refresh table demo.${dbName}.void_only"""
+ spark_iceberg """insert into demo.${dbName}.void_only values (5, 'd',
50)"""
+ checkRows("void_only", "id, region, metric")
+ order_qt_void_only_insert """select id, region, metric from void_only"""
+
+ // A bare INSERT OVERWRITE replaces the whole table, so every remaining
file uses spec 1.
+ sql """insert overwrite table void_only select id, region, metric + 1 from
void_only"""
+ checkRows("void_only", "id, region, metric")
+ order_qt_void_only_overwrite """select id, region, metric from void_only"""
+ order_qt_void_only_overwrite_specs """select distinct spec_id from
void_only\$data_files"""
+
+ upgradeToMergeOnReadV2("void_only")
+ sql """delete from void_only where id = 1"""
+ sql """update void_only set metric = metric + 100 where id = 3"""
+ sql """
+ merge into void_only t
+ using (select 4 as id, 'c' as region, 400 as metric union all select
6, 'f', 60) s
+ on t.id = s.id
+ when matched then update set metric = s.metric
+ when not matched then insert (id, region, metric) values (s.id,
s.region, s.metric)
+ """
+ checkRows("void_only", "id, region, metric")
+ order_qt_void_only_dml """select id, region, metric from void_only"""
+ order_qt_void_only_delete_specs """select distinct spec_id from
void_only\$delete_files"""
+
+ // The void field keeps referencing the dropped column region.
+ spark_iceberg """drop table if exists
demo.${dbName}.void_only_dropped_source"""
+ spark_iceberg """
+ create table demo.${dbName}.void_only_dropped_source (id int, region
string, metric int)
+ using iceberg partitioned by (region)
+ tblproperties ('format-version' = '1', 'write.format.default' = 'orc')
+ """
+ spark_iceberg """
+ insert into demo.${dbName}.void_only_dropped_source values (1, 'a',
10), (2, 'b', 20)
+ """
+ spark_iceberg """alter table demo.${dbName}.void_only_dropped_source drop
partition field region"""
+ spark_iceberg """alter table demo.${dbName}.void_only_dropped_source drop
column region"""
+ sql """refresh database ${dbName}"""
+ sql """insert into void_only_dropped_source values (3, 30)"""
+ spark_iceberg """refresh table demo.${dbName}.void_only_dropped_source"""
+ spark_iceberg """insert into demo.${dbName}.void_only_dropped_source
values (4, 40)"""
+ checkRows("void_only_dropped_source", "id, metric")
+ order_qt_void_only_dropped_source_insert """select id, metric from
void_only_dropped_source"""
+
+ // Rewriting the historical files under spec 1 also clears their non-null
region values, which
+ // row-level DML could not carry into delete files once the column is gone.
+ sql """insert overwrite table void_only_dropped_source select id, metric +
1 from void_only_dropped_source"""
+ checkRows("void_only_dropped_source", "id, metric")
+ order_qt_void_only_dropped_source_overwrite """select id, metric from
void_only_dropped_source"""
+
+ upgradeToMergeOnReadV2("void_only_dropped_source")
+ sql """delete from void_only_dropped_source where id = 1"""
+ sql """update void_only_dropped_source set metric = metric + 100 where id
= 3"""
+ sql """
+ merge into void_only_dropped_source t
+ using (select 4 as id, 400 as metric union all select 6, 60) s
+ on t.id = s.id
+ when matched then update set metric = s.metric
+ when not matched then insert (id, metric) values (s.id, s.metric)
+ """
+ checkRows("void_only_dropped_source", "id, metric")
+ order_qt_void_only_dropped_source_dml """select id, metric from
void_only_dropped_source"""
+
+ // Spec 1 = [void(region), identity(category)] after region is dropped:
new files must carry
+ // a null region and their category value. Spark cannot write this table
once region is gone
+ // (its partition key needs an accessor for every source field), but it
reads what Doris writes.
+ spark_iceberg """drop table if exists demo.${dbName}.void_and_identity"""
+ spark_iceberg """
+ create table demo.${dbName}.void_and_identity (
+ id int, region string, category string, metric int)
+ using iceberg partitioned by (region, category)
+ tblproperties ('format-version' = '1', 'write.format.default' =
'parquet')
+ """
+ spark_iceberg """
+ insert into demo.${dbName}.void_and_identity values (1, 'a', 'x', 10),
(2, 'b', 'y', 20)
+ """
+ spark_iceberg """alter table demo.${dbName}.void_and_identity drop
partition field region"""
+ spark_iceberg """alter table demo.${dbName}.void_and_identity drop column
region"""
+ sql """refresh database ${dbName}"""
+ sql """insert into void_and_identity values (3, 'x', 30), (4, 'z', 40),
(5, 'z', 50)"""
+ checkRows("void_and_identity", "id, category, metric")
+ // Spark prunes on the identity field, so it only finds Doris rows with
the right partition value.
+ checkRows("void_and_identity", "id, category, metric", "category = 'z'")
+ order_qt_void_and_identity_insert """select id, category, metric from
void_and_identity"""
+
+ sql """insert overwrite table void_and_identity select id, category,
metric + 1 from void_and_identity"""
+ checkRows("void_and_identity", "id, category, metric")
+ checkRows("void_and_identity", "id, category, metric", "category = 'x'")
+ order_qt_void_and_identity_overwrite """select id, category, metric from
void_and_identity"""
+
+ upgradeToMergeOnReadV2("void_and_identity")
+ sql """delete from void_and_identity where id = 1"""
+ sql """update void_and_identity set metric = metric + 100 where category =
'z'"""
+ sql """
+ merge into void_and_identity t
+ using (select 2 as id, 'y' as category, 200 as metric
+ union all select 6, 'w', 60) s
+ on t.id = s.id
+ when matched then update set metric = s.metric
+ when not matched then insert (id, category, metric) values (s.id,
s.category, s.metric)
+ """
+ checkRows("void_and_identity", "id, category, metric")
+ checkRows("void_and_identity", "id, category, metric", "category = 'w'")
+ order_qt_void_and_identity_dml """select id, category, metric from
void_and_identity"""
+ order_qt_void_and_identity_delete_specs """select distinct spec_id from
void_and_identity\$delete_files"""
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
