github-actions[bot] commented on code in PR #65851:
URL: https://github.com/apache/doris/pull/65851#discussion_r3673466013
##########
be/src/format/transformer/vorc_transformer.cpp:
##########
@@ -595,6 +640,162 @@ Status VOrcTransformer::write(const Block& block) {
return Status::OK();
}
+static Status normalize_iceberg_nullable_column(const ColumnPtr& column, const
DataTypePtr& type,
+ const iceberg::NestedField&
nested_field,
+ ColumnPtr* normalized_column,
+ const NullMap* skipped_rows) {
+ const auto& nullable_column = assert_cast<const ColumnNullable&>(*column);
+ const auto& null_map = nullable_column.get_null_map_data();
+ NullMap combined_null_map;
+ const NullMap* combined_skipped_rows = &null_map;
+ if (skipped_rows != nullptr) {
+ combined_null_map.resize(null_map.size());
+ for (size_t row = 0; row < null_map.size(); ++row) {
+ combined_null_map[row] = null_map[row] | (*skipped_rows)[row];
+ }
+ combined_skipped_rows = &combined_null_map;
+ }
+ ColumnPtr nested_column;
+
RETURN_IF_ERROR(normalize_iceberg_binary_column(nullable_column.get_nested_column_ptr(),
+ remove_nullable(type),
nested_field,
+ &nested_column,
combined_skipped_rows));
+ *normalized_column = ColumnNullable::create(
+ IColumn::mutate(std::move(nested_column)),
+
IColumn::mutate(nullable_column.get_null_map_column_ptr()->clone()));
+ return Status::OK();
+}
+
+static Status normalize_iceberg_uuid_column(const ColumnPtr& column,
ColumnPtr* normalized_column,
+ const NullMap* skipped_rows) {
+ DORIS_CHECK(check_and_get_column<ColumnString>(*column) != nullptr ||
+ check_and_get_column<ColumnVarbinary>(*column) != nullptr);
+ auto binary_column = column->clone_empty();
+ binary_column->reserve(column->size());
+ for (size_t row = 0; row < column->size(); ++row) {
+ std::array<uint8_t, 16> bytes;
+ if (skipped_rows == nullptr || (*skipped_rows)[row] == 0) {
+
RETURN_IF_ERROR(parse_iceberg_uuid_to_bytes(column->get_data_at(row), &bytes));
+ } else {
+ bytes.fill(0);
+ }
+ binary_column->insert_data(reinterpret_cast<const
char*>(bytes.data()), bytes.size());
+ }
+ *normalized_column = std::move(binary_column);
+ return Status::OK();
+}
+
+static Status normalize_iceberg_fixed_column(const ColumnPtr& column,
+ const iceberg::NestedField&
nested_field,
+ ColumnPtr* normalized_column,
+ const NullMap* skipped_rows) {
+ const auto expected_length = cast_set<size_t>(
+ assert_cast<const
iceberg::FixedType*>(nested_field.field_type())->get_length());
+ for (size_t row = 0; row < column->size(); ++row) {
+ if (skipped_rows != nullptr && (*skipped_rows)[row] != 0) {
+ continue;
+ }
+ const auto value = column->get_data_at(row);
+ if (value.size != expected_length) {
+ return Status::InvalidArgument("Iceberg FIXED[{}] ORC value has {}
bytes at row {}",
+ expected_length, value.size, row);
+ }
+ }
+ *normalized_column = column;
+ return Status::OK();
+}
+
+static Status normalize_iceberg_struct_column(const ColumnPtr& column, const
DataTypePtr& type,
+ const iceberg::NestedField&
nested_field,
+ ColumnPtr* normalized_column,
+ const NullMap* skipped_rows) {
+ const auto& struct_column = assert_cast<const ColumnStruct&>(*column);
+ const auto& struct_type = assert_cast<const DataTypeStruct&>(*type);
+ const auto& fields = nested_field.field_type()->as_struct_type()->fields();
+ DORIS_CHECK(struct_column.tuple_size() == fields.size());
+ Columns children;
+ children.reserve(fields.size());
+ for (size_t index = 0; index < fields.size(); ++index) {
+ ColumnPtr child;
+
RETURN_IF_ERROR(normalize_iceberg_binary_column(struct_column.get_column_ptr(index),
+
struct_type.get_element(index),
+ fields[index], &child,
skipped_rows));
+ children.push_back(std::move(child));
+ }
+ *normalized_column = ColumnStruct::create(std::move(children));
+ return Status::OK();
+}
+
+static Status normalize_iceberg_array_column(const ColumnPtr& column, const
DataTypePtr& type,
+ const iceberg::NestedField&
nested_field,
+ ColumnPtr* normalized_column) {
+ const auto& array_column = assert_cast<const ColumnArray&>(*column);
+ const auto& array_type = assert_cast<const DataTypeArray&>(*type);
+ ColumnPtr elements;
+ RETURN_IF_ERROR(normalize_iceberg_binary_column(
Review Comment:
[P1] Preserve nullable collection masks during ORC normalization
`skipped_rows` reaches this array through
`normalize_iceberg_nullable_column`, but is dropped here before recursion into
the flattened elements. Doris can retain materialized child values under NULL
parents: `IF(cond, non_nullable_array, NULL)` wraps the full array with a null
map. An optional `ARRAY<UUID>` row can therefore be NULL while a hidden element
contains invalid legacy UUID text; normalization still parses it and rejects a
semantically valid write. Please expand each skipped parent row across the
collection offsets and pass the element-level mask through ARRAY and MAP child
normalization, including both map keys and values, and test nullable collection
parents that hide invalid UUID/FIXED values.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergWriteSchemaContext.java:
##########
@@ -0,0 +1,715 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.iceberg;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.datasource.mvcc.MvccSnapshot;
+import org.apache.doris.datasource.mvcc.MvccUtil;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Array;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.CreateMap;
+import
org.apache.doris.nereids.trees.expressions.functions.scalar.CreateNamedStruct;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Unhex;
+import org.apache.doris.nereids.trees.expressions.literal.ArrayLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.BooleanLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.DateTimeV2Literal;
+import org.apache.doris.nereids.trees.expressions.literal.DateV2Literal;
+import org.apache.doris.nereids.trees.expressions.literal.DecimalV3Literal;
+import org.apache.doris.nereids.trees.expressions.literal.DoubleLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.FloatLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.literal.MapLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.StringLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.StructLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.TimestampTzLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.VarBinaryLiteral;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.nereids.types.DateTimeV2Type;
+import org.apache.doris.nereids.types.DecimalV3Type;
+import org.apache.doris.nereids.types.StructType;
+import org.apache.doris.nereids.types.TimeStampTzType;
+import org.apache.doris.nereids.types.VarBinaryType;
+import org.apache.doris.nereids.util.TypeCoercionUtils;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.io.BaseEncoding;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.PartitionSpecParser;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SchemaParser;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.SortField;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.SortOrderParser;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.SnapshotUtil;
+
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.UUID;
+
+/**
+ * Statement-scoped Iceberg write schema and write-default values.
+ *
+ * <p>The context pins one Iceberg schema before analysis. The analyzer,
planner sink and
+ * transaction preflight must all use this same instance so a concurrent
schema change cannot
+ * combine expressions from one schema with a writer schema from another one.
+ */
+public final class IcebergWriteSchemaContext {
+ private final long tableId;
+ private final String tableName;
+ private final Schema schema;
+ private final int formatVersion;
+ private final Optional<String> branchName;
+ private final String schemaJson;
+ private final Schema mergeSchema;
+ private final String mergeSchemaJson;
+ private final PartitionSpec partitionSpec;
+ private final String partitionSpecJson;
+ private final SortOrder sortOrder;
+ private final String sortOrderJson;
+ private final FileFormat fileFormat;
+ private final MetricsConfig metricsConfig;
+ private final String fileCompression;
+ private final String dataLocation;
+ private final Map<String, String> writerProperties;
+ private final List<Column> columns;
+ private final List<Column> mergeColumns;
+ private final Map<Integer, Types.NestedField> fieldsById;
+ private final Map<Integer, Expression> writeDefaultsById;
+
+ /** Pin the statement snapshot's current table schema under the catalog
authentication boundary. */
+ public static IcebergWriteSchemaContext create(
+ IcebergExternalTable dorisTable, Optional<String> branchName) {
+ Objects.requireNonNull(dorisTable, "dorisTable should not be null");
+ Objects.requireNonNull(branchName, "branchName should not be null");
+ try {
+ return
dorisTable.getCatalog().getExecutionAuthenticator().execute(() -> {
+ Table table = dorisTable.getIcebergTable();
+ Schema schema = branchName.isPresent()
+ ? resolveBranchSchema(table, branchName.get(),
dorisTable.getName())
+ : resolveStatementSchema(table, dorisTable);
+ if (branchName.isPresent()) {
+ validateBranchWriterSchema(
+ schema, table.schema(), branchName.get(),
dorisTable.getName());
+ }
+ int formatVersion = IcebergUtils.getFormatVersion(table);
+ Map<String, String> properties =
ImmutableMap.copyOf(table.properties());
+ return new IcebergWriteSchemaContext(
+ dorisTable.getId(), dorisTable.getName(), schema,
formatVersion, branchName,
+ bindPartitionSpec(table.spec(), schema,
dorisTable.getName()),
+ bindSortOrder(table.sortOrder(), schema,
dorisTable.getName()),
+ IcebergUtils.getFileFormat(table),
MetricsConfig.forTable(table),
+ IcebergUtils.getFileCompress(table),
IcebergUtils.dataLocation(table), properties,
+ dorisTable.getCatalog().getEnableMappingVarbinary(),
+ dorisTable.getCatalog().getEnableMappingTimestampTz());
+ });
+ } catch (Exception e) {
+ throw new AnalysisException("Failed to pin Iceberg write schema
for table "
+ + dorisTable.getName() + ": " + e.getMessage(), e);
+ }
+ }
+
+ @VisibleForTesting
+ public static IcebergWriteSchemaContext forSchema(Schema schema, int
formatVersion,
+ boolean enableMappingVarbinary, boolean enableMappingTimestampTz) {
+ return new IcebergWriteSchemaContext(-1L, "test_table", schema,
formatVersion,
+ Optional.empty(), PartitionSpec.unpartitioned(),
SortOrder.unsorted(),
+ FileFormat.PARQUET, MetricsConfig.getDefault(),
+ TableProperties.PARQUET_COMPRESSION_DEFAULT_SINCE_1_4_0,
+ "file:///tmp/test_table/data", ImmutableMap.of(),
+ enableMappingVarbinary, enableMappingTimestampTz);
+ }
+
+ @VisibleForTesting
+ public static IcebergWriteSchemaContext forSchema(Schema schema, int
formatVersion,
+ PartitionSpec partitionSpec, SortOrder sortOrder, FileFormat
fileFormat,
+ MetricsConfig metricsConfig, String fileCompression, String
dataLocation,
+ Map<String, String> writerProperties,
+ boolean enableMappingVarbinary, boolean enableMappingTimestampTz) {
+ return new IcebergWriteSchemaContext(-1L, "test_table", schema,
formatVersion,
+ Optional.empty(), partitionSpec, sortOrder, fileFormat,
metricsConfig,
+ fileCompression, dataLocation, writerProperties,
+ enableMappingVarbinary, enableMappingTimestampTz);
+ }
+
+ private IcebergWriteSchemaContext(long tableId, String tableName, Schema
schema,
+ int formatVersion, Optional<String> branchName,
+ PartitionSpec partitionSpec, SortOrder sortOrder, FileFormat
fileFormat,
+ MetricsConfig metricsConfig, String fileCompression, String
dataLocation,
+ Map<String, String> writerProperties,
+ boolean enableMappingVarbinary, boolean enableMappingTimestampTz) {
+ this.tableId = tableId;
+ this.tableName = Objects.requireNonNull(tableName, "tableName should
not be null");
+ this.schema = Objects.requireNonNull(schema, "schema should not be
null");
+ this.formatVersion = formatVersion;
+ this.branchName = Objects.requireNonNull(branchName, "branchName
should not be null");
+ this.schemaJson = SchemaParser.toJson(schema);
+ this.mergeSchema = formatVersion >=
IcebergUtils.ICEBERG_ROW_LINEAGE_MIN_VERSION
+ ? IcebergUtils.appendRowLineageFieldsForV3(schema) : schema;
+ this.mergeSchemaJson = SchemaParser.toJson(mergeSchema);
+ this.partitionSpec = Objects.requireNonNull(partitionSpec,
"partitionSpec should not be null");
+ this.partitionSpecJson = PartitionSpecParser.toJson(partitionSpec);
+ this.sortOrder = Objects.requireNonNull(sortOrder, "sortOrder should
not be null");
+ this.sortOrderJson = SortOrderParser.toJson(sortOrder);
+ this.fileFormat = Objects.requireNonNull(fileFormat, "fileFormat
should not be null");
+ this.metricsConfig = Objects.requireNonNull(metricsConfig,
"metricsConfig should not be null");
+ this.fileCompression = Objects.requireNonNull(
+ fileCompression, "fileCompression should not be null");
+ this.dataLocation = Objects.requireNonNull(dataLocation, "dataLocation
should not be null");
+ this.writerProperties = ImmutableMap.copyOf(
+ Objects.requireNonNull(writerProperties, "writerProperties
should not be null"));
+ validateWriterMetadataSources(schema, partitionSpec, sortOrder,
tableName);
+
+ List<Column> parsedColumns = IcebergUtils.parseSchema(
+ schema, enableMappingVarbinary, enableMappingTimestampTz);
+ this.columns = ImmutableList.copyOf(parsedColumns);
+ List<Column> writerColumns = new ArrayList<>(parsedColumns);
+ writerColumns.add(IcebergRowId.createHiddenColumn());
+ if (formatVersion >= IcebergUtils.ICEBERG_ROW_LINEAGE_MIN_VERSION) {
+ Column rowIdColumn = IcebergUtils.parseField(
+ org.apache.iceberg.MetadataColumns.ROW_ID,
+ enableMappingVarbinary, enableMappingTimestampTz);
+ rowIdColumn.setIsVisible(false);
+ writerColumns.add(rowIdColumn);
+ Column sequenceColumn = IcebergUtils.parseField(
+
org.apache.iceberg.MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER,
+ enableMappingVarbinary, enableMappingTimestampTz);
+ sequenceColumn.setIsVisible(false);
+ writerColumns.add(sequenceColumn);
+ }
+ this.mergeColumns = ImmutableList.copyOf(writerColumns);
+
+ ImmutableMap.Builder<Integer, Types.NestedField> byId =
ImmutableMap.builder();
+ ImmutableMap.Builder<Integer, Expression> defaults =
ImmutableMap.builder();
+ for (Types.NestedField field : schema.columns()) {
+ byId.put(field.fieldId(), field);
+ if (field.writeDefault() != null) {
+ DataType targetType =
DataType.fromCatalogType(IcebergUtils.icebergTypeToDorisType(
+ field.type(), enableMappingVarbinary,
enableMappingTimestampTz));
+ defaults.put(field.fieldId(), toDorisExpression(
+ field.type(), field.writeDefault(), targetType,
+ enableMappingVarbinary, enableMappingTimestampTz));
+ }
+ }
+ this.fieldsById = byId.build();
+ this.writeDefaultsById = defaults.build();
+ }
+
+ private static PartitionSpec bindPartitionSpec(
+ PartitionSpec partitionSpec, Schema schema, String tableName) {
+ if (!partitionSpec.isPartitioned()) {
+ return PartitionSpec.builderFor(schema)
+ .withSpecId(partitionSpec.specId())
+ .build();
+ }
+ try {
+ return PartitionSpecParser.fromJson(schema,
PartitionSpecParser.toJson(partitionSpec));
+ } catch (RuntimeException e) {
+ throw new AnalysisException("Iceberg partition spec " +
partitionSpec.specId()
+ + " is incompatible with pinned schema " +
schema.schemaId()
+ + " for table " + tableName + ": " + e.getMessage(), e);
+ }
+ }
+
+ private static SortOrder bindSortOrder(SortOrder sortOrder, Schema schema,
String tableName) {
+ if (!sortOrder.isSorted()) {
+ return SortOrder.unsorted();
+ }
+ try {
+ return SortOrderParser.fromJson(schema,
SortOrderParser.toJson(sortOrder));
+ } catch (RuntimeException e) {
+ throw new AnalysisException("Iceberg sort order " +
sortOrder.orderId()
+ + " is incompatible with pinned schema " +
schema.schemaId()
+ + " for table " + tableName + ": " + e.getMessage(), e);
+ }
+ }
+
+ private static void validateWriterMetadataSources(
+ Schema schema, PartitionSpec partitionSpec, SortOrder sortOrder,
String tableName) {
+ Map<Integer, Types.NestedField> topLevelFields =
schema.columns().stream()
+
.collect(ImmutableMap.toImmutableMap(Types.NestedField::fieldId, field ->
field));
+ for (PartitionField field : partitionSpec.fields()) {
+ if (!topLevelFields.containsKey(field.sourceId())) {
+ throw new AnalysisException("Iceberg partition field " +
field.fieldId()
+ + " references source field " + field.sourceId()
+ + " outside pinned top-level schema " +
schema.schemaId()
+ + " for table " + tableName);
+ }
+ }
+ for (SortField field : sortOrder.fields()) {
+ if (schema.findField(field.sourceId()) == null) {
+ throw new AnalysisException("Iceberg sort field references
source field "
+ + field.sourceId() + " outside pinned schema " +
schema.schemaId()
+ + " for table " + tableName);
+ }
+ }
+ }
+
+ private static Schema resolveBranchSchema(Table table, String branchName,
String tableName) {
+ SnapshotRef ref = table.refs().get(branchName);
+ if (ref == null) {
+ throw new AnalysisException(branchName + " is not founded in " +
tableName);
+ }
+ if (!ref.isBranch()) {
+ throw new AnalysisException(branchName
+ + " is a tag, not a branch. Tags cannot be targets for
producing snapshots");
+ }
+ return SnapshotUtil.schemaFor(table, ref.snapshotId());
+ }
+
+ private static Schema resolveStatementSchema(Table table,
IcebergExternalTable dorisTable) {
+ Optional<MvccSnapshot> snapshot =
MvccUtil.getSnapshotFromContext(dorisTable);
+ if (!snapshot.isPresent()) {
+ return table.schema();
+ }
+ Preconditions.checkState(snapshot.get() instanceof IcebergMvccSnapshot,
+ "Expected an Iceberg MVCC snapshot for table %s",
dorisTable.getName());
+ long schemaId = ((IcebergMvccSnapshot) snapshot.get())
+ .getSnapshotCacheValue().getSnapshot().getSchemaId();
+ Schema schema = table.schemas().get(Math.toIntExact(schemaId));
+ return Preconditions.checkNotNull(schema,
+ "Iceberg schema %s is not available in the statement table
metadata for %s",
+ schemaId, dorisTable.getName());
+ }
+
+ /**
+ * Reject branch writes whose files cannot satisfy the table-current
schema.
+ *
+ * <p>Iceberg resolves columns from the branch-head schema, but stamps the
new branch snapshot
+ * with the table-current schema. A current required field without an
initial default must
+ * therefore also be present and required in the pinned branch writer
schema.
+ */
+ private static void validateBranchWriterSchema(
+ Schema branchSchema, Schema currentSchema, String branchName,
String tableName) {
+ Map<Integer, Types.NestedField> branchFields =
+ TypeUtil.indexById(branchSchema.asStruct());
+ Map<Integer, Types.NestedField> currentFields =
+ TypeUtil.indexById(currentSchema.asStruct());
+ Map<Integer, Integer> currentParents =
+ TypeUtil.indexParents(currentSchema.asStruct());
+ for (Types.NestedField currentField : currentFields.values()) {
+ Types.NestedField branchField =
branchFields.get(currentField.fieldId());
+ if (branchField != null) {
+ if (currentField.isRequired() && currentField.initialDefault()
== null
+ && branchField.isOptional()) {
+ throw incompatibleBranchSchema(
+ branchSchema, currentSchema, branchName,
tableName, currentField);
+ }
+ continue;
+ }
+ Types.NestedField highestMissingField = currentField;
+ Integer parentId = currentParents.get(currentField.fieldId());
+ while (parentId != null && !branchFields.containsKey(parentId)) {
+ highestMissingField =
Preconditions.checkNotNull(currentFields.get(parentId),
+ "Iceberg parent field %s is absent from current
schema", parentId);
+ parentId = currentParents.get(parentId);
+ }
+ if (highestMissingField.isRequired()
+ && highestMissingField.initialDefault() == null) {
+ throw incompatibleBranchSchema(
+ branchSchema, currentSchema, branchName, tableName,
highestMissingField);
+ }
+ }
+ }
+
+ private static AnalysisException incompatibleBranchSchema(
+ Schema branchSchema, Schema currentSchema, String branchName,
String tableName,
+ Types.NestedField field) {
+ return new AnalysisException("Iceberg table current schema " +
currentSchema.schemaId()
+ + " cannot label files written with pinned branch " +
branchName + " schema "
+ + branchSchema.schemaId() + " for table " + tableName + ":
required field "
+ + field.name() + " (id " + field.fieldId()
+ + ") has no initial default; retry after updating the branch
schema");
+ }
+
+ /** Resolve a write default by the pinned target field name. */
+ public Expression resolveWriteDefault(String columnName) {
+ Column column = columns.stream()
+ .filter(targetColumn ->
targetColumn.getName().equalsIgnoreCase(columnName))
+ .findFirst()
+ .orElseThrow(() -> new AnalysisException(
+ "Cannot find column information for DEFAULT(" +
columnName + ")"));
+ return resolveWriteDefault(column);
+ }
+
+ /** Resolve the value used for an omitted column or an explicit DEFAULT. */
+ public Expression resolveWriteDefault(Column column) {
+ Types.NestedField field = fieldsById.get(column.getUniqueId());
+ if (field == null) {
+ throw new AnalysisException("Column " + column.getName()
+ + " is not present in pinned Iceberg schema " +
getSchemaId());
+ }
+ Expression writeDefault = writeDefaultsById.get(field.fieldId());
+ if (writeDefault != null) {
+ return writeDefault;
+ }
+ DataType targetType = DataType.fromCatalogType(column.getType());
+ if (field.isOptional()) {
+ return new NullLiteral(targetType);
+ }
+ throw new AnalysisException("Column has no write default and is
required, column=" + field.name());
+ }
+
+ /** Validate that the fresh table can commit files described by the pinned
writer metadata. */
+ public void validateCurrentSchema(Table table) {
+ validateCurrentSchema(table, false);
+ }
+
+ /**
+ * Validate that the fresh table can commit files described by the pinned
writer metadata.
+ *
+ * <p>Every overwrite additionally requires the pinned spec to remain
current because both
+ * dynamic replacement and static replacement semantics depend on whether
and how that spec is
+ * partitioned. Appends can safely write an older retained spec, so they
only require the pinned
+ * definition to remain available.
+ */
+ public void validateCurrentSchema(Table table, boolean
requireCurrentPartitionSpec) {
Review Comment:
[P1] Fence writes to the pinned Iceberg table identity
This validates schema/format/spec/order, but never verifies that the freshly
loaded table is the same Iceberg table instance analyzed by the statement. If
`db.t` is dropped and recreated with the same initial schema/spec/order IDs,
these checks pass even though Iceberg assigns the replacement a different table
UUID. The BE can then write with the predecessor context's pinned data location
and commit those file paths into the replacement table; the final retry wrapper
repeats this same incomplete check. Please pin `TableMetadata.uuid()` in the
context and compare it against the exact base metadata at preflight and commit
(with an appropriate v1 fallback), and cover INSERT, overwrite, and
UPDATE/MERGE across a drop/recreate.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -530,30 +601,901 @@ void enableCurrentIcebergScanSemantics() {
params.setIcebergScanSemanticsVersion(ICEBERG_SCAN_SEMANTICS_VERSION);
}
+ /**
+ * Build the schema metadata carrier used by both scanners and
equality-delete readers.
+ *
+ * <p>Batch-mode delete files are planned asynchronously after scan
parameters are sent to BE.
+ * The authenticated manifest preflight therefore supplies the live
equality field IDs before
+ * the schema carrier is serialized. Only historical fields referenced by
those delete files are
+ * added, so an unrelated dropped type cannot make an otherwise supported
scan fail.
+ */
+ @VisibleForTesting
+ List<NestedField> getSchemaFieldsForScan(
+ Schema scanSchema, Set<Integer> equalityDeleteFieldIds) throws
UserException {
+ List<NestedField> fields = new ArrayList<>(scanSchema.columns());
+ if (isSystemTable || equalityDeleteFieldIds.isEmpty()) {
+ return fields;
+ }
+
+ Set<Integer> missingFieldIds = new HashSet<>(equalityDeleteFieldIds);
+
missingFieldIds.removeAll(TypeUtil.indexById(scanSchema.asStruct()).keySet());
+ if (missingFieldIds.isEmpty()) {
+ return fields;
+ }
+
+ List<Schema> schemaHistory = getMetadataSchemaHistory();
+ // Schema IDs may be reused when evolution returns to an earlier
schema, while the metadata
+ // list may also contain schemas committed after a time-travel or
branch target. Follow the
+ // actual scan snapshot's parent chain first so the field definition
active on that lineage
+ // wins. Then use the complete metadata list as a fallback for
schema-only changes and
+ // expired ancestors. A fallback definition may come from a later
rename, so BE resolves an
+ // ID-less equality key through the target mapping first and the
delete file's original key
+ // name second. Initial-default and field identity remain bound to the
stable field ID.
+ Snapshot snapshot = createTableScan().snapshot();
+ while (snapshot != null) {
+ Integer schemaId = snapshot.schemaId();
+ if (schemaId != null) {
+ Schema historicalSchema = icebergTable.schemas().get(schemaId);
+ Preconditions.checkState(historicalSchema != null,
+ "Iceberg snapshot schema %s is absent from table
metadata", schemaId);
+ addHistoricalEqualityFields(fields, missingFieldIds,
historicalSchema);
+ }
+ Long parentId = snapshot.parentId();
+ snapshot = parentId == null ? null :
icebergTable.snapshot(parentId);
+ }
+ for (int index = schemaHistory.size() - 1; index >= 0; index--) {
+ addHistoricalEqualityFields(fields, missingFieldIds,
schemaHistory.get(index));
+ }
+ Preconditions.checkState(missingFieldIds.isEmpty(),
+ "Iceberg equality-delete fields are absent from schema
history: %s",
+ missingFieldIds);
+ return fields;
+ }
+
+ private List<Schema> getMetadataSchemaHistory() {
+ Preconditions.checkState(icebergTable instanceof HasTableOperations,
+ "Iceberg table does not expose metadata schema history: %s",
icebergTable.name());
+ return ((HasTableOperations)
icebergTable).operations().current().schemas();
+ }
+
+ /**
+ * Return only schemas that can describe files visible from the selected
target.
+ *
+ * <p>The query schema is included explicitly because a schema-only commit
does not create a
+ * snapshot. Other schemas are taken from the selected snapshot's parent
lineage and from
+ * cherry-picked source snapshots (including their ancestry), excluding
later main-branch and
+ * unrelated branch schemas from the rolling-upgrade fence. An empty
optional means snapshot
+ * expiration truncated any required lineage, so callers must
conservatively require current
+ * scan semantics.
+ */
+ @VisibleForTesting
+ Optional<List<Schema>> getRequiredFieldSchemaHistory(Schema scanSchema)
throws UserException {
+ List<Schema> schemas = new ArrayList<>();
+ Set<Integer> schemaIds = new HashSet<>();
+ schemas.add(scanSchema);
+ schemaIds.add(scanSchema.schemaId());
+
+ Snapshot selectedSnapshot = createTableScan().snapshot();
+ Deque<Snapshot> snapshots = new ArrayDeque<>();
+ if (selectedSnapshot != null) {
+ snapshots.add(selectedSnapshot);
+ }
+ Set<Long> visitedSnapshotIds = new HashSet<>();
+ while (!snapshots.isEmpty()) {
+ Snapshot snapshot = snapshots.removeFirst();
+ if (!visitedSnapshotIds.add(snapshot.snapshotId())) {
+ continue;
+ }
+ Integer schemaId = snapshot.schemaId();
+ if (schemaId != null && schemaIds.add(schemaId)) {
+ Schema lineageSchema = icebergTable.schemas().get(schemaId);
+ Preconditions.checkState(lineageSchema != null,
+ "Iceberg snapshot schema %s is absent from table
metadata", schemaId);
+ schemas.add(lineageSchema);
+ }
+ Long parentId = snapshot.parentId();
+ if (parentId != null) {
+ Snapshot parent = icebergTable.snapshot(parentId);
+ if (parent == null) {
+ return Optional.empty();
+ }
+ snapshots.addLast(parent);
+ }
+ String sourceSnapshotId =
+
snapshot.summary().get(SnapshotSummary.SOURCE_SNAPSHOT_ID_PROP);
+ if (sourceSnapshotId != null) {
+ Snapshot sourceSnapshot =
+
icebergTable.snapshot(Long.parseLong(sourceSnapshotId));
+ if (sourceSnapshot == null) {
+ return Optional.empty();
+ }
+ snapshots.addLast(sourceSnapshot);
+ }
+ }
+ return Optional.of(schemas);
+ }
+
+ private static void addHistoricalEqualityFields(List<NestedField> fields,
+ Set<Integer> missingFieldIds, Schema historicalSchema) {
+ Map<Integer, NestedField> historicalFields =
+ TypeUtil.indexById(historicalSchema.asStruct());
+ Set<Integer> selectedFieldIds = new HashSet<>();
+ for (Integer fieldId : missingFieldIds) {
+ NestedField field = historicalFields.get(fieldId);
+ if (field != null) {
+ Preconditions.checkState(field.type().isPrimitiveType(),
+ "Iceberg equality-delete field %s must be primitive",
fieldId);
+ selectedFieldIds.add(fieldId);
+ }
+ }
+ if (selectedFieldIds.isEmpty()) {
+ return;
+ }
+
+ Schema selectedSchema = TypeUtil.select(historicalSchema,
selectedFieldIds);
+ mergeHistoricalEqualityFields(fields, selectedSchema.columns());
+ missingFieldIds.removeAll(selectedFieldIds);
+ }
+
+ private static void mergeHistoricalEqualityFields(
+ List<NestedField> fields, List<NestedField> historicalFields) {
+ for (NestedField historicalField : historicalFields) {
+ int currentIndex = -1;
+ for (int index = 0; index < fields.size(); index++) {
+ if (fields.get(index).fieldId() == historicalField.fieldId()) {
+ currentIndex = index;
+ break;
+ }
+ }
+ if (currentIndex < 0) {
+ fields.add(historicalField);
+ continue;
+ }
+
+ NestedField currentField = fields.get(currentIndex);
+ Type mergedType = mergeHistoricalEqualityType(
+ currentField.type(), historicalField.type());
+ if (mergedType != currentField.type()) {
+ fields.set(currentIndex, Types.NestedField.from(currentField)
+ .ofType(mergedType)
+ .build());
+ }
+ }
+ }
+
+ private static Type mergeHistoricalEqualityType(Type currentType, Type
historicalType) {
+ Preconditions.checkState(currentType.typeId() ==
historicalType.typeId(),
+ "Iceberg equality-delete ancestor type changed from %s to %s",
+ historicalType, currentType);
+ switch (currentType.typeId()) {
+ case STRUCT:
+ List<NestedField> mergedFields =
+ new ArrayList<>(currentType.asStructType().fields());
+ mergeHistoricalEqualityFields(
+ mergedFields, historicalType.asStructType().fields());
+ if (mergedFields.equals(currentType.asStructType().fields())) {
+ return currentType;
+ }
+ return Types.StructType.of(mergedFields);
+ case LIST:
+ Types.ListType currentList = currentType.asListType();
+ Types.ListType historicalList = historicalType.asListType();
+ Preconditions.checkState(currentList.elementId() ==
historicalList.elementId(),
+ "Iceberg equality-delete list element id changed from
%s to %s",
+ historicalList.elementId(), currentList.elementId());
+ Type mergedElement = mergeHistoricalEqualityType(
+ currentList.elementType(),
historicalList.elementType());
+ if (mergedElement == currentList.elementType()) {
+ return currentType;
+ }
+ return currentList.isElementOptional()
+ ? Types.ListType.ofOptional(currentList.elementId(),
mergedElement)
+ : Types.ListType.ofRequired(currentList.elementId(),
mergedElement);
+ case MAP:
+ Types.MapType currentMap = currentType.asMapType();
+ Types.MapType historicalMap = historicalType.asMapType();
+ Preconditions.checkState(currentMap.keyId() ==
historicalMap.keyId()
+ && currentMap.valueId() ==
historicalMap.valueId(),
+ "Iceberg equality-delete map field ids changed from
(%s, %s) to (%s, %s)",
+ historicalMap.keyId(), historicalMap.valueId(),
+ currentMap.keyId(), currentMap.valueId());
+ Type mergedKey = mergeHistoricalEqualityType(
+ currentMap.keyType(), historicalMap.keyType());
+ Type mergedValue = mergeHistoricalEqualityType(
+ currentMap.valueType(), historicalMap.valueType());
+ if (mergedKey == currentMap.keyType()
+ && mergedValue == currentMap.valueType()) {
+ return currentType;
+ }
+ return currentMap.isValueOptional()
+ ? Types.MapType.ofOptional(
+ currentMap.keyId(), currentMap.valueId(),
+ mergedKey, mergedValue)
+ : Types.MapType.ofRequired(
+ currentMap.keyId(), currentMap.valueId(),
+ mergedKey, mergedValue);
+ default:
+ Preconditions.checkState(currentType.equals(historicalType),
+ "Iceberg equality-delete field type changed from %s to
%s",
+ historicalType, currentType);
+ return currentType;
+ }
+ }
+
+ @VisibleForTesting
+ static boolean requiresRecursiveInitialDefaultMaterialization(
+ Schema scanSchema, List<SlotDescriptor> projectedSlots) {
+ return requiresProjectedIcebergField(scanSchema, projectedSlots,
+ (field, isTopLevel) -> field.initialDefault() != null
+ && (!isTopLevel || field.type().isNestedType()));
+ }
+
+ @VisibleForTesting
+ static boolean requiresMissingRequiredFieldRejection(
+ Schema scanSchema, List<SlotDescriptor> projectedSlots,
+ Optional<List<Schema>> historicalSchemas) {
+ return !historicalSchemas.isPresent()
+ || requiresMissingRequiredFieldRejection(
+ scanSchema, projectedSlots, historicalSchemas.get());
+ }
+
+ @VisibleForTesting
+ static boolean requiresMissingRequiredFieldRejection(
+ Schema scanSchema, List<SlotDescriptor> projectedSlots,
+ List<Schema> historicalSchemas) {
+ Map<Integer, NestedField> fieldById =
TypeUtil.indexById(scanSchema.asStruct());
+ Map<Integer, Integer> parentById =
TypeUtil.indexParents(scanSchema.asStruct());
+ Set<Integer> collectionWrapperFieldIds = new HashSet<>();
+ collectCollectionWrapperFieldIds(scanSchema.asStruct(),
collectionWrapperFieldIds);
+ Set<Integer> potentiallyMissingRequiredFieldIds = new HashSet<>();
+ for (Schema historicalSchema : historicalSchemas) {
+ Map<Integer, NestedField> historicalFieldById =
+ TypeUtil.indexById(historicalSchema.asStruct());
+ for (NestedField field : fieldById.values()) {
+ NestedField historicalField =
historicalFieldById.get(field.fieldId());
+ if (historicalField != null) {
+ if (!collectionWrapperFieldIds.contains(field.fieldId())
+ && field.isRequired() && field.initialDefault() ==
null
+ && historicalField.isOptional()) {
+
potentiallyMissingRequiredFieldIds.add(field.fieldId());
+ }
+ continue;
+ }
+ NestedField highestMissingField = field;
+ Integer parentId = parentById.get(field.fieldId());
+ while (parentId != null &&
!historicalFieldById.containsKey(parentId)) {
+ highestMissingField =
Preconditions.checkNotNull(fieldById.get(parentId),
+ "Iceberg parent field %s is absent from scan
schema", parentId);
+ parentId = parentById.get(parentId);
+ }
+ // If the highest missing ancestor is optional, the old
physical subtree is NULL
+ // and no required descendant is materialized. A non-null
initial default is
+ // already covered by
requiresRecursiveInitialDefaultMaterialization().
+ if
(!collectionWrapperFieldIds.contains(highestMissingField.fieldId())
+ && highestMissingField.isRequired()
+ && highestMissingField.initialDefault() == null) {
+
potentiallyMissingRequiredFieldIds.add(highestMissingField.fieldId());
+ }
+ }
+ }
+ return requiresProjectedIcebergField(scanSchema, projectedSlots,
+ (field, isTopLevel) ->
potentiallyMissingRequiredFieldIds.contains(
+ field.fieldId()));
+ }
+
+ private static void collectCollectionWrapperFieldIds(
+ Type type, Set<Integer> collectionWrapperFieldIds) {
+ switch (type.typeId()) {
+ case STRUCT:
+ for (NestedField field : type.asStructType().fields()) {
+ collectCollectionWrapperFieldIds(field.type(),
collectionWrapperFieldIds);
+ }
+ break;
+ case LIST:
+ Types.ListType listType = (Types.ListType) type;
+ collectionWrapperFieldIds.add(listType.elementId());
+ collectCollectionWrapperFieldIds(
+ listType.elementType(), collectionWrapperFieldIds);
+ break;
+ case MAP:
+ Types.MapType mapType = (Types.MapType) type;
+ collectionWrapperFieldIds.add(mapType.keyId());
+ collectionWrapperFieldIds.add(mapType.valueId());
+ collectCollectionWrapperFieldIds(mapType.keyType(),
collectionWrapperFieldIds);
+ collectCollectionWrapperFieldIds(mapType.valueType(),
collectionWrapperFieldIds);
+ break;
+ default:
+ break;
+ }
+ }
+
+ private static boolean requiresProjectedIcebergField(
+ Schema scanSchema, List<SlotDescriptor> projectedSlots,
+ ProjectedFieldRequirement requirement) {
+ Map<Integer, NestedField> fieldById =
TypeUtil.indexById(scanSchema.asStruct());
+ Set<Integer> topLevelFieldIds = new HashSet<>();
+ for (NestedField field : scanSchema.columns()) {
+ topLevelFieldIds.add(field.fieldId());
+ }
+ for (SlotDescriptor slot : projectedSlots) {
+ Column column = slot.getColumn();
+ List<ColumnAccessPath> accessPaths = slot.getAllAccessPaths();
+ if (accessPaths != null && !accessPaths.isEmpty()) {
+ for (ColumnAccessPath accessPath : accessPaths) {
+ List<String> path = accessPath.getPath();
+ Preconditions.checkState(!path.isEmpty(),
+ "Iceberg column access path must not be empty");
+
Preconditions.checkState(matchesAccessPathComponent(column, path.get(0)),
+ "Iceberg access path root %s does not match column
%s", path.get(0),
+ column.getName());
+ if (requiresProjectedIcebergField(
+ column, path, 1, fieldById,
+ topLevelFieldIds.contains(column.getUniqueId()),
requirement)) {
+ return true;
+ }
+ }
+ } else if (requiresProjectedIcebergField(
+ column, slot.getType(), fieldById,
+ topLevelFieldIds.contains(column.getUniqueId()),
requirement)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean requiresProjectedIcebergField(
+ Column column, org.apache.doris.catalog.Type projectedType,
+ Map<Integer, NestedField> fieldById, boolean isTopLevel,
+ ProjectedFieldRequirement requirement) {
+ if (requiresIcebergField(column, fieldById, isTopLevel, requirement)) {
+ return true;
+ }
+ if (column.getChildren() == null) {
+ return false;
+ }
+ if (projectedType.isStructType()) {
+ for (StructField projectedField : ((StructType)
projectedType).getFields()) {
+ Column child = findChildByName(column,
projectedField.getName());
+ Preconditions.checkState(child != null,
+ "Projected Iceberg child %s is absent from column %s",
+ projectedField.getName(), column.getName());
+ if (requiresProjectedIcebergField(
+ child, projectedField.getType(), fieldById, false,
requirement)) {
+ return true;
+ }
+ }
+ } else if (projectedType.isArrayType()) {
+ Preconditions.checkState(column.getChildren().size() == 1,
+ "Iceberg array column %s must have one child",
column.getName());
+ if (requiresProjectedIcebergField(
+ column.getChildren().get(0), ((ArrayType)
projectedType).getItemType(),
+ fieldById, false, requirement)) {
+ return true;
+ }
+ } else if (projectedType.isMapType()) {
+ Preconditions.checkState(column.getChildren().size() == 2,
+ "Iceberg map column %s must have two children",
column.getName());
+ MapType mapType = (MapType) projectedType;
+ if (requiresProjectedIcebergField(
+ column.getChildren().get(0), mapType.getKeyType(),
fieldById, false,
+ requirement)
+ || requiresProjectedIcebergField(
+ column.getChildren().get(1),
mapType.getValueType(), fieldById, false,
+ requirement)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean requiresProjectedIcebergField(
+ Column column, List<String> path, int pathIndex,
+ Map<Integer, NestedField> fieldById, boolean isTopLevel,
+ ProjectedFieldRequirement requirement) {
+ if (requiresIcebergField(column, fieldById, isTopLevel, requirement)) {
+ return true;
+ }
+ if (pathIndex == path.size()) {
+ return requiresProjectedIcebergField(column, fieldById,
requirement);
+ }
+
+ String component = path.get(pathIndex);
+ if (AccessPathInfo.ACCESS_NULL.equals(component)
+ || AccessPathInfo.ACCESS_OFFSET.equals(component)) {
+ return false;
+ }
+ Preconditions.checkState(column.getChildren() != null,
+ "Iceberg access path continues below primitive column %s",
column.getName());
+
+ if (AccessPathInfo.ACCESS_ALL.equals(component)) {
+ if (column.getType().isArrayType()) {
+ Preconditions.checkState(column.getChildren().size() == 1,
+ "Iceberg array column %s must have one child",
column.getName());
+ return requiresProjectedIcebergField(
+ column.getChildren().get(0), path, pathIndex + 1,
fieldById, false,
+ requirement);
+ }
+ Preconditions.checkState(column.getType().isMapType(),
+ "Unexpected Iceberg access-all path below column %s",
column.getName());
+ Preconditions.checkState(column.getChildren().size() == 2,
+ "Iceberg map column %s must have two children",
column.getName());
+ Column key = column.getChildren().get(0);
+ // element_at(map, key) reads the complete key subtree, while any
path after '*'
+ // describes only the selected value subtree.
+ if (requiresIcebergField(key, fieldById, false, requirement)
+ || requiresProjectedIcebergField(key, fieldById,
requirement)) {
+ return true;
+ }
+ return requiresProjectedIcebergField(
+ column.getChildren().get(1), path, pathIndex + 1,
fieldById, false,
+ requirement);
+ }
+ if (column.getType().isMapType()) {
+ Preconditions.checkState(column.getChildren().size() == 2,
+ "Iceberg map column %s must have two children",
column.getName());
+ int childIndex;
+ if (AccessPathInfo.ACCESS_MAP_KEYS.equals(component)) {
+ childIndex = 0;
+ } else {
+
Preconditions.checkState(AccessPathInfo.ACCESS_MAP_VALUES.equals(component),
+ "Unexpected Iceberg map access path component %s",
component);
+ childIndex = 1;
+ }
+ return requiresProjectedIcebergField(
+ column.getChildren().get(childIndex), path, pathIndex + 1,
fieldById, false,
+ requirement);
+ }
+
+ Column child = findAccessPathChild(column, component);
+ Preconditions.checkState(child != null,
+ "Iceberg access path child %s is absent from column %s",
component,
+ column.getName());
+ return requiresProjectedIcebergField(
+ child, path, pathIndex + 1, fieldById, false, requirement);
+ }
+
+ private static boolean requiresProjectedIcebergField(
+ Column column, Map<Integer, NestedField> fieldById,
+ ProjectedFieldRequirement requirement) {
+ if (column.getChildren() == null) {
+ return false;
+ }
+ for (Column child : column.getChildren()) {
+ if (requiresIcebergField(child, fieldById, false, requirement)
+ || requiresProjectedIcebergField(child, fieldById,
requirement)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean requiresIcebergField(
+ Column column, Map<Integer, NestedField> fieldById, boolean
isTopLevel,
+ ProjectedFieldRequirement requirement) {
+ NestedField field = fieldById.get(column.getUniqueId());
+ return field != null && requirement.requires(field, isTopLevel);
+ }
+
+ private interface ProjectedFieldRequirement {
+ boolean requires(NestedField field, boolean isTopLevel);
+ }
+
+ /**
+ * Detect a reused name that current BEs resolve before an older sibling's
historical alias.
+ *
+ * <p>A smooth-upgrade source BE recognizes only the original semantics
marker and performs one
+ * ordered name/alias pass. If a sibling retains another sibling's current
name as an alias, the
+ * two BE generations can bind the same projected path to different field
IDs and types.
+ */
+ @VisibleForTesting
+ static boolean hasCurrentNameAliasCollision(
+ Schema schema, Optional<Map<Integer, List<String>>> nameMapping) {
+ return !getCurrentNameAliasCollisionFieldIds(schema,
nameMapping).isEmpty();
+ }
+
+ @VisibleForTesting
+ static void checkNameMappingBackendCompatibility(
+ Schema schema,
+ List<SlotDescriptor> projectedSlots,
+ Set<Integer> equalityDeleteFieldIds,
+ Optional<Map<Integer, List<String>>> nameMapping,
+ Iterable<Backend> backends) throws UserException {
+ Set<Integer> collisionFieldIds =
+ getCurrentNameAliasCollisionFieldIds(schema, nameMapping);
+ if (collisionFieldIds.isEmpty()) {
+ return;
+ }
+ boolean projectedCollision = requiresProjectedIcebergField(
+ schema, projectedSlots,
+ (field, isTopLevel) ->
collisionFieldIds.contains(field.fieldId()));
+ if (!projectedCollision && !equalityDeleteFieldIds.isEmpty()) {
+ Map<Integer, Integer> parentById =
TypeUtil.indexParents(schema.asStruct());
+ for (Integer equalityDeleteFieldId : equalityDeleteFieldIds) {
+ Integer fieldId = equalityDeleteFieldId;
+ while (fieldId != null) {
+ if (collisionFieldIds.contains(fieldId)) {
+ projectedCollision = true;
+ break;
+ }
+ fieldId = parentById.get(fieldId);
+ }
+ if (projectedCollision) {
+ break;
+ }
+ }
+ }
+ if (projectedCollision) {
+ checkCurrentIcebergScanSemanticsBackendCompatibility(backends);
+ }
+ }
+
+ private static Set<Integer> getCurrentNameAliasCollisionFieldIds(
+ Schema schema, Optional<Map<Integer, List<String>>> nameMapping) {
+ Set<Integer> collisionFieldIds = new HashSet<>();
+ if (nameMapping.isPresent()) {
+ collectCurrentNameAliasCollisionFieldIds(
+ schema.asStruct(), nameMapping.get(), collisionFieldIds);
+ }
+ return collisionFieldIds;
+ }
+
+ private static void collectCurrentNameAliasCollisionFieldIds(
+ Type type, Map<Integer, List<String>> nameMapping,
+ Set<Integer> collisionFieldIds) {
+ switch (type.typeId()) {
+ case STRUCT:
+ List<NestedField> fields = type.asStructType().fields();
+ for (NestedField field : fields) {
+ List<String> aliases =
+ nameMapping.getOrDefault(field.fieldId(),
Collections.emptyList());
+ for (String alias : aliases) {
+ for (NestedField sibling : fields) {
+ if (sibling.fieldId() != field.fieldId()
+ && sibling.name().equalsIgnoreCase(alias))
{
+ collisionFieldIds.add(field.fieldId());
+ collisionFieldIds.add(sibling.fieldId());
+ }
+ }
+ }
+ collectCurrentNameAliasCollisionFieldIds(
+ field.type(), nameMapping, collisionFieldIds);
+ }
+ return;
+ case LIST:
+ collectCurrentNameAliasCollisionFieldIds(
+ type.asListType().elementType(), nameMapping,
collisionFieldIds);
+ return;
+ case MAP:
+ collectCurrentNameAliasCollisionFieldIds(
+ type.asMapType().keyType(), nameMapping,
collisionFieldIds);
+ collectCurrentNameAliasCollisionFieldIds(
+ type.asMapType().valueType(), nameMapping,
collisionFieldIds);
+ return;
+ default:
+ return;
+ }
+ }
+
+ private static boolean matchesAccessPathComponent(Column column, String
component) {
+ return Integer.toString(column.getUniqueId()).equals(component)
+ || column.getName().equalsIgnoreCase(component);
+ }
+
+ private static Column findAccessPathChild(Column column, String component)
{
+ for (Column child : column.getChildren()) {
+ if (matchesAccessPathComponent(child, component)) {
+ return child;
+ }
+ }
+ return null;
+ }
+
+ private static Column findChildByName(Column column, String childName) {
+ for (Column child : column.getChildren()) {
+ if (child.getName().equalsIgnoreCase(childName)) {
+ return child;
+ }
+ }
+ return null;
+ }
+
+ @VisibleForTesting
+ Set<Integer> getEqualityDeleteFieldIdsForScan() throws UserException {
+ TableScan scan = createTableScan();
+ if (scan.snapshot() == null) {
+ return Collections.emptySet();
+ }
+ try {
+ return preExecutionAuthenticator.execute(
+ () -> loadEqualityDeleteFieldIds(scan));
+ } catch (Exception e) {
+ Optional<NotSupportedException> opt =
checkNotSupportedException(e);
+ if (opt.isPresent()) {
+ throw opt.get();
+ }
+ throw new UserException(ExceptionUtils.getRootCauseMessage(e), e);
+ }
+ }
+
+ /**
+ * Skip exhaustive delete-file planning when the exact snapshot summary
already proves that
+ * metadata-only COUNT(*) is safe. A usable count requires the summary's
equality-delete total
+ * to be zero, so no equality field IDs can affect this scan.
+ */
+ @VisibleForTesting
+ Set<Integer> getEqualityDeleteFieldIdsForPlanning() throws UserException {
+ if (prepareTableLevelSnapshotCount()) {
+ return Collections.emptySet();
+ }
+ return getEqualityDeleteFieldIdsForScan();
+ }
+
+ @VisibleForTesting
+ Set<Integer> loadEqualityDeleteFieldIds(TableScan scan) {
+ ConnectContext context = ConnectContext.get();
+ Preconditions.checkNotNull(context);
+ Preconditions.checkNotNull(context.getStatementContext());
+ List<FileScanTask> rewriteTasks =
+ context.getStatementContext().getIcebergRewriteFileScanTasks();
+ if (rewriteTasks != null) {
+ return collectEqualityDeleteFieldIdsFromTasks(rewriteTasks);
+ }
+ if (isBatchMode()) {
+ return loadEqualityDeleteFieldIdsFromDeleteManifests(scan);
+ }
+
+ List<FileScanTask> tasks = new ArrayList<>();
+ try (CloseableIterable<FileScanTask> plannedTasks =
planFileScanTaskWithoutReuse(scan)) {
+ for (FileScanTask task : plannedTasks) {
+ tasks.add(task);
+ }
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to close Iceberg file scan
tasks", e);
+ }
+ preplannedFileScanTasks = tasks;
+ return collectEqualityDeleteFieldIdsFromTasks(tasks);
+ }
+
+ /**
+ * Load applicable equality-delete metadata for batch scans.
+ *
+ * <p>Batch split planning must remain asynchronous, so this preflight
cannot consume
+ * {@link TableScan#planFiles()}. The target snapshot's equality-delete
total avoids manifest
+ * reads for the common no-delete case. Otherwise, the preflight reads
filtered manifest
+ * metadata and applies the same {@link DeleteFileIndex#forDataFile(long,
DataFile)}
+ * delete-to-data contract as final task planning.
+ */
+ @VisibleForTesting
+ Set<Integer> loadEqualityDeleteFieldIdsFromDeleteManifests(TableScan scan)
{
+ Snapshot snapshot = Preconditions.checkNotNull(scan.snapshot());
+ String totalEqualityDeletes =
+ snapshot.summary().get(SnapshotSummary.TOTAL_EQ_DELETES_PROP);
+ if (totalEqualityDeletes != null &&
Long.parseLong(totalEqualityDeletes) == 0) {
+ return Collections.emptySet();
+ }
+
+ Expression dataFilter = scan.filter();
+ boolean caseSensitive = scan.isCaseSensitive();
+ Map<Integer, PartitionSpec> specsById = icebergTable.specs();
+ List<DeleteFile> deleteFiles = new ArrayList<>();
+ for (ManifestFile manifest :
snapshot.deleteManifests(icebergTable.io())) {
+ if (manifest.content() != ManifestContent.DELETES
+ || (!manifest.hasAddedFiles() &&
!manifest.hasExistingFiles())) {
+ continue;
+ }
+ PartitionSpec spec = Preconditions.checkNotNull(
+ specsById.get(manifest.partitionSpecId()),
+ "Iceberg partition spec %s is absent from table metadata",
+ manifest.partitionSpecId());
+ Expression partitionFilter =
+ Projections.inclusive(spec,
caseSensitive).project(dataFilter);
+ if (!ManifestEvaluator.forPartitionFilter(
+ partitionFilter, spec, caseSensitive).eval(manifest)) {
+ continue;
+ }
+ try (ManifestReader<DeleteFile> reader =
ManifestFiles.readDeleteManifest(
+ manifest, icebergTable.io(), specsById)) {
+ ManifestReader<DeleteFile> filteredReader = reader
+ .filterRows(dataFilter)
+ .filterPartitions(partitionFilter)
+ .caseSensitive(caseSensitive);
+ for (DeleteFile deleteFile : filteredReader) {
+ if (deleteFile.content() == FileContent.EQUALITY_DELETES) {
+ deleteFiles.add(deleteFile);
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(
+ "Failed to close Iceberg delete manifest " +
manifest.path(), e);
+ }
+ }
+ if (deleteFiles.isEmpty()) {
+ return Collections.emptySet();
+ }
+
+ DeleteFileIndex deleteIndex = DeleteFileIndex.builderFor(deleteFiles)
+ .specsById(specsById)
+ .caseSensitive(caseSensitive)
+ .build();
+ Set<Integer> equalityDeleteFieldIds = new HashSet<>();
+ for (ManifestFile manifest :
snapshot.dataManifests(icebergTable.io())) {
+ if (manifest.content() != ManifestContent.DATA
+ || (!manifest.hasAddedFiles() &&
!manifest.hasExistingFiles())) {
+ continue;
+ }
+ PartitionSpec spec = Preconditions.checkNotNull(
+ specsById.get(manifest.partitionSpecId()),
+ "Iceberg partition spec %s is absent from table metadata",
+ manifest.partitionSpecId());
+ Expression partitionFilter =
+ Projections.inclusive(spec,
caseSensitive).project(dataFilter);
+ if (!ManifestEvaluator.forPartitionFilter(
+ partitionFilter, spec, caseSensitive).eval(manifest)) {
+ continue;
+ }
+ try (ManifestReader<DataFile> reader =
+ ManifestFiles.read(manifest, icebergTable.io())) {
Review Comment:
[P1] Bind filtered data manifests with the current schema
This two-argument overload reconstructs its `PartitionSpec` from the
manifest's embedded schema, but `dataFilter` was built against
`icebergTable.schema()`. After `old_name` is renamed to `new_name`, a batch
scan with equality deletes and `WHERE new_name = ...` can reach an older data
manifest here and fail when `ManifestReader` binds the current-name predicate
against that historical schema, before any splits are dispatched. Iceberg's
1.10.1 `ManifestFiles.read` contract explicitly directs filtered readers to the
three-argument overload so the latest table schema is used. Please pass
`specsById` here as the delete-manifest path already does, and cover a filtered
batch scan across a 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]
