github-actions[bot] commented on code in PR #65851: URL: https://github.com/apache/doris/pull/65851#discussion_r3714402517
########## be/src/format/table/iceberg_default_value.h: ########## @@ -0,0 +1,488 @@ +// 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. + +#pragma once + +#include <gen_cpp/ExternalTableSchema_types.h> +#include <rapidjson/document.h> +#include <rapidjson/stringbuffer.h> +#include <rapidjson/writer.h> + +#include <cstddef> +#include <deque> +#include <string> +#include <string_view> +#include <unordered_map> +#include <utility> + +#include "common/status.h" +#include "core/assert_cast.h" +#include "core/column/column.h" +#include "core/data_type/data_type.h" +#include "core/data_type/data_type_array.h" +#include "core/data_type/data_type_map.h" +#include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_struct.h" +#include "core/data_type/primitive_type.h" +#include "core/field.h" +#include "util/string_util.h" +#include "util/url_coding.h" + +namespace doris::iceberg { + +namespace detail { + +inline const schema::external::TField* get_field_ptr(const schema::external::TFieldPtr& field_ptr) { + if (!field_ptr.__isset.field_ptr || field_ptr.field_ptr == nullptr) { + return nullptr; + } + return field_ptr.field_ptr.get(); +} + +inline const schema::external::TField* find_struct_child( + const schema::external::TStructField& struct_field, const std::string& name) { + if (!struct_field.__isset.fields) { + return nullptr; + } + for (const auto& child_ptr : struct_field.fields) { + const auto* child = get_field_ptr(child_ptr); + if (child != nullptr && child->__isset.name && iequal(child->name, name)) { + return child; + } + } + for (const auto& child_ptr : struct_field.fields) { + const auto* child = get_field_ptr(child_ptr); + if (child == nullptr || !child->__isset.name_mapping) { + continue; + } + for (const auto& alias : child->name_mapping) { + if (iequal(alias, name)) { + return child; + } + } + } + return nullptr; +} + +inline int hex_value(char c) { + if (c >= '0' && c <= '9') { + return c - '0'; + } + if (c >= 'a' && c <= 'f') { + return c - 'a' + 10; + } + if (c >= 'A' && c <= 'F') { + return c - 'A' + 10; + } + return -1; +} + +inline Status decode_hex(std::string_view encoded, std::string* decoded) { + DORIS_CHECK(decoded != nullptr); + if ((encoded.size() & 1U) != 0) { + return Status::InvalidArgument("Invalid odd-length Iceberg binary default"); + } + decoded->resize(encoded.size() / 2); + for (size_t index = 0; index < encoded.size(); index += 2) { + const int high = hex_value(encoded[index]); + const int low = hex_value(encoded[index + 1]); + if (high < 0 || low < 0) { + return Status::InvalidArgument("Invalid hexadecimal Iceberg binary default"); + } + (*decoded)[index / 2] = static_cast<char>((high << 4) | low); + } + return Status::OK(); +} + +inline Status decode_json_binary(std::string_view encoded, std::string* decoded) { + DORIS_CHECK(decoded != nullptr); + const bool is_uuid = encoded.size() == 36 && encoded[8] == '-' && encoded[13] == '-' && + encoded[18] == '-' && encoded[23] == '-'; + if (is_uuid) { + std::string uuid_hex; + uuid_hex.reserve(32); + for (size_t index = 0; index < encoded.size(); ++index) { + if (index != 8 && index != 13 && index != 18 && index != 23) { + uuid_hex.push_back(encoded[index]); + } + } + return decode_hex(uuid_hex, decoded); + } + return decode_hex(encoded, decoded); +} + +inline std::string json_scalar_text(const rapidjson::Value& value) { + if (value.IsString()) { + return {value.GetString(), value.GetStringLength()}; + } + rapidjson::StringBuffer buffer; + rapidjson::Writer<rapidjson::StringBuffer> writer(buffer); + value.Accept(writer); + return {buffer.GetString(), buffer.GetSize()}; +} + +inline void normalize_timestamp_for_doris(PrimitiveType primitive_type, std::string* value) { + if (primitive_type != TYPE_DATETIME && primitive_type != TYPE_DATETIMEV2 && + primitive_type != TYPE_TIMESTAMPTZ) { + return; + } + if (const size_t separator = value->find('T'); separator != std::string::npos) { + (*value)[separator] = ' '; + } + if (primitive_type == TYPE_TIMESTAMPTZ) { + return; + } + if (value->ends_with('Z')) { + value->pop_back(); + return; + } + const size_t time_start = value->find(' '); + if (time_start == std::string::npos) { + return; + } + const size_t offset = value->find_first_of("+-", time_start + 1); + if (offset != std::string::npos) { + value->erase(offset); + } +} + +inline Status make_null_field(const schema::external::TField& field, const DataTypePtr& data_type, + Field* result) { + DORIS_CHECK(data_type != nullptr); + DORIS_CHECK(result != nullptr); + if (field.__isset.is_optional && !field.is_optional) { + return Status::InvalidArgument("Required Iceberg field '{}' has a null default", + field.name); + } + if (!data_type->is_nullable()) { + return Status::InternalError( + "Optional Iceberg field '{}' has a null default, but its Doris type '{}' is not " + "nullable", + field.name, data_type->get_name()); + } + *result = Field(); + return Status::OK(); +} + +inline Status build_initial_default_field(const schema::external::TField& field, + const DataTypePtr& data_type, + std::deque<std::string>* binary_storage, Field* result); + +inline Status build_json_default_field(const schema::external::TField& field, + const DataTypePtr& data_type, + const rapidjson::Value& json_value, + std::deque<std::string>* binary_storage, Field* result); + +inline Status build_json_struct_default(const schema::external::TField& field, + const DataTypePtr& value_type, + const rapidjson::Value& json_value, + std::deque<std::string>* binary_storage, Field* result) { + if (!json_value.IsObject() || !field.__isset.nestedField || + !field.nestedField.__isset.struct_field || !field.nestedField.struct_field.__isset.fields) { + return Status::InvalidArgument("Invalid Iceberg struct default for field '{}'", field.name); + } + + const auto& struct_type = assert_cast<const DataTypeStruct&>(*value_type); + Struct struct_value; + struct_value.reserve(struct_type.get_elements().size()); + for (size_t index = 0; index < struct_type.get_elements().size(); ++index) { + const auto& child_name = struct_type.get_element_name(index); + const auto* child = find_struct_child(field.nestedField.struct_field, child_name); + if (child == nullptr || !child->__isset.id) { + return Status::InvalidArgument( + "Iceberg struct default for field '{}' is missing metadata for projected " + "child '{}'", + field.name, child_name); + } + + const std::string child_id = std::to_string(child->id); + const auto member = json_value.FindMember(child_id.c_str()); + Field child_value; + if (member == json_value.MemberEnd()) { + RETURN_IF_ERROR(build_initial_default_field(*child, struct_type.get_element(index), + binary_storage, &child_value)); + } else { + RETURN_IF_ERROR(build_json_default_field(*child, struct_type.get_element(index), + member->value, binary_storage, &child_value)); + } + struct_value.push_back(std::move(child_value)); + } + *result = Field::create_field<TYPE_STRUCT>(std::move(struct_value)); + return Status::OK(); +} + +// The recursive item TField describes the element schema and its field-level default metadata. It +// cannot represent a particular list literal's length or per-position values, so the parent +// initial-default keeps those values in Iceberg's single-value JSON array. +inline Status build_json_array_default(const schema::external::TField& field, + const DataTypePtr& value_type, + const rapidjson::Value& json_value, + std::deque<std::string>* binary_storage, Field* result) { + if (!json_value.IsArray() || !field.__isset.nestedField || + !field.nestedField.__isset.array_field || + !field.nestedField.array_field.__isset.item_field) { + return Status::InvalidArgument("Invalid Iceberg list default for field '{}'", field.name); + } + const auto* element = get_field_ptr(field.nestedField.array_field.item_field); + if (element == nullptr) { + return Status::InvalidArgument( + "Iceberg list default for field '{}' has incomplete element metadata", field.name); + } + + const auto& array_type = assert_cast<const DataTypeArray&>(*value_type); + Array array_value; + array_value.reserve(json_value.Size()); + for (const auto& json_element : json_value.GetArray()) { + Field element_value; + RETURN_IF_ERROR(build_json_default_field(*element, array_type.get_nested_type(), + json_element, binary_storage, &element_value)); + array_value.push_back(std::move(element_value)); + } + *result = Field::create_field<TYPE_ARRAY>(std::move(array_value)); + return Status::OK(); +} + +// The recursive key/value TFields describe entry schemas and field-level default metadata. They +// cannot represent the number, order, or concrete values of map entries, so the parent +// initial-default keeps the entries in Iceberg's single-value JSON key/value arrays. +inline Status build_json_map_default(const schema::external::TField& field, + const DataTypePtr& value_type, + const rapidjson::Value& json_value, + std::deque<std::string>* binary_storage, Field* result) { + if (!json_value.IsObject() || !json_value.HasMember("keys") || !json_value["keys"].IsArray() || + !json_value.HasMember("values") || !json_value["values"].IsArray() || + !field.__isset.nestedField || !field.nestedField.__isset.map_field || + !field.nestedField.map_field.__isset.key_field || + !field.nestedField.map_field.__isset.value_field) { + return Status::InvalidArgument("Invalid Iceberg map default for field '{}'", field.name); + } + const auto& keys = json_value["keys"]; + const auto& values = json_value["values"]; + if (keys.Size() != values.Size()) { + return Status::InvalidArgument( + "Iceberg map default for field '{}' has {} keys but {} values", field.name, + keys.Size(), values.Size()); + } + + const auto* key = get_field_ptr(field.nestedField.map_field.key_field); + const auto* value = get_field_ptr(field.nestedField.map_field.value_field); + if (key == nullptr || value == nullptr) { + return Status::InvalidArgument( + "Iceberg map default for field '{}' has incomplete key/value metadata", field.name); + } + + const auto& map_type = assert_cast<const DataTypeMap&>(*value_type); + Array key_fields; + Array value_fields; + key_fields.reserve(keys.Size()); + value_fields.reserve(values.Size()); + for (rapidjson::SizeType index = 0; index < keys.Size(); ++index) { + Field key_value; + Field mapped_value; + RETURN_IF_ERROR(build_json_default_field(*key, map_type.get_key_type(), keys[index], + binary_storage, &key_value)); + RETURN_IF_ERROR(build_json_default_field(*value, map_type.get_value_type(), values[index], + binary_storage, &mapped_value)); + key_fields.push_back(std::move(key_value)); + value_fields.push_back(std::move(mapped_value)); + } + Map map_value; + map_value.push_back(Field::create_field<TYPE_ARRAY>(std::move(key_fields))); + map_value.push_back(Field::create_field<TYPE_ARRAY>(std::move(value_fields))); + *result = Field::create_field<TYPE_MAP>(std::move(map_value)); + return Status::OK(); +} + +inline Status build_json_scalar_default(const schema::external::TField& field, + const DataTypePtr& value_type, + const rapidjson::Value& json_value, + std::deque<std::string>* binary_storage, Field* result) { + const auto primitive_type = value_type->get_primitive_type(); + std::string serialized_value = json_scalar_text(json_value); + const bool binary_like = (field.__isset.initial_default_value_is_base64 && + field.initial_default_value_is_base64) || + primitive_type == TYPE_VARBINARY; + if (binary_like) { + if (!json_value.IsString()) { + return Status::InvalidArgument( + "Iceberg binary default for field '{}' is not a JSON string", field.name); + } + binary_storage->emplace_back(); + RETURN_IF_ERROR(decode_json_binary(serialized_value, &binary_storage->back())); + if (primitive_type == TYPE_VARBINARY) { + *result = Field::create_field<TYPE_VARBINARY>(StringView(binary_storage->back())); + } else if (is_string_type(primitive_type)) { + *result = Field::create_field<TYPE_STRING>(binary_storage->back()); + } else { + return Status::InvalidArgument( + "Iceberg binary default for field '{}' has incompatible Doris type '{}'", + field.name, value_type->get_name()); + } + return Status::OK(); + } + + if (is_string_type(primitive_type)) { + if (!json_value.IsString()) { + return Status::InvalidArgument("Iceberg string default for field '{}' is not a string", + field.name); + } + *result = Field::create_field<TYPE_STRING>(std::move(serialized_value)); + return Status::OK(); + } + normalize_timestamp_for_doris(primitive_type, &serialized_value); + RETURN_IF_ERROR(value_type->get_serde()->from_fe_string(serialized_value, *result)); + return Status::OK(); +} + +inline Status build_json_default_field(const schema::external::TField& field, + const DataTypePtr& data_type, + const rapidjson::Value& json_value, + std::deque<std::string>* binary_storage, Field* result) { + DORIS_CHECK(data_type != nullptr); + DORIS_CHECK(binary_storage != nullptr); + DORIS_CHECK(result != nullptr); + if (json_value.IsNull()) { + return make_null_field(field, data_type, result); + } + + const auto value_type = remove_nullable(data_type); + switch (value_type->get_primitive_type()) { + case TYPE_STRUCT: + return build_json_struct_default(field, value_type, json_value, binary_storage, result); + case TYPE_ARRAY: + return build_json_array_default(field, value_type, json_value, binary_storage, result); + case TYPE_MAP: + return build_json_map_default(field, value_type, json_value, binary_storage, result); + default: + return build_json_scalar_default(field, value_type, json_value, binary_storage, result); + } +} + +inline Status build_initial_default_field(const schema::external::TField& field, + const DataTypePtr& data_type, + std::deque<std::string>* binary_storage, Field* result) { + DORIS_CHECK(data_type != nullptr); + DORIS_CHECK(binary_storage != nullptr); + DORIS_CHECK(result != nullptr); + if (!field.__isset.initial_default_value) { + if (field.__isset.is_optional && !field.is_optional) { + return Status::InvalidArgument( + "Required Iceberg field '{}' is missing from the data file and has no initial " + "default", + field.name); + } + return make_null_field(field, data_type, result); + } + + const auto value_type = remove_nullable(data_type); + const auto primitive_type = value_type->get_primitive_type(); + if (is_complex_type(primitive_type)) { + rapidjson::Document document; + document.Parse(field.initial_default_value.data(), field.initial_default_value.size()); + if (document.HasParseError()) { + return Status::InvalidArgument("Invalid Iceberg JSON initial default for field '{}'", + field.name); + } + return build_json_default_field(field, data_type, document, binary_storage, result); + } + + const bool default_is_base64 = (field.__isset.initial_default_value_is_base64 && + field.initial_default_value_is_base64) || + primitive_type == TYPE_VARBINARY; + if (default_is_base64) { + binary_storage->emplace_back(); + if (!base64_decode(field.initial_default_value, &binary_storage->back())) { + return Status::InvalidArgument("Invalid Base64 Iceberg initial default for field '{}'", + field.name); + } + if (primitive_type == TYPE_VARBINARY) { + *result = Field::create_field<TYPE_VARBINARY>(StringView(binary_storage->back())); + } else if (is_string_type(primitive_type)) { + *result = Field::create_field<TYPE_STRING>(binary_storage->back()); + } else { + return Status::InvalidArgument( + "Iceberg field '{}' marks its initial default as Base64, but Doris type '{}' " + "cannot contain binary data", + field.name, value_type->get_name()); + } + return Status::OK(); + } + + RETURN_IF_ERROR(value_type->get_serde()->from_fe_string(field.initial_default_value, *result)); Review Comment: [P1] Decode non-finite initial defaults before the OLAP parser. This new materializer receives Iceberg FLOAT/DOUBLE initial defaults as `NaN`, `Infinity`, or `-Infinity`, but `from_fe_string` delegates to `DataTypeNumberSerDe::from_olap_string`, which explicitly rejects every parsed NaN/infinity. Consequently an older file missing a newly defaulted field fails to scan instead of materializing the legal Iceberg value; complex/defaulted children hit the same call at line 347, and format-v2 duplicates both paths at `iceberg_reader.cpp:360,435`. The existing typed-Cast fix is write-side only. Please construct the typed non-finite Field (or use a parser that admits these three spellings) and add old-file V1/V2 Parquet/ORC coverage. ########## fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java: ########## @@ -1728,57 +1758,440 @@ private static List<String> requestedLowerNames(List<ConnectorColumnHandle> colu return names; } + @VisibleForTesting + static boolean hasApplicableEqualityDeletes(TableScan scan) { + Snapshot snapshot = scan.snapshot(); + if (snapshot == null + || "0".equals(snapshot.summary().get(TOTAL_EQUALITY_DELETES))) { + return false; + } + // planFiles binds delete files to the exact filtered data-file tasks after partition and sequence + // pruning. A snapshot summary of zero returns above without planning; a positive or missing summary + // needs this exact proof. Iterate whole-file tasks lazily and stop at the first equality delete: this + // keeps memory O(1), does not create or retain byte-split tasks, and avoids snapshot-wide delete + // counters forcing new-BE-only semantics when no dispatched task can consume an equality delete. + try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) { + for (FileScanTask task : tasks) { + for (DeleteFile delete : task.deletes()) { + if (delete.content() == FileContent.EQUALITY_DELETES) { + return true; + } + } + } + } catch (IOException e) { + throw new DorisConnectorException( + "Failed to inspect applicable Iceberg equality deletes: " + e.getMessage(), e); + } + return false; + } + /** - * Ensure the schema-evolution dict carries the table's equality-delete KEY columns even when the query - * does not project them (#65502). Equality-delete keys are hidden scan dependencies: BE resolves a key - * that is missing from an OLD data file by looking its field id up in this dict to get the column type + - * iceberg initial default; without the entry BE materializes the key as NULL and mis-applies the delete. - * The keys are the table's declared identifier fields (what equality-delete writers key on) -> a few - * columns, DCHECK-safe superset (BE looks up only its own scan slots; the pin/top-N branches already ship - * the full schema). If the table declares NO identifier yet the scan carries equality deletes (whose - * equality_ids we cannot cheaply enumerate here), fall back to the full schema. Non-identifier / - * append-only / position-delete-only tables are unaffected (the pruned dict is returned verbatim). + * Build a schema carrier that can resolve any equality key reachable before the selected schema without + * enumerating data files, manifests, or byte-split tasks. Its retained state is bounded by table schema + * history rather than scan cardinality. At execution time BE looks fields up by the exact IDs on each + * {@link FileScanTask#deletes()}; unrelated carrier fields never participate in delete matching. + * + * <p>The selected snapshot lineage wins when a field was renamed. The metadata schema list, in its actual + * chronology up to the selected schema (schema IDs are identifiers, not a sequence), fills schema-only + * changes and expired ancestors. Current fields remain first, so a dropped/re-added name still resolves the + * projected current field by name while a historical equality key resolves by its stable field ID.</p> */ - private List<String> withEqualityDeleteKeyColumns(Table table, List<String> requested) { - if (requested.isEmpty()) { - // An empty requested list already makes buildCurrentSchema fall back to the FULL schema (every - // top-level column) — a superset that covers every equality-delete key — so there is nothing to - // force-include. Returning early also preserves that all-columns fallback (a non-empty identifier - // set would otherwise prune it to identifier-only) and skips the table.schema()/currentSnapshot() - // probe when it cannot change the result. - return requested; - } - Schema schema = table.schema(); - Set<Integer> identifierFieldIds = schema.identifierFieldIds(); - if (identifierFieldIds.isEmpty()) { - return hasEqualityDeletes(table) ? Collections.emptyList() : requested; - } - Set<String> present = new HashSet<>(); - for (String name : requested) { - present.add(name.toLowerCase(Locale.ROOT)); - } - List<String> result = new ArrayList<>(requested); - for (int fieldId : identifierFieldIds) { - Types.NestedField field = schema.findField(fieldId); + @VisibleForTesting + static List<NestedField> schemaForPotentialEqualityDeletes( + Table table, TableScan scan, Schema scanSchema) { + List<Schema> metadataSchemas = metadataSchemaHistory(table); + int selectedSchemaIndex = -1; + for (int index = 0; index < metadataSchemas.size(); index++) { + if (metadataSchemas.get(index).schemaId() == scanSchema.schemaId()) { + selectedSchemaIndex = index; + } + } + int lastRelevantIndex = selectedSchemaIndex >= 0 + ? selectedSchemaIndex : metadataSchemas.size() - 1; + Set<Integer> missing = new HashSet<>(); + for (int index = 0; index <= lastRelevantIndex; index++) { + Schema schema = metadataSchemas.get(index); + for (NestedField field : TypeUtil.indexById(schema.asStruct()).values()) { + if (field.type().isPrimitiveType()) { + missing.add(field.fieldId()); + } + } + } + missing.removeAll(TypeUtil.indexById(scanSchema.asStruct()).keySet()); + if (missing.isEmpty()) { + return scanSchema.columns(); + } + + List<NestedField> fields = new ArrayList<>(scanSchema.columns()); + Map<Integer, Schema> schemasById = table.schemas(); + Snapshot snapshot = scan.snapshot(); + while (snapshot != null && !missing.isEmpty()) { + Integer schemaId = snapshot.schemaId(); + if (schemaId != null) { + Schema historicalSchema = schemasById.get(schemaId); + if (historicalSchema == null) { + throw new IllegalStateException( + "Iceberg snapshot schema " + schemaId + " is absent from table metadata"); + } + addHistoricalEqualityFields(fields, missing, historicalSchema); + } + if (missing.isEmpty()) { + break; + } + Long parentId = snapshot.parentId(); + snapshot = parentId == null ? null : table.snapshot(parentId); + } + for (int index = lastRelevantIndex; index >= 0 && !missing.isEmpty(); index--) { + addHistoricalEqualityFields(fields, missing, metadataSchemas.get(index)); + } + if (!missing.isEmpty()) { + throw new IllegalStateException( + "Iceberg historical primitive fields are absent from schema history: " + missing); + } + return fields; + } + + private static List<Schema> metadataSchemaHistory(Table table) { + if (table instanceof HasTableOperations) { + return ((HasTableOperations) table).operations().current().schemas(); + } + return new ArrayList<>(table.schemas().values()); + } + + 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) { + if (!field.type().isPrimitiveType()) { + throw new IllegalStateException( + "Iceberg equality-delete field " + fieldId + " must be primitive"); + } + 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) { + if (currentType.typeId() != historicalType.typeId()) { + throw new IllegalStateException("Iceberg equality-delete ancestor type changed from " + + historicalType + " to " + currentType); + } + switch (currentType.typeId()) { + case STRUCT: + List<NestedField> mergedFields = + new ArrayList<>(currentType.asStructType().fields()); + mergeHistoricalEqualityFields(mergedFields, historicalType.asStructType().fields()); + return mergedFields.equals(currentType.asStructType().fields()) + ? currentType : Types.StructType.of(mergedFields); + case LIST: + Types.ListType currentList = currentType.asListType(); + Types.ListType historicalList = historicalType.asListType(); + if (currentList.elementId() != historicalList.elementId()) { + throw new IllegalStateException( + "Iceberg equality-delete list element field ID changed"); + } + 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(); + if (currentMap.keyId() != historicalMap.keyId() + || currentMap.valueId() != historicalMap.valueId()) { + throw new IllegalStateException( + "Iceberg equality-delete map field IDs changed"); + } + 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: + if (!currentType.equals(historicalType)) { + throw new IllegalStateException("Iceberg equality-delete field type changed from " + + historicalType + " to " + currentType); + } + return currentType; + } + } + + private static boolean requiresCurrentScanSemantics( + Table table, TableScan scan, Schema scanSchema, List<ConnectorColumnHandle> columns, + boolean hasApplicableEqualityDeletes, + Optional<Map<Integer, List<String>>> nameMapping) { + if (hasApplicableEqualityDeletes) { + return true; + } + Set<Integer> projectedFieldIds = projectedFieldIds(scanSchema, columns); + Set<Integer> topLevelFieldIds = new HashSet<>(); + for (NestedField field : scanSchema.columns()) { + topLevelFieldIds.add(field.fieldId()); + } + Map<Integer, NestedField> fields = TypeUtil.indexById(scanSchema.asStruct()); + for (Integer fieldId : projectedFieldIds) { + NestedField field = fields.get(fieldId); + if (field != null && field.initialDefault() != null + && (!topLevelFieldIds.contains(fieldId) || field.type().isNestedType())) { + return true; + } + } + if (hasProjectedNameAliasCollision(scanSchema, projectedFieldIds, nameMapping)) { + return true; + } + return selectedHistoryRequiresMissingRequiredFieldRejection( + table, scanSchema, projectedFieldIds, scan.snapshot()); + } + + @VisibleForTesting + static Set<Integer> projectedFieldIds( + Schema scanSchema, List<ConnectorColumnHandle> columns) { + Set<Integer> projected = new HashSet<>(); + if (columns == null || columns.isEmpty()) { Review Comment: [P1] Keep historical-only projected slots in this compatibility gate. On a snapshot-pinned read, the tuple and `scanSchema` come from the historical schema, but Iceberg's column-handle callback still returns the latest schema. `PluginDrivenScanNode.buildColumnHandles` silently drops a pinned top-level field that was later removed. If the query also selects one surviving field, `columns` is nonempty here, so the full-schema fallback does not run and none of the v2-semantics checks sees the dropped projection. The full pinned dictionary still lets a current BE evaluate that slot, while a semantics-v1 source BE can produce its legacy NULL/default behavior. Please derive this fence from the pinned scan slots, or conservatively add unmatched pinned fields, and test a time-travel projection containing one surviving and one later-dropped v2-only field. -- 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]
