github-actions[bot] commented on code in PR #69188:
URL: https://github.com/apache/doris/pull/69188#discussion_r4292649786
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionUtils.java:
##########
@@ -138,6 +139,46 @@ static List<String> getIdentityPartitionColumns(Table
table) {
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: its SOURCE
column name, or the field name itself.
+ * Iceberg refuses to name a partition field after a column other than its
own source
+ * ({@code PartitionSpec.Builder#checkAndAddPartitionName}), so the two
lookups cannot disagree.
+ */
+ private static String toPartitionFieldName(Schema schema, PartitionSpec
spec, String name) {
+ for (PartitionField field : spec.fields()) {
+ NestedField source = schema.findField(field.sourceId());
+ if ((source != null && source.name().equalsIgnoreCase(name))
Review Comment:
[P1] Resolve source columns before matching partition-field aliases
The two lookups can disagree because [Iceberg's
builder](https://github.com/apache/iceberg/blob/apache-iceberg-1.12.0/api/src/main/java/org/apache/iceberg/PartitionSpec.java#L358-L402)
checks names case-sensitively by default. With schema `(id INT, other INT)`
and spec `identity(other, "ID"), identity(id, "id_identity")`,
`PARTITION(id=7)` validates against source `id`, but this loop returns the
earlier field `ID`, whose source is `other`. An overwrite supplying `other=9`
then writes row `(id=7, other=9)` while BE stamps `other`'s partition as 7; the
commit also filters `other=7` instead of `id=7`. This can overwrite the wrong
partitions and commit files whose rows contradict their partition metadata.
Resolve the validated identity source first, and only consider field aliases as
an unambiguous fallback.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java:
##########
@@ -1911,23 +1919,60 @@ public void
validateStaticPartitionColumns(ConnectorSession session, ConnectorTa
throw new DorisConnectorException(String.format(
"Table %s is not partitioned, cannot use static partition
syntax", tableName));
}
- Map<String, PartitionField> partitionFieldMap = new HashMap<>();
+ for (String colName : staticPartitionColumnNames) {
+ validateStaticPartitionColumn(table.schema(), spec, colName,
tableName);
+ }
+ }
+
+ /** The single-name rule of {@link #validateStaticPartitionColumns}. */
+ private static void validateStaticPartitionColumn(Schema schema,
PartitionSpec spec, String colName,
+ String tableName) {
for (PartitionField field : spec.fields()) {
- partitionFieldMap.put(field.name(), field);
+ Types.NestedField source = schema.findField(field.sourceId());
+ if (source == null || !source.name().equalsIgnoreCase(colName)) {
Review Comment:
[P1] Keep nested source fields distinct from top-level columns
`NestedField.name()` is only the leaf name. For a supported table `(id INT,
payload STRUCT<id:INT>)` partitioned by `identity(payload.id)` as `nested_id`,
this accepts `PARTITION(id=2)` even though the top-level `id` is not a
partition source. `BindSink` materializes the literal into that top-level
column, while `toPartitionFieldKeys` rewrites it to `nested_id`. An overwrite
supplying `payload.id=9` therefore writes that value into the file but stamps
its nested partition as 2 through BE's static override. Even an initially empty
table can commit this inconsistent file; the overwrite path does not validate
added files against the filter. Resolve the source's full path/field identity
and reject nested static targets unless the binder can materialize that same
source.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java:
##########
@@ -1911,23 +1919,60 @@ public void
validateStaticPartitionColumns(ConnectorSession session, ConnectorTa
throw new DorisConnectorException(String.format(
"Table %s is not partitioned, cannot use static partition
syntax", tableName));
}
- Map<String, PartitionField> partitionFieldMap = new HashMap<>();
+ for (String colName : staticPartitionColumnNames) {
+ validateStaticPartitionColumn(table.schema(), spec, colName,
tableName);
+ }
+ }
+
+ /** The single-name rule of {@link #validateStaticPartitionColumns}. */
+ private static void validateStaticPartitionColumn(Schema schema,
PartitionSpec spec, String colName,
+ String tableName) {
for (PartitionField field : spec.fields()) {
- partitionFieldMap.put(field.name(), field);
+ Types.NestedField source = schema.findField(field.sourceId());
+ if (source == null || !source.name().equalsIgnoreCase(colName)) {
+ continue;
+ }
+ // The SOURCE column of a partition field: a legal static target
only for an identity transform,
+ // because a transformed partition value is not the column value
and cannot be written through it.
+ if (field.transform().isIdentity()) {
+ return;
+ }
+ throw new DorisConnectorException(String.format(
Review Comment:
[P2] Search all matching transforms before rejecting the source
A valid Iceberg spec can contain `bucket("id", 8)` before `identity("id")`.
`PARTITION(id=7)` previously resolved the identity field named `id`, but the
new loop stops at the bucket field with the same source and throws, never
reaching the identity field. This breaks static INSERT and OVERWRITE on such
tables solely because of spec field order. Look for a matching identity source
before reporting a non-identity error, and make `toPartitionFieldName` select
that same identity field rather than the first source match.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java:
##########
@@ -1911,23 +1919,60 @@ public void
validateStaticPartitionColumns(ConnectorSession session, ConnectorTa
throw new DorisConnectorException(String.format(
"Table %s is not partitioned, cannot use static partition
syntax", tableName));
}
- Map<String, PartitionField> partitionFieldMap = new HashMap<>();
+ for (String colName : staticPartitionColumnNames) {
+ validateStaticPartitionColumn(table.schema(), spec, colName,
tableName);
Review Comment:
[P2] Validate static names against the pinned branch writer schema
Create an identity-partitioned `t(id INT, value STRING)`, create branch
`old` from a snapshot, then rename main's source column `id` to `key`. `INSERT
INTO t@branch(old) PARTITION(id=2) (value) VALUES('x')` still binds to the
branch's old `id` schema, but this call validates against current
`table.schema()`: it misses source `key` and rejects partition field `id`,
advising the user to use `key`, which is absent from that branch. The previous
field-keyed validation accepted this ordinary INSERT, and the existing writer
materializes and appends it using the pinned branch schema. Use the already
active `IcebergWriteSchemaContext` for validation so it agrees with binding and
planning, and cover a branch static INSERT after a main-only rename.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
