github-actions[bot] commented on code in PR #66345:
URL: https://github.com/apache/doris/pull/66345#discussion_r3696025151
##########
fe/fe-core/src/main/java/org/apache/doris/planner/PluginDrivenTableSink.java:
##########
@@ -121,12 +124,27 @@ public PluginDrivenTableSink(PluginDrivenExternalTable
targetTable,
ConnectorWritePlanProvider writePlanProvider, ConnectorSession
connectorSession,
ConnectorTableHandle tableHandle, List<ConnectorColumn>
connectorColumns,
TSortInfo writeSortInfo, WriteOperation writeOperation, boolean
requireMergeCardinalityCheck) {
+ this(targetTable, writePlanProvider, connectorSession, tableHandle,
connectorColumns,
+ connectorColumns, writeSortInfo, writeOperation,
requireMergeCardinalityCheck);
Review Comment:
[P1] Keep the row-level projection and validation baseline on one schema
generation. UPDATE and MERGE each call `getBaseSchema(true)` once while
building `[operation,row_id,data...]` and again for `sink.getCols()`. With
Iceberg table-cache TTL 0, a reorder can land between them; this overload then
aliases the later list as `boundTargetColumns`, so S1 validates against S1
while BE writes the S0-ordered expressions positionally under S1. A same-shaped
`FIXED`/`CHAR` reorder silently swaps values. Please capture the schema once
and thread that exact list through projection, sink, and validation, with a
consecutive-read drift test.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalConnectorTableSink.java:
##########
@@ -46,6 +48,8 @@
*/
public class PhysicalConnectorTableSink<CHILD_TYPE extends Plan> extends
PhysicalBaseExternalTableSink<CHILD_TYPE> {
+ private final List<Column> boundTargetSchema;
Review Comment:
[P1] Resolve positional consumers from this bound schema.
`getRequirePhysicalProperties()` still reloads Hive/MaxCompute partition names
and full-schema positions, then indexes the older child output; the Iceberg
write-sort path likewise returns live ordinals and indexes that output before
schema validation. A refresh or reorder can therefore hash/sort the wrong
expression or throw out of bounds. Please carry stable partition/sort
identities with this generation and derive every output position from it.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java:
##########
@@ -230,6 +232,43 @@ public ConnectorSinkPlan planWrite(ConnectorSession
session, ConnectorWriteHandl
}
}
+ private void validateBoundWriteColumns(Table table, ConnectorWriteHandle
handle,
+ WriteOperation writeOperation) {
+ // V3 row-lineage columns are engine-generated metadata, not fields in
table.schema(). Excluding
+ // their neutral marker keeps the comparison on the complete user
schema for INSERT and MERGE.
+ List<ConnectorColumn> boundColumns =
handle.getBoundTargetColumns().stream()
+ .filter(column -> !column.isReservedPassthrough())
+ .collect(Collectors.toList());
+ if (writeOperation == WriteOperation.DELETE || boundColumns.isEmpty())
{
+ return;
+ }
+ List<NestedField> currentColumns = table.schema().columns();
+ boolean hasSyntheticRowId = boundColumns.size() ==
currentColumns.size() + 1
+ &&
DORIS_ICEBERG_ROWID_COL.equals(boundColumns.get(boundColumns.size() -
1).getName());
+ if (boundColumns.size() != currentColumns.size() &&
!hasSyntheticRowId) {
+ throw new DorisConnectorException("Iceberg table schema changed
after the write was bound; retry the "
+ + "statement with the latest schema");
+ }
+ boolean enableVarbinary = Boolean.parseBoolean(properties.getOrDefault(
+ IcebergConnectorProperties.ENABLE_MAPPING_VARBINARY, "false"));
+ boolean enableTimestampTz =
Boolean.parseBoolean(properties.getOrDefault(
+ IcebergConnectorProperties.ENABLE_MAPPING_TIMESTAMP_TZ,
"false"));
+ for (int i = 0; i < currentColumns.size(); ++i) {
+ NestedField current = currentColumns.get(i);
+ ConnectorColumn bound = boundColumns.get(i);
+ ConnectorType currentType = IcebergTypeMapping.fromIcebergType(
+ current.type(), enableVarbinary, enableTimestampTz);
+ if (!current.name().equalsIgnoreCase(bound.getName())
+ || !currentType.equals(bound.getType())
Review Comment:
[P1] Compare recursive Iceberg field identities. The bound conversion has
already lost every nested `Column.uniqueId`, and `ConnectorType.equals`
intentionally excludes child IDs and optionality. For example, a `FIXED(n)`
struct child can be dropped and re-added with the same name/type/position, pass
this check as the same `CHAR(n)` shape, and have the old bound child payload
written under the replacement field ID from the live schema JSON. Please carry
and compare recursive IDs/requiredness (or an authoritative schema generation)
and add a nested recreate test.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java:
##########
@@ -726,13 +725,33 @@ public PlanFragment visitPhysicalConnectorTableSink(
// an instanceof Iceberg. Ordinary connector INSERTs keep
WriteOperation.INSERT (byte-identical).
WriteOperation writeOperation = connectorTableSink.isRewrite()
? WriteOperation.REWRITE : WriteOperation.INSERT;
+ // The write list can omit explicit/static-partition columns, but
schema-drift validation must
+ // retain the complete generation captured by BindSink instead of
comparing that subset.
PluginDrivenTableSink providerSink = new
PluginDrivenTableSink(targetTable,
- writePlanProvider, connSession, providerTableHandle,
connectorColumns, writeSortInfo,
- writeOperation);
+ writePlanProvider, connSession, providerTableHandle,
connectorColumns,
+ boundTargetColumns, writeSortInfo, writeOperation, false);
rootFragment.setSink(providerSink);
return rootFragment;
}
+ private static ConnectorColumn toWriteConnectorColumn(Column column) {
+ ConnectorColumn result = new ConnectorColumn(column.getName(),
+ ConnectorColumnConverter.toConnectorType(column.getType()),
Review Comment:
[P1] Normalize the bound type before comparing it. A real unchanged Iceberg
`INTEGER` arrives as `ConnectorType("INT", -1, -1)`, becomes `ScalarType.INT`
with `0/0`, and is converted here to `ConnectorType("INT", 0, 0)`;
`ConnectorType.equals` compares those parameters, so ordinary
INSERT/UPDATE/MERGE/REWRITE is rejected as "schema changed". Timestamps and
several other scalars diverge too. The added provider tests construct canonical
`ConnectorType.of("INT")` directly and miss this path; please compare one
canonical representation and add a translator-to-provider test.
--
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]
