[fix](iceberg) Write v1 tables whose partition spec has void fields

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"
@@ -132,6 +133,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(
@@ -526,6 +535,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(

be/test/exec/sink/writer/iceberg/viceberg_table_writer_test.cpp

@@ -284,6 +284,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);

fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java

@@ -621,6 +621,10 @@ public ConnectorWritePartitionSpec getWritePartitioning(ConnectorSession session
                 ? 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 ConnectorWriteDistribution getWriteDistribution(ConnectorSession session,
         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;

fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriteSchemaContext.java

@@ -222,7 +222,9 @@ private static ConnectorType stringWriteType(Type type, ConnectorType mapped) {
 
     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 @@ private static SortOrder bindSortOrder(SortOrder sortOrder, Schema schema, Strin
     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()

fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java

@@ -941,6 +941,48 @@ public void overwriteEmptyUnpartitionedRejectsDataCommittedBetweenScanAndBegin()
         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();

fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java

@@ -504,6 +504,68 @@ public void writeSchemaAllowsNestedPrimitivePartitionSource() {
                 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();

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
+