github-actions[bot] commented on code in PR #66345:
URL: https://github.com/apache/doris/pull/66345#discussion_r3701866818
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTable.java:
##########
@@ -735,6 +735,39 @@ private List<Column>
appendSyntheticWriteColumns(List<Column> schema) {
return result;
}
+ /** Immutable write-facing views captured from one schema-cache
generation. */
+ public static final class WriteSchemaSnapshot {
+ private final List<Column> fullSchema;
+ private final List<Column> partitionColumns;
+
+ private WriteSchemaSnapshot(List<Column> fullSchema, List<Column>
partitionColumns) {
+ this.fullSchema = Collections.unmodifiableList(new
ArrayList<>(fullSchema));
+ this.partitionColumns = Collections.unmodifiableList(new
ArrayList<>(partitionColumns));
+ }
+
+ public List<Column> getFullSchema() {
+ return fullSchema;
+ }
+
+ public List<Column> getPartitionColumns() {
+ return partitionColumns;
+ }
+ }
+
+ /**
+ * Captures schema and partition identities from one cache value for write
binding. Reading them through
+ * separate table APIs can straddle a concurrent refresh and make the
planner hash an older output by a
+ * newer column ordinal.
+ */
+ public WriteSchemaSnapshot getWriteSchemaSnapshot() {
Review Comment:
[P1] Bind the remote table identity together with this write snapshot. If
Iceberg table U0 is dropped and recreated as U1 after `BindSink` captures these
columns but before `getWriteMetadataIdentity()` performs the first
`sharedWritableTable` load, every later step sees U1. With identical field
IDs/schema/defaults, format, sort order, and partition spec, all current
comparisons pass and the INSERT that was bound to U0 commits into U1. Iceberg's
operations-level UUID refresh check only helps after that object has already
loaded U0, so please carry the bind-time table UUID/generation in this
snapshot/write handle and compare it at begin (with a
drop/recreate-before-first-load regression test).
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java:
##########
@@ -230,6 +234,121 @@ public ConnectorSinkPlan planWrite(ConnectorSession
session, ConnectorWriteHandl
}
}
+ private void validateBoundWriteColumns(Table table, ConnectorWriteHandle
handle,
+ WriteOperation writeOperation) {
+ List<ConnectorColumn> boundTargetColumns =
handle.getBoundTargetColumns();
+ if (!boundTargetColumns.isEmpty()
+ && (writeOperation == WriteOperation.REWRITE
+ || writeOperation == WriteOperation.UPDATE
+ || writeOperation == WriteOperation.MERGE)) {
+ long boundLineageColumns = boundTargetColumns.stream()
+ .filter(ConnectorColumn::isReservedPassthrough)
+ .count();
+ long currentLineageColumns =
IcebergWriterHelper.getFormatVersion(table) >= 3 ? 2 : 0;
+ // Format version changes the physical sink arity without changing
table.schema(); the reserved
+ // columns are the bind-time witness that the output and v3
schema-json belong to one generation.
+ if (boundLineageColumns != currentLineageColumns) {
+ throw new DorisConnectorException(
+ "Iceberg write metadata changed after the write was
bound; retry the statement");
+ }
+ }
+ // 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 = boundTargetColumns.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);
+ // Do not compare top-level nullability: Doris widens Iceberg
required columns in its read schema
+ // so evolution default-fill may yield NULL. Nested requiredness
remains authoritative in
+ // sameBoundType, while current schema JSON enforces writes at the
root.
+ if (!current.name().equalsIgnoreCase(bound.getName())
Review Comment:
[P1] Compare the bound write default here as part of the generation fence.
Doris has already expanded omitted columns (and explicit `DEFAULT`) from the
bind-time `ConnectorColumn.getDefaultValue()`. If another writer changes an
Iceberg field's `writeDefault` from, say, 42 to 7 before `beginWrite()`
refreshes the table, the name/type/field-id and format/sort/spec checks all
still pass, so this statement writes 42 into the refreshed table instead of
retrying against the current default. `parseSchema()` deliberately exposes
`NestedField.writeDefault()` for this materialization, so it needs to be
included in this comparison (with a regression test for default-only schema
evolution between bind and begin).
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -607,6 +632,12 @@ private List<ConnectorScanRange> planScanInternal(
Optional<ConnectorExpression> filter,
boolean countPushdown) {
IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
+ if (iceHandle.isResolvedEmptySnapshot()) {
Review Comment:
[P1] Please do not apply the empty-data-snapshot shortcut to every system
table. `getSysTableHandle()` preserves the resolved-empty flag, so a newly
created table now returns no ranges even for `$metadata_log_entries`. Iceberg
builds that table from the metadata history and always includes the current
metadata file, so it has a creation row before any data snapshot exists. Keep
the empty MVCC fence for the snapshot-dependent data/file tables, but let
snapshot-independent metadata tables plan normally (and add a resolved-empty
system-table 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]
