This is an automated email from the ASF dual-hosted git repository.
Gabriel39 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 37632033807 [fix](iceberg) Fix static partition writes for renamed
identity fields (#69188)
37632033807 is described below
commit 37632033807ac6160a0491ba9361fbe49962622e
Author: daidai <[email protected]>
AuthorDate: Tue Oct 20 10:55:11 2026 +0800
[fix](iceberg) Fix static partition writes for renamed identity fields
(#69188)
### What problem does this PR solve?
Issue Number: N/A
Related PR: N/A
Problem Summary:
Doris could not write an Iceberg partition by value when the identity
partition field had been renamed away from its source column (`ALTER
TABLE t ADD PARTITION KEY id AS id_identity`):
- `INSERT OVERWRITE TABLE t PARTITION (id = 2) (value, metric) VALUES
('b', 2)` failed with `Unknown partition column 'id' in table 't'.
Available partition columns: [id_identity]`, although `id` is the column
Spark resolves for a static partition clause and the one `SHOW CREATE
TABLE` renders for this table (`PARTITION BY LIST (\`id\`)`);
- `... PARTITION (id_identity = 2) (value, metric) ...` passed the
validation but then failed with `Column has no default value,
column=id`: `id_identity` is a partition FIELD name, not a table column,
so the static literal could not be materialized into any row column;
- `... PARTITION (id_identity = 2) SELECT id, value, metric ...` (no
column list) was accepted and wrote data files whose `id` column value
disagreed with their own partition value.
The two checks contradicted each other: the validation required a name
that the binder then rejected, so there was no usable spelling.
Fix:
- resolve a `PARTITION (col = val)` name to the identity partition
field's SOURCE column, case-insensitively — the identifier the binder
can materialize
(`PluginDrivenExternalTable#materializeStaticPartitionValues`);
- reject a renamed partition FIELD name with the source column to use
instead, and list the usable source columns for an unknown name;
non-identity transforms keep their existing rejection;
- rewrite the static partition spec keys to partition FIELD names before
the BE sink (`static_partition_values`) and the commit row filter
consume them, so the thrift contract, the overwrite row filter and the
BE writer stay unchanged.
Behavior is aligned with Spark:
- Spark's `Analyzer.partitionColumnNames` takes the identity transform's
field reference, so a static partition column is the SOURCE column
(`PARTITION (id = 2)`), not the renamed field name;
- `Analyzer.addStaticPartitionColumns` materializes that literal into
the source column and rejects the same column in an explicit insert
column list — Doris does the same after this change;
- the resulting overwrite is an `OverwriteFiles.overwriteByRowFilter(id
= 2)` (`Iceberg SparkWrite`), which is exactly what Doris builds once
the spec is field-keyed.
Effect after the fix: `PARTITION (id = 2)` overwrites only that
partition, the new files carry `id = 2` next to partition value `2`, and
Spark reads the same rows; `PARTITION (id_identity = 2)` now reports
`'id_identity' ... is a partition field name, not a table column; write
the identity partition by its source column 'id' instead.` instead of a
contradictory error or a corrupted write. Tables whose partition field
name already equals its source column are unaffected.
### Release note
Fix `INSERT [OVERWRITE] ... PARTITION (col = val)` on Iceberg tables
whose identity partition field was renamed with `ADD PARTITION KEY ...
AS ...`: the static value is now written through the identity source
column (`PARTITION (id = 2)`), matching Spark, instead of reporting
contradictory errors or writing files whose data column disagreed with
the partition value. Non-identity transforms and unpartitioned tables
keep their existing rejection.
### Check List (For Author)
- Test
- [x] Regression test
- [x] Unit Test
- [x] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason
- Manual test detail: full `fe-connector-iceberg` module unit tests (93
test classes) plus checkstyle pass locally with 0 failures on top of
current master; the new regression case (TC39) needs the external docker
Iceberg environment to run, so it is expected to be covered by the
regression pipeline.
- Behavior changed:
- [ ] No.
- [x] Yes.
- Does this need documentation?
- [x] No.
- [ ] Yes.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
.../iceberg/IcebergConnectorMetadata.java | 101 ++++++++++---
.../connector/iceberg/IcebergPartitionUtils.java | 99 +++++++++++++
.../iceberg/IcebergWritePlanProvider.java | 32 +++-
.../iceberg/IcebergConnectorMetadataTest.java | 162 +++++++++++++++++++--
.../iceberg/IcebergConnectorTransactionTest.java | 35 +++++
.../iceberg/IcebergPartitionUtilsTest.java | 70 +++++++++
.../iceberg/IcebergWritePlanProviderTest.java | 21 +++
.../test_iceberg_static_partition_overwrite.out | 3 +
.../test_iceberg_static_partition_overwrite.groovy | 62 ++++++--
9 files changed, 535 insertions(+), 50 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
index 134bf0cd9de..1e3aa164e26 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
@@ -1939,12 +1939,20 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
}
/**
- * Rejects an illegal static-partition column on {@code INSERT [OVERWRITE]
... PARTITION (col=val)}: the
- * column must be an <em>identity</em> partition field of this
(partitioned) table. Mirrors the legacy
- * fe-resident {@code BindSink.validateStaticPartition} (now dead on the
connector-driven path), but the
- * iceberg {@link PartitionSpec} knowledge and the messages live in the
connector. The lookup is keyed by
- * partition <em>field</em> name (e.g. {@code category_bucket} for {@code
bucket(4, category)}), matching
- * legacy — so a source-column name that is not itself an identity field
reads as "Unknown partition column".
+ * Rejects an illegal static-partition column on {@code INSERT [OVERWRITE]
... PARTITION (col=val)}: the name
+ * must be the SOURCE column of an <em>identity</em> partition field of
this (partitioned) table.
+ *
+ * <p>That is the identifier Spark resolves for a static partition clause
+ * ({@code Analyzer.partitionColumnNames} takes the identity transform's
field reference), the one
+ * {@code SHOW CREATE TABLE} renders ({@code PARTITION BY LIST (`id`)}),
and the only one the engine can
+ * materialize: an identity partition value IS the column value, so the
binder fills that column from the
+ * PARTITION literal ({@code
PluginDrivenExternalTable#materializeStaticPartitionValues}).</p>
+ *
+ * <p>Consequently a partition FIELD name that was renamed away from its
source column
+ * ({@code ADD PARTITION KEY id AS id_identity}) is rejected with the
source name to use instead. The field
+ * name is not a table column, so no value could be written for it —
accepting it is exactly the
+ * "PARTITION names a column the binder then cannot find" contradiction
this replaces. Non-identity
+ * transforms stay rejected, and an unknown name lists the source columns
that a static write can use.</p>
*/
@Override
public void validateStaticPartitionColumns(ConnectorSession session,
ConnectorTableHandle handle,
@@ -1953,30 +1961,83 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
return;
}
IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
- Table table = loadTable(session, iceHandle);
- PartitionSpec spec = table.spec();
String tableName = iceHandle.getTableName();
+ // Validate against the schema/spec this statement actually writes
with. A branch-targeted write pins the
+ // BRANCH's writer schema before analysis is complete (getWriteColumns
-> IcebergStatementScope.writeSchema)
+ // and binding, planning and the BE sink all consume that pinned copy,
so after a main-only column rename
+ // the current metadata can disagree with the schema the branch
statement binds to.
+ Optional<IcebergWriteSchemaContext> pinned =
IcebergStatementScope.activeWriteSchema(
+ session, iceHandle.getDbName(), tableName);
+ Schema schema;
+ PartitionSpec spec;
+ if (pinned.isPresent()) {
+ schema = pinned.get().getSchema();
+ spec = pinned.get().getPartitionSpec();
+ } else {
+ Table table = loadTable(session, iceHandle);
+ schema = table.schema();
+ spec = table.spec();
+ }
if (!spec.isPartitioned()) {
throw new DorisConnectorException(String.format(
"Table %s is not partitioned, cannot use static partition
syntax", tableName));
}
- Map<String, PartitionField> partitionFieldMap = new HashMap<>();
- for (PartitionField field : spec.fields()) {
- partitionFieldMap.put(field.name(), field);
- }
for (String colName : staticPartitionColumnNames) {
- PartitionField field = partitionFieldMap.get(colName);
- if (field == null) {
+ validateStaticPartitionColumn(schema, spec, colName, tableName);
+ }
+ }
+
+ /** The single-name rule of {@link #validateStaticPartitionColumns}. */
+ private static void validateStaticPartitionColumn(Schema schema,
PartitionSpec spec, String colName,
+ String tableName) {
+ // The identity partition over this top-level column is the target.
Searching by SOURCE first (across all
+ // fields, not the first source match) keeps a spec that carries a
non-identity field on the same column
+ // legal - e.g. bucket(8, id) declared before identity(id) - and never
lets a nested leaf or a case-folded
+ // field alias stand in for the column the binder will materialize.
+ if (IcebergPartitionUtils.identityPartitionBySourceColumn(schema,
spec, colName) != null) {
+ return;
+ }
+ PartitionField bySource =
IcebergPartitionUtils.partitionFieldBySourceColumn(schema, spec, colName);
+ if (bySource != null) {
+ throw nonIdentityStaticPartition(colName, bySource);
+ }
+ PartitionField byName = partitionFieldByName(spec, colName);
+ if (byName != null) {
+ if (byName.transform().isIdentity()) {
+ // Renamed identity field (ADD PARTITION KEY ... AS ...):
point at the source column the engine can
+ // materialize, instead of accepting a name that no row column
carries.
+ String source =
IcebergPartitionUtils.topLevelSourceColumnName(schema, byName.sourceId());
+ if (source == null) {
+ throw new DorisConnectorException(String.format(
+ "Static partition column '%s' in table '%s' cannot
be written by value: its source is"
+ + " a nested field, and the engine can
only materialize the literal into a"
+ + " top-level column.", byName.name(),
tableName));
+ }
throw new DorisConnectorException(String.format(
- "Unknown partition column '%s' in table '%s'.
Available partition columns: %s",
- colName, tableName, partitionFieldMap.keySet()));
+ "Static partition column '%s' in table '%s' is a
partition field name, not a table"
+ + " column; write the identity partition by
its source column '%s' instead.",
+ byName.name(), tableName, source));
}
- if (!field.transform().isIdentity()) {
- throw new DorisConnectorException(String.format(
- "Cannot use static partition syntax for non-identity
partition field '%s'"
- + " (transform: %s).", colName,
field.transform().toString()));
+ throw nonIdentityStaticPartition(colName, byName);
+ }
+ throw new DorisConnectorException(String.format(
+ "Unknown partition column '%s' in table '%s'. Available
partition columns: %s",
+ colName, tableName,
IcebergPartitionUtils.identityPartitionSourceColumns(schema, spec)));
+ }
+
+ private static DorisConnectorException nonIdentityStaticPartition(String
colName, PartitionField field) {
+ return new DorisConnectorException(String.format(
+ "Cannot use static partition syntax for non-identity partition
field '%s'"
+ + " (transform: %s).", colName,
field.transform().toString()));
+ }
+
+ private static PartitionField partitionFieldByName(PartitionSpec spec,
String name) {
+ for (PartitionField field : spec.fields()) {
+ if (field.name().equalsIgnoreCase(name)) {
+ return field;
}
}
+ return null;
}
// ========== Predicate pushdown ==========
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionUtils.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionUtils.java
index 24094ea76da..d9035efb111 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionUtils.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionUtils.java
@@ -35,6 +35,7 @@ import org.apache.iceberg.PartitionData;
import org.apache.iceberg.PartitionField;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Partitioning;
+import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.Table;
@@ -138,6 +139,104 @@ final class IcebergPartitionUtils {
return new ArrayList<>(partitionColumns);
}
+ /**
+ * Rewrites a static-partition spec ({@code INSERT [OVERWRITE] ...
PARTITION (col = val)}) so every key is the
+ * matching partition FIELD name, the identifier both downstream consumers
address: the BE sink matches
+ * {@code static_partition_values} against {@code PartitionField#name()},
and the commit row filter resolves
+ * the field to its source column.
+ *
+ * <p>The engine's PARTITION clause names the identity partition's SOURCE
column instead — the identifier Spark
+ * resolves for a static partition clause ({@code
Analyzer.partitionColumnNames} takes the identity transform's
+ * field reference) and the one {@code SHOW CREATE TABLE} renders — so the
two spellings must be bridged once,
+ * here, before either consumer sees the spec. A key that matches no field
of the current spec is kept verbatim
+ * and left to the callers' fail-loud guards.</p>
+ */
+ static Map<String, String> toPartitionFieldKeys(Map<String, String>
staticPartitions, Schema schema,
+ PartitionSpec spec) {
+ if (staticPartitions == null || staticPartitions.isEmpty() || spec ==
null) {
+ return staticPartitions;
+ }
+ Map<String, String> canonical = new HashMap<>(staticPartitions.size());
+ for (Map.Entry<String, String> entry : staticPartitions.entrySet()) {
+ canonical.put(toPartitionFieldName(schema, spec, entry.getKey()),
entry.getValue());
+ }
+ return canonical;
+ }
+
+ /**
+ * The partition field a static-partition name addresses: the IDENTITY
field over that top-level source column
+ * first, then a field with that name. The source lookup must win — a
field may carry a name that collides with
+ * another column's name in a different case ({@code identity(other,
"ID")} next to {@code identity(id,
+ * "id_identity")} is legal, because {@code PartitionSpec.Builder} checks
names case-SENSITIVELY), and the
+ * engine's name is the source column it validated.
+ */
+ private static String toPartitionFieldName(Schema schema, PartitionSpec
spec, String name) {
+ PartitionField identity = identityPartitionBySourceColumn(schema,
spec, name);
+ if (identity != null) {
+ return identity.name();
+ }
+ for (PartitionField field : spec.fields()) {
+ if (field.name().equalsIgnoreCase(name)) {
+ return field.name();
+ }
+ }
+ return name;
+ }
+
+ /**
+ * The top-level column a partition field's source resolves to, or {@code
null} when the source is a nested
+ * field. {@link NestedField#name()} is only the LEAF name, so a nested
{@code payload.id} source must not be
+ * mistaken for the top-level column {@code id}: the engine can only
materialize a top-level column.
+ */
+ static String topLevelSourceColumnName(Schema schema, int sourceId) {
+ for (NestedField column : schema.columns()) {
+ if (column.fieldId() == sourceId) {
+ return column.name();
+ }
+ }
+ return null;
+ }
+
+ /** The spec's <em>identity</em> partition over the top-level column
{@code name}, or null. */
+ static PartitionField identityPartitionBySourceColumn(Schema schema,
PartitionSpec spec, String name) {
+ for (PartitionField field : spec.fields()) {
+ if (!field.transform().isIdentity()) {
+ continue;
+ }
+ String source = topLevelSourceColumnName(schema, field.sourceId());
+ if (source != null && source.equalsIgnoreCase(name)) {
+ return field;
+ }
+ }
+ return null;
+ }
+
+ /** Any partition field over the top-level column {@code name}, or null
(used to report a non-identity one). */
+ static PartitionField partitionFieldBySourceColumn(Schema schema,
PartitionSpec spec, String name) {
+ for (PartitionField field : spec.fields()) {
+ String source = topLevelSourceColumnName(schema, field.sourceId());
+ if (source != null && source.equalsIgnoreCase(name)) {
+ return field;
+ }
+ }
+ return null;
+ }
+
+ /** The top-level source columns of the spec's identity partitions, in
field order. */
+ static List<String> identityPartitionSourceColumns(Schema schema,
PartitionSpec spec) {
+ List<String> names = new ArrayList<>();
+ for (PartitionField field : spec.fields()) {
+ if (!field.transform().isIdentity()) {
+ continue;
+ }
+ String source = topLevelSourceColumnName(schema, field.sourceId());
+ if (source != null) {
+ names.add(source);
+ }
+ }
+ return names;
+ }
+
/**
* Whether <b>any</b> partition spec of the table is partitioned. Scan
planning must use this instead of
* the current default spec: after evolving to an unpartitioned spec, data
files written under an older
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 57f2fa1e338..5c8d0dc7ca3 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
@@ -805,19 +805,35 @@ public class IcebergWritePlanProvider implements
ConnectorWritePlanProvider {
// Branch-targeted INSERT (INSERT INTO tbl@branch): the branch is
threaded from the generic insert
// command context onto the write handle; beginWrite validates it
against the table refs and points
// the commit at the branch. Empty for a default-ref write.
- return new IcebergWriteContext(op, handle.isOverwrite(),
staticPartitionSpec(handle),
+ return new IcebergWriteContext(op, handle.isOverwrite(),
staticPartitionSpec(handle, schemaContext),
handle.getBranchName(), readSnapshotId, readSnapshotResolved,
schemaContext, handle.getBoundWriteMetadataIdentity());
}
- private static Map<String, String>
staticPartitionSpec(ConnectorWriteHandle handle) {
+ /**
+ * The static-partition spec ({@code INSERT [OVERWRITE] ... PARTITION (col
= val)}) in the shape both consumers
+ * need: partition FIELD names, with SQL NULL folded in.
+ *
+ * <p>The engine keys the spec by the identity partition's SOURCE column
(what Spark resolves and what
+ * {@code SHOW CREATE TABLE} renders), while the BE sink matches {@code
static_partition_values} against
+ * {@code PartitionField#name()} and the commit row filter resolves that
field back to its source column. The
+ * rewrite happens here, once, so the wire contract and the commit filter
stay field-keyed — a source-column
+ * key reaching either of them would silently make the write dynamic on BE
and fail the commit validation.</p>
+ */
+ private static Map<String, String>
staticPartitionSpec(ConnectorWriteHandle handle,
+ IcebergWriteSchemaContext schemaContext) {
Map<String, String> spec = handle.getStaticPartitionSpec();
- if (handle.getStaticPartitionNullKeys().isEmpty()) {
- return spec;
+ Map<String, String> normalized = spec;
+ if (!handle.getStaticPartitionNullKeys().isEmpty()) {
+ // The engine carries SQL NULL separately; both commit validation
and BE must see the same null keys.
+ normalized = new HashMap<>(spec);
+ normalized.replaceAll((key, value) ->
+ handle.getStaticPartitionNullKeys().contains(key) ? null :
value);
+ }
+ if (schemaContext != null) {
+ normalized =
IcebergPartitionUtils.toPartitionFieldKeys(normalized,
schemaContext.getSchema(),
+ schemaContext.getPartitionSpec());
}
- // The engine carries SQL NULL separately; both commit validation and
BE must see the same null keys.
- Map<String, String> normalized = new HashMap<>(spec);
- normalized.replaceAll((key, value) ->
handle.getStaticPartitionNullKeys().contains(key) ? null : value);
return normalized;
}
@@ -872,7 +888,7 @@ public class IcebergWritePlanProvider implements
ConnectorWritePlanProvider {
// Overwrite + static partition values (INSERT OVERWRITE ...
PARTITION).
tSink.setOverwrite(handle.isOverwrite());
- Map<String, String> staticPartitionSpec = staticPartitionSpec(handle);
+ Map<String, String> staticPartitionSpec = staticPartitionSpec(handle,
schemaContext);
if (handle.isOverwrite() && staticPartitionSpec != null &&
!staticPartitionSpec.isEmpty()) {
Map<String, String> values = new java.util.HashMap<>();
java.util.Set<String> nullKeys = new java.util.HashSet<>();
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataTest.java
index 6d3efdbfe42..a8975d561ad 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataTest.java
@@ -19,6 +19,7 @@ package org.apache.doris.connector.iceberg;
import org.apache.doris.connector.spi.ConnectorColumn;
import org.apache.doris.connector.spi.ConnectorDatabaseMetadata;
+import org.apache.doris.connector.spi.ConnectorSession;
import org.apache.doris.connector.spi.ConnectorTableSchema;
import org.apache.doris.connector.spi.ConnectorViewDefinition;
import org.apache.doris.connector.spi.DorisConnectorException;
@@ -32,6 +33,7 @@ import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.RowLevelOperationMode;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
@@ -315,21 +317,157 @@ public class IcebergConnectorMetadataTest {
@Test
public void validateStaticPartitionColumnsRejectsUnknownColumn() {
- // WHY: a PARTITION(col=..) naming a column that is not a partition
FIELD must fail loud at analysis time
- // (byte-identical to legacy validateStaticPartition), else the
unknown column is silently swallowed by the
- // sink's materialize block and surfaces as an unrelated planning
error ("Cannot find snapshot"). The lookup
- // is keyed by partition FIELD name, so "id" (the bucket's SOURCE
column, not a field) is also "unknown" —
- // mirroring regression test_iceberg_static_partition_overwrite TC29
PARTITION(category) over bucket(category).
- // MUTATION: dropping the containsKey/null check -> unknown column
admitted -> red.
+ // WHY: a PARTITION(col=..) naming nothing the spec can write must
fail loud at analysis time, else the name
+ // is silently swallowed by the sink's materialize block and surfaces
as an unrelated planning error
+ // ("Cannot find snapshot"). The available list is the identity
partitions' SOURCE columns — the names a
+ // static write can actually use — not the raw field-name set.
+ // MUTATION: dropping the last-resort throw -> unknown column admitted
-> red.
Schema schema = idNameSchema();
IcebergConnectorMetadata md = metadataWithSpec(schema,
PartitionSpec.builderFor(schema).identity("name").bucket("id",
8).build());
- for (String bad : new String[] {"invalid_col", "id"}) {
- DorisConnectorException e =
Assertions.assertThrows(DorisConnectorException.class,
- () -> md.validateStaticPartitionColumns(
- null, new IcebergTableHandle("db1", "t1"),
Collections.singletonList(bad)));
- Assertions.assertTrue(e.getMessage().contains("Unknown partition
column"), e.getMessage());
- }
+ DorisConnectorException e =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> md.validateStaticPartitionColumns(
+ null, new IcebergTableHandle("db1", "t1"),
Collections.singletonList("invalid_col")));
+ Assertions.assertTrue(e.getMessage().contains("Unknown partition
column"), e.getMessage());
+ Assertions.assertTrue(e.getMessage().contains("[name]"),
e.getMessage());
+ }
+
+ @Test
+ public void
validateStaticPartitionColumnsRejectsBucketSourceColumnAsNonIdentity() {
+ // WHY: "id" feeds bucket(8, id), so it cannot be written statically —
but it is a real column, and it is the
+ // name a user reads off SHOW CREATE TABLE (`BUCKET(8, `id`)`).
Reporting it as an unknown COLUMN would send
+ // them looking for a spelling mistake; the non-identity reason is the
actionable one (Spark rejects the same
+ // name as "not a partition column"). Upstream #65991 case 5.5 / TC29
PARTITION(category) over bucket(category).
+ // MUTATION: matching only identity fields and skipping this branch ->
wrong message -> red.
+ Schema schema = idNameSchema();
+ IcebergConnectorMetadata md = metadataWithSpec(schema,
+ PartitionSpec.builderFor(schema).identity("name").bucket("id",
8).build());
+ DorisConnectorException e =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> md.validateStaticPartitionColumns(
+ null, new IcebergTableHandle("db1", "t1"),
Collections.singletonList("id")));
+ Assertions.assertTrue(
+ e.getMessage().contains("Cannot use static partition syntax
for non-identity partition field 'id'"),
+ e.getMessage());
+ }
+
+ @Test
+ public void
validateStaticPartitionColumnsUsesTheIdentitySourceColumnNotTheRenamedField() {
+ // WHY: ADD PARTITION KEY id AS id_identity renames the partition
FIELD away from its source column. A static
+ // partition value must then be written through the SOURCE column
(Spark's contract: partitionColumnNames
+ // takes the identity transform's field reference, and the static
value is materialized into that column),
+ // while the field name is not a table column at all — accepting it is
what produced the binder's
+ // "Column has no default value, column=id". MUTATION: keying the
lookup by field name -> 'id' reads as
+ // unknown (red) and 'id_identity' is admitted even though no row
column can carry it (red).
+ Schema schema = idNameSchema();
+ IcebergConnectorMetadata md = metadataWithSpec(schema,
+ PartitionSpec.builderFor(schema).identity("id",
"id_identity").build());
+ Assertions.assertDoesNotThrow(() -> md.validateStaticPartitionColumns(
+ null, new IcebergTableHandle("db1", "t1"),
Collections.singletonList("id")));
+ // Doris folds identifiers case-insensitively, so the same name in any
case must resolve.
+ Assertions.assertDoesNotThrow(() -> md.validateStaticPartitionColumns(
+ null, new IcebergTableHandle("db1", "t1"),
Collections.singletonList("ID")));
+ DorisConnectorException e =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> md.validateStaticPartitionColumns(
+ null, new IcebergTableHandle("db1", "t1"),
Collections.singletonList("id_identity")));
+ Assertions.assertTrue(e.getMessage().contains("is a partition field
name, not a table column"),
+ e.getMessage());
+ Assertions.assertTrue(e.getMessage().contains("source column 'id'"),
e.getMessage());
+ }
+
+ @Test
+ public void
validateStaticPartitionColumnsPrefersTheIdentitySourceOverACaseFoldedAlias() {
+ // PartitionSpec.Builder checks field names case-SENSITIVELY, so a
field may be named "ID" over source
+ // "other" next to identity(id) named "id_identity". The static name
"id" addresses the SOURCE column, so
+ // it must resolve to that identity field and never to the earlier
alias. MUTATION: matching field names
+ // before sources -> "id" resolves to the field over "other" -> red.
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(2, "other",
Types.IntegerType.get()));
+ PartitionSpec spec = PartitionSpec.builderFor(schema)
+ .identity("other", "ID")
+ .identity("id", "id_identity")
+ .build();
+ IcebergConnectorMetadata md = metadataWithSpec(schema, spec);
+
+ Assertions.assertDoesNotThrow(() -> md.validateStaticPartitionColumns(
+ null, new IcebergTableHandle("db1", "t1"),
Collections.singletonList("id")));
+ }
+
+ @Test
+ public void
validateStaticPartitionColumnsFindsIdentityAfterANonIdentityFieldOnTheSameSource()
{
+ // A legal spec may declare bucket(8, id) before identity(id); static
writes address the identity field, so
+ // the identity must be found regardless of field order. MUTATION:
failing on the first source match ->
+ // a writable table is rejected as non-identity -> red.
+ Schema schema = idNameSchema();
+ PartitionSpec spec = PartitionSpec.builderFor(schema).bucket("id",
8).identity("id").build();
+ IcebergConnectorMetadata md = metadataWithSpec(schema, spec);
+
+ Assertions.assertDoesNotThrow(() -> md.validateStaticPartitionColumns(
+ null, new IcebergTableHandle("db1", "t1"),
Collections.singletonList("id")));
+ }
+
+ @Test
+ public void validateStaticPartitionColumnsRejectsNestedPartitionSources() {
+ // NestedField#name() is only the LEAF name: a nested identity source
must never be reachable through the
+ // top-level column sharing that leaf name, because the binder
materializes the literal into top-level
+ // columns only - accepting it would commit a file whose nested
partition value came from another column.
+ // MUTATION: matching the source by leaf name -> "id" is accepted ->
red.
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(2, "payload", Types.StructType.of(
+ Types.NestedField.required(3, "id",
Types.IntegerType.get()))));
+ PartitionSpec spec =
PartitionSpec.builderFor(schema).identity("payload.id", "nested_id").build();
+ IcebergConnectorMetadata md = metadataWithSpec(schema, spec);
+
+ DorisConnectorException leafName =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> md.validateStaticPartitionColumns(
+ null, new IcebergTableHandle("db1", "t1"),
Collections.singletonList("id")));
+ Assertions.assertTrue(leafName.getMessage().contains("Unknown
partition column 'id'"),
+ leafName.getMessage());
+ DorisConnectorException byFieldName =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> md.validateStaticPartitionColumns(
+ null, new IcebergTableHandle("db1", "t1"),
Collections.singletonList("nested_id")));
+ Assertions.assertTrue(byFieldName.getMessage().contains("source is a
nested field"),
+ byFieldName.getMessage());
+ }
+
+ @Test
+ public void
validateStaticPartitionColumnsUsesThePinnedBranchWriterSchema() {
+ // INSERT INTO t@branch(old) pins the BRANCH's writer schema before
analysis completes, and binding,
+ // planning and the BE sink all consume that pinned copy. After a
main-only rename (id -> key) the current
+ // metadata no longer has column "id" while the branch still does, so
validating against the current schema
+ // would reject the branch statement's own partition column and advise
a name that branch does not carry.
+ // MUTATION: validating against table.schema() -> the pinned assertion
turns red (the unpinned one below
+ // pins the divergence that makes the pin observable).
+ Schema branchSchema = idNameSchema();
+ InMemoryCatalog catalog = new InMemoryCatalog();
+ catalog.initialize("pinned", Collections.emptyMap());
+ catalog.createNamespace(Namespace.of("db1"));
+ Table branchTable = catalog.createTable(TableIdentifier.of("db1",
"t1"), branchSchema,
+ PartitionSpec.builderFor(branchSchema).identity("id").build());
+ IcebergWriteSchemaContext pinned = IcebergWriteSchemaContext.create(
+ branchTable, "t1", Optional.empty(), false, false);
+ TestStatementScope scope = new TestStatementScope();
+ ConnectorSession session = new
IcebergStatementScopeTest.ScopeSession(1L, "q1", scope);
+ IcebergStatementScope.writeSchema(session, "db1", "t1",
Optional.empty(), () -> pinned);
+
+ Schema renamedSchema = new Schema(
+ Types.NestedField.required(1, "key", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "name", Types.StringType.get()));
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ ops.table = new FakeIcebergTable("t1", renamedSchema,
+ PartitionSpec.builderFor(renamedSchema).identity("key",
"id").build(),
+ "s3://bucket/db1/t1", Collections.emptyMap());
+ IcebergConnectorMetadata md = new IcebergConnectorMetadata(ops,
+ IcebergCatalogProperties.of(Collections.emptyMap()), new
RecordingConnectorContext());
+
+ Assertions.assertDoesNotThrow(() -> md.validateStaticPartitionColumns(
+ session, new IcebergTableHandle("db1", "t1"),
Collections.singletonList("id")));
+ DorisConnectorException currentOnly =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> md.validateStaticPartitionColumns(
+ null, new IcebergTableHandle("db1", "t1"),
Collections.singletonList("id")));
+ Assertions.assertTrue(currentOnly.getMessage().contains("source column
'key'"),
+ currentOnly.getMessage());
}
@Test
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 1b6d83d1dd3..6489c1d317e 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
@@ -961,6 +961,41 @@ public class IcebergConnectorTransactionTest {
Assertions.assertEquals("1", snap.summary().get("added-data-files"));
}
+ @Test
+ public void
overwriteStaticPartitionFiltersOnTheSourceColumnOfARenamedIdentityField() {
+ // ADD PARTITION KEY id AS id_identity: the write planner hands the
commit the partition FIELD key
+ // ("id_identity"), and the row filter must still be built on the
SOURCE column ("id") — the column the
+ // identity value is derived from, and what Spark's static-partition
overwrite filters on. The retained
+ // partition proves the filter did not widen (a field-name expression
would fail to bind instead).
+ // MUTATION: evaluating the filter on the field name -> iceberg binder
rejects "id_identity" -> red.
+ 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, "value",
Types.StringType.get()));
+ PartitionSpec spec = PartitionSpec.builderFor(schema).identity("id",
"id_identity").build();
+ Table table = catalog.createTable(id, schema, spec,
props("write.format.default", "parquet"));
+ table.newAppend().appendFile(partitionedDataFile(spec,
+ "s3://b/db1/t1/id_identity=1/seed.parquet", 1L,
"id_identity=1")).commit();
+
+ IcebergConnectorTransaction txn =
txnFor(opsReturning(catalog.loadTable(id)),
+ new RecordingConnectorContext());
+ txn.beginWrite(SESSION, "db1", "t1",
+ overwriteStaticCtx(table,
Collections.singletonMap("id_identity", "2")));
+ txn.addCommitData(commitBytes(dataFileItem(
+ "s3://b/db1/t1/id_identity=2/f1.parquet", 1L, 1024L,
Collections.singletonList("2"))));
+ txn.commit();
+
+ List<DataFile> remaining = currentDataFiles(catalog.loadTable(id));
+ Assertions.assertEquals(2, remaining.size(), remaining.toString());
+ Assertions.assertTrue(remaining.stream().anyMatch(
+ f -> f.path().toString().contains("id_identity=1")),
+ "the untouched partition must survive: " + remaining);
+ Assertions.assertTrue(remaining.stream().anyMatch(
+ f -> f.path().toString().contains("id_identity=2")),
+ "the overwritten partition must be added: " + remaining);
+ }
+
@Test
public void promotedDateStaticOverwriteRejectsUnsafeRetainedBounds() {
InMemoryCatalog catalog = freshCatalog();
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergPartitionUtilsTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergPartitionUtilsTest.java
index cb0bd8d2aee..a4cba8bd550 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergPartitionUtilsTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergPartitionUtilsTest.java
@@ -291,6 +291,76 @@ public class IcebergPartitionUtilsTest {
Assertions.assertEquals(Arrays.asList("payload", "int_col"), cols);
}
+ // ---- toPartitionFieldKeys ----
+
+ @Test
+ public void staticPartitionKeysResolveTheSourceColumnToThePartitionField()
{
+ // ADD PARTITION KEY id AS id_identity: the engine's PARTITION clause
names the SOURCE column "id", while
+ // the BE sink matches static_partition_values against the partition
FIELD name and the commit row filter
+ // resolves that field to its source column. Both consumers therefore
need the field-keyed spec.
+ // MUTATION: passing the engine's spelling through -> BE sees no
static column (write silently becomes
+ // dynamic) and the commit filter stops matching -> red.
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "region",
Types.StringType.get()));
+ PartitionSpec spec = PartitionSpec.builderFor(schema)
+ .identity("id", "id_identity")
+ .identity("region")
+ .build();
+
+ Map<String, String> canonical =
IcebergPartitionUtils.toPartitionFieldKeys(
+ new HashMap<>(Collections.singletonMap("id", "7")), schema,
spec);
+ Assertions.assertEquals(Collections.singletonMap("id_identity", "7"),
canonical);
+
+ // A field name (the un-renamed spelling, and the bucket-style field
name a caller may already hold) is
+ // canonicalized to itself, and the lookup folds case like every Doris
identifier.
+ Map<String, String> mixed = new HashMap<>();
+ mixed.put("REGION", "us");
+ mixed.put("ID_IDENTITY", "8");
+ Map<String, String> expected = new HashMap<>();
+ expected.put("region", "us");
+ expected.put("id_identity", "8");
+ Assertions.assertEquals(expected,
IcebergPartitionUtils.toPartitionFieldKeys(mixed, schema, spec));
+
+ // A name the current spec does not know is kept verbatim: the callers
own the fail-loud guard for it.
+ Map<String, String> unknown =
IcebergPartitionUtils.toPartitionFieldKeys(
+ new HashMap<>(Collections.singletonMap("ghost", "1")), schema,
spec);
+ Assertions.assertEquals(Collections.singletonMap("ghost", "1"),
unknown);
+ }
+
+ @Test
+ public void
staticPartitionKeysPreferTheIdentitySourceOverACaseFoldedFieldAlias() {
+ // A partition field may carry a name that collides with another
column's name in a different case
+ // (`identity(other, "ID")` beside `identity(id, "id_identity")`),
because the spec builder checks names
+ // case-SENSITIVELY. The engine's PARTITION name is the SOURCE column,
so the key must follow the identity
+ // field over that source rather than the earlier alias.
+ // MUTATION: matching field names first -> the key becomes "ID"
(source "other") -> BE stamps and the commit
+ // filters the WRONG partition -> red.
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(2, "other",
Types.IntegerType.get()));
+ PartitionSpec spec = PartitionSpec.builderFor(schema)
+ .identity("other", "ID")
+ .identity("id", "id_identity")
+ .build();
+
+ Assertions.assertEquals(Collections.singletonMap("id_identity", "7"),
+ IcebergPartitionUtils.toPartitionFieldKeys(
+ new HashMap<>(Collections.singletonMap("id", "7")),
schema, spec));
+ }
+
+ @Test
+ public void
staticPartitionKeysFindTheIdentityFieldRegardlessOfFieldOrder() {
+ // bucket(8, id) may be declared before identity(id); the static
target is still the identity field.
+ // MUTATION: taking the first source match -> the bucket field's name
is returned -> red.
+ Schema schema = new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()));
+ PartitionSpec spec = PartitionSpec.builderFor(schema).bucket("id",
8).identity("id").build();
+
+ Assertions.assertEquals(Collections.singletonMap("id", "7"),
+ IcebergPartitionUtils.toPartitionFieldKeys(
+ new HashMap<>(Collections.singletonMap("id", "7")),
schema, spec));
+ }
+
// ---- getIdentityPartitionInfoMap ----
@Test
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 8ab5566dc80..0518bc94e52 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
@@ -1071,6 +1071,27 @@ public class IcebergWritePlanProviderTest {
"INSERT OVERWRITE ... PARTITION must pass the static partition
values to BE");
}
+ @Test
+ public void planWriteStaticPartitionKeysTheRenamedPartitionField() {
+ // ADD PARTITION KEY id AS id_identity: the engine keys the spec by
the SOURCE column (the identifier Spark
+ // resolves for a static PARTITION clause, and the one an identity
value can be materialized through), while
+ // the BE writer matches static_partition_values against the partition
FIELD name. Shipping "id" unchanged
+ // would make BE treat the write as dynamic, so the planner must
rewrite the key to the field name.
+ // MUTATION: dropping the rewrite -> {"id"="7"} != {"id_identity"="7"}
-> red.
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "value",
Types.StringType.get()));
+ Table table = freshCatalog().createTable(TableIdentifier.of("db1",
"t1"), schema,
+ PartitionSpec.builderFor(schema).identity("id",
"id_identity").build(),
+ Collections.singletonMap("write.format.default", "parquet"));
+
+ TIcebergTableSink sink = planSink(table, contextWithStorage(),
+ new WriteHandle(new IcebergTableHandle("db1",
"t1")).overwrite(true)
+ .writeContext(Collections.singletonMap("id", "7")));
+
+ Assertions.assertEquals(Collections.singletonMap("id_identity", "7"),
sink.getStaticPartitionValues());
+ }
+
@Test
public void planWriteStaticPartitionNullMarkerReachesSink() {
Table table = partitionedSortedTable(freshCatalog());
diff --git
a/regression-test/data/external_table_p0/iceberg/write/test_iceberg_static_partition_overwrite.out
b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_static_partition_overwrite.out
index b600e2b3950..ef9e4289eef 100644
---
a/regression-test/data/external_table_p0/iceberg/write/test_iceberg_static_partition_overwrite.out
+++
b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_static_partition_overwrite.out
@@ -215,3 +215,6 @@
11 Frank 1 85.5 false B
4 David 2 90.0 true A
+-- !renamed_static_source_column --
+1 a 1
+2 b2 22
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_static_partition_overwrite.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_static_partition_overwrite.groovy
index f8636c23433..c251d6d5447 100644
---
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_static_partition_overwrite.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_static_partition_overwrite.groovy
@@ -1097,7 +1097,7 @@ suite("test_iceberg_static_partition_overwrite",
"p0,external") {
PARTITION (category='food')
SELECT 10, 'Eve'
"""
- exception "Unknown partition column"
+ exception "Cannot use static partition syntax for non-identity
partition field"
}
// Using correct partition field name should trigger non-identity error
test {
@@ -1130,7 +1130,7 @@ suite("test_iceberg_static_partition_overwrite",
"p0,external") {
PARTITION (description='hello')
SELECT 10, 'Eve'
"""
- exception "Unknown partition column"
+ exception "Cannot use static partition syntax for non-identity
partition field"
}
// Using correct partition field name should trigger non-identity error
test {
@@ -1163,7 +1163,7 @@ suite("test_iceberg_static_partition_overwrite",
"p0,external") {
PARTITION (ts='2025-01-25 10:00:00')
SELECT 10, 'Eve'
"""
- exception "Unknown partition column"
+ exception "Cannot use static partition syntax for non-identity
partition field"
}
// Using correct partition field name should trigger non-identity error
test {
@@ -1196,7 +1196,7 @@ suite("test_iceberg_static_partition_overwrite",
"p0,external") {
PARTITION (event_date='2024-06-15')
SELECT 10, 'Eve'
"""
- exception "Unknown partition column"
+ exception "Cannot use static partition syntax for non-identity
partition field"
}
// Using correct partition field name should trigger non-identity error
test {
@@ -1229,7 +1229,7 @@ suite("test_iceberg_static_partition_overwrite",
"p0,external") {
PARTITION (event_date='2025-01-15')
SELECT 10, 'Eve'
"""
- exception "Unknown partition column"
+ exception "Cannot use static partition syntax for non-identity
partition field"
}
// Using correct partition field name should trigger non-identity error
test {
@@ -1262,7 +1262,7 @@ suite("test_iceberg_static_partition_overwrite",
"p0,external") {
PARTITION (ts='2025-01-25 10:00:00')
SELECT 10, 'Eve'
"""
- exception "Unknown partition column"
+ exception "Cannot use static partition syntax for non-identity
partition field"
}
// Using correct partition field name should trigger non-identity error
test {
@@ -1297,7 +1297,7 @@ suite("test_iceberg_static_partition_overwrite",
"p0,external") {
PARTITION (id=1)
SELECT 'Eve', 'bj'
"""
- exception "Unknown partition column"
+ exception "Cannot use static partition syntax for non-identity
partition field"
}
// Using correct partition field name should trigger non-identity error
test {
@@ -1330,7 +1330,7 @@ suite("test_iceberg_static_partition_overwrite",
"p0,external") {
PARTITION (amount=100)
SELECT 10, 'Eve'
"""
- exception "Unknown partition column"
+ exception "Cannot use static partition syntax for non-identity
partition field"
}
// Using correct partition field name should trigger non-identity error
test {
@@ -1363,7 +1363,7 @@ suite("test_iceberg_static_partition_overwrite",
"p0,external") {
PARTITION (user_id=1001)
SELECT 10, 'Eve'
"""
- exception "Unknown partition column"
+ exception "Cannot use static partition syntax for non-identity
partition field"
}
// Using correct partition field name should trigger non-identity error
test {
@@ -1404,7 +1404,7 @@ suite("test_iceberg_static_partition_overwrite",
"p0,external") {
PARTITION (ts='2025-01-25 10:00:00')
SELECT 10, 'Eve', 'bj'
"""
- exception "Unknown partition column"
+ exception "Cannot use static partition syntax for non-identity
partition field"
}
// Using correct partition field name should trigger non-identity error
test {
@@ -1415,4 +1415,46 @@ suite("test_iceberg_static_partition_overwrite",
"p0,external") {
"""
exception "Cannot use static partition syntax for non-identity
partition field"
}
+
+ //
============================================================================
+ // Test Case 39: An identity partition field renamed away from its source
column
+ //
============================================================================
+
+ // ADD PARTITION KEY id AS id_identity keeps the identity transform on
column id but names the partition
+ // field id_identity. Such a partition is written by its SOURCE column -
the identifier Spark resolves for a
+ // static PARTITION clause and the one SHOW CREATE TABLE renders - while
the field name is not a table column.
+ String tb_renamed = db1 + "_renamed_identity"
+ sql """ DROP TABLE IF EXISTS ${tb_renamed} """
+ sql """
+ CREATE TABLE ${tb_renamed} (
+ id BIGINT,
+ value STRING,
+ metric INT
+ ) ENGINE=iceberg
+ """
+ sql """ ALTER TABLE ${tb_renamed} ADD PARTITION KEY id AS id_identity """
+ sql """ INSERT INTO ${tb_renamed} VALUES (1, 'a', 1), (2, 'b', 2) """
+
+ // Only the id=2 partition is replaced, and the literal is materialized
into the row's id column (an identity
+ // partition value must equal the column value, otherwise the data file
contradicts its own partition).
+ sql """
+ INSERT OVERWRITE TABLE ${tb_renamed}
+ PARTITION (id = 2) (value, metric) VALUES ('b2', 22)
+ """
+ order_qt_renamed_static_source_column """ SELECT id, value, metric FROM
${tb_renamed} ORDER BY id """
+ def renamedRows = sql("SELECT id, value, metric FROM ${tb_renamed} ORDER
BY id")
+ spark_iceberg "REFRESH TABLE demo.${db1}.${tb_renamed}"
+ assertSparkDorisResultEquals(spark_iceberg("""
+ SELECT id, value, metric FROM demo.${db1}.${tb_renamed} ORDER BY id
+ """), renamedRows)
+
+ // The partition FIELD name is not a table column: report it with the
source column to use instead, rather than
+ // the previous "PARTITION names a column the binder then cannot find"
contradiction.
+ test {
+ sql """
+ INSERT OVERWRITE TABLE ${tb_renamed}
+ PARTITION (id_identity = 2) (value, metric) VALUES ('b3', 33)
+ """
+ exception "is a partition field name, not a table column"
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
