github-actions[bot] commented on code in PR #65851:
URL: https://github.com/apache/doris/pull/65851#discussion_r3710915137
##########
be/src/format/table/iceberg_reader.cpp:
##########
@@ -258,84 +519,89 @@ Status
IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
// - Prefix with __equality_delete_column__ to avoid name conflicts
// - Correctly map table_col_name → file_col_name in table_info_node
const static std::string EQ_DELETE_PRE = "__equality_delete_column__";
- std::unordered_map<int, const FieldSchema*> field_id_to_file_column;
bool all_file_columns_have_field_ids = true;
bool any_file_column_has_field_id = false;
for (int i = 0; i < field_desc->size(); ++i) {
const auto* field_schema = field_desc->get_column(i);
if (field_schema) {
if (field_schema->field_id < 0) {
all_file_columns_have_field_ids = false;
- } else {
+ }
+ if (parquet_subtree_has_iceberg_id(*field_schema)) {
any_file_column_has_field_id = true;
- field_id_to_file_column[field_schema->field_id] = field_schema;
}
}
}
const bool use_field_ids_for_hidden_keys =
supports_iceberg_scan_semantics_v1(&get_scan_params())
? any_file_column_has_field_id
: all_file_columns_have_field_ids;
- const auto struct_node =
-
std::dynamic_pointer_cast<TableSchemaChangeHelper::StructNode>(ctx->table_info_node);
- DORIS_CHECK(struct_node != nullptr);
+ const auto find_file_column_by_name = [&](const std::string& name) ->
const FieldSchema* {
+ for (int j = 0; j < field_desc->size(); ++j) {
+ const auto* candidate = field_desc->get_column(j);
+ if (candidate != nullptr && iequal(candidate->name, name)) {
+ return candidate;
+ }
+ }
+ return nullptr;
+ };
// Rebuild _expand_col_names with proper file-column-based names
std::vector<std::string> new_expand_col_names;
+ DORIS_CHECK(_expand_col_names.size() == _expand_col_field_ids.size());
+ DORIS_CHECK(_expand_col_names.size() == _expand_columns.size());
for (size_t i = 0; i < _expand_col_names.size(); ++i) {
const auto& old_name = _expand_col_names[i];
- // Find the field_id for this expand column
- int field_id = -1;
- for (auto& [fid, name] : _id_to_block_column_name) {
- if (name == old_name) {
- field_id = fid;
- break;
- }
- }
+ const int32_t field_id = _expand_col_field_ids[i];
const FieldSchema* file_column = nullptr;
+ ParquetEqualityFieldPath file_path;
+ bool complete_file_path = false;
if (use_field_ids_for_hidden_keys) {
- auto id_it = field_id_to_file_column.find(field_id);
- if (id_it != field_id_to_file_column.end()) {
- file_column = id_it->second;
+ complete_file_path =
+ find_parquet_equality_field_path_by_id(field_desc,
field_id, &file_path);
+ if (!complete_file_path &&
supports_iceberg_scan_semantics_v2(&get_scan_params())) {
+ const auto table_path = _find_schema_field_path(field_id);
+ if (!table_path.empty()) {
+ complete_file_path =
find_parquet_equality_field_prefix_by_id_path(
Review Comment:
[P1] Preserve the ID-less physical prefix for a missing nested key. With
scan semantics v2, an ID-bearing sibling selects ID mode, but if the
equality-key leaf is absent this helper requires the ID-less wrapper itself to
match the transported wrapper ID and returns an empty path. The caller then
registers the whole key as missing and hashes NULL for every row, including
rows where the physical wrapper is present and the missing child should take
its initial default; an equality delete for that default is silently missed.
Please use the ordinary mapper's unique ID-less-wrapper/ID-bearing-descendant
fallback here and in the ORC helper, and cover a missing leaf below a nullable
ID-less wrapper that retains another ID-bearing child.
##########
fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/write/ConnectorWritePlanProvider.java:
##########
@@ -41,6 +42,25 @@
*/
public interface ConnectorWritePlanProvider {
+ /**
+ * Returns the statement-pinned data columns for a write target.
+ *
+ * <p>A connector whose remote schema can evolve concurrently may override
this method to resolve the
+ * target schema once before analysis. The engine uses the returned
columns for omitted-column and
+ * explicit-DEFAULT expansion and then keeps the same statement scope
through sink planning and commit.
+ * {@link Optional#empty()} preserves the cached table-schema behavior for
connectors that do not need
+ * request-scoped write metadata.</p>
+ *
+ * @param session the current session
+ * @param tableHandle the target table handle
+ * @param branchName the named write branch, if any
+ * @return pinned write columns, or empty to use the table's cached schema
+ */
+ default Optional<List<ConnectorColumn>> getWriteColumns(ConnectorSession
session,
Review Comment:
[P1] Version this connector SPI expansion. This public method and
`ConnectorColumnHandle.withProjectedFieldIds` change the shared plugin surface,
but the PR leaves `connector.plugin.api.version`, the pinned 3.0 assertion, and
both recorded baselines unchanged. The connector compatibility contract
explicitly treats any shared SPI method addition as a MAJOR change; these
returned provider/handle types also are not in the current frozen-type list, so
the drift escapes the gate. Please bump the API to 4.0, refresh both surfaces,
and extend the surface test to freeze connector-returned SPI types.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -1728,57 +1758,437 @@ private static List<String>
requestedLowerNames(List<ConnectorColumnHandle> colu
return names;
}
+ private static boolean mayHaveEqualityDeletes(Snapshot snapshot) {
+ if (snapshot == null) {
+ return false;
+ }
+ return mayHaveEqualityDeletes(snapshot.summary());
+ }
+
+ @VisibleForTesting
+ static boolean mayHaveEqualityDeletes(Map<String, String> snapshotSummary)
{
+ String equalityDeletes = snapshotSummary.get(TOTAL_EQUALITY_DELETES);
+ // A missing counter is unknown (replace/cherry-pick snapshots can
omit it), so retain the bounded
+ // schema-history carrier. The exact task/delete binding is still
decided by Iceberg during split planning.
+ return equalityDeletes == null || !equalityDeletes.equals("0");
+ }
+
/**
- * 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);
+ private static List<NestedField> schemaForPotentialEqualityDeletes(
+ Table table, TableScan scan, Schema scanSchema) {
+ List<Schema> history = potentialEqualityDeleteSchemaHistory(table,
scan, scanSchema);
+ Set<Integer> missing = new HashSet<>();
+ for (Schema schema : history) {
+ 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());
+ for (Schema historicalSchema : history) {
+ addHistoricalEqualityFields(fields, missing, historicalSchema);
+ }
+ if (!missing.isEmpty()) {
+ throw new IllegalStateException(
+ "Iceberg historical primitive fields are absent from
schema history: " + missing);
+ }
+ return fields;
+ }
+
+ private static List<Schema> potentialEqualityDeleteSchemaHistory(
+ Table table, TableScan scan, Schema scanSchema) {
+ List<Schema> history = new ArrayList<>();
+ Set<Integer> seenSchemaIds = new HashSet<>();
+ addSchemaIfAbsent(history, seenSchemaIds, scanSchema);
+
+ Snapshot snapshot = scan.snapshot();
+ while (snapshot != null) {
+ Integer schemaId = snapshot.schemaId();
+ if (schemaId != null) {
+ Schema historicalSchema = table.schemas().get(schemaId);
+ if (historicalSchema == null) {
+ throw new IllegalStateException(
+ "Iceberg snapshot schema " + schemaId + " is
absent from table metadata");
+ }
+ addSchemaIfAbsent(history, seenSchemaIds, historicalSchema);
+ }
+ Long parentId = snapshot.parentId();
+ snapshot = parentId == null ? null : table.snapshot(parentId);
+ }
+
+ 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;
+ for (int index = lastRelevantIndex; index >= 0; index--) {
+ addSchemaIfAbsent(history, seenSchemaIds,
metadataSchemas.get(index));
+ }
+ return history;
+ }
+
+ private static void addSchemaIfAbsent(
+ List<Schema> schemas, Set<Integer> seenSchemaIds, Schema schema) {
+ if (seenSchemaIds.add(schema.schemaId())) {
+ schemas.add(schema);
+ }
+ }
+
+ 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 mayHaveEqualityDeletes,
+ Optional<Map<Integer, List<String>>> nameMapping) {
+ if (mayHaveEqualityDeletes) {
+ 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;
+ }
+ Optional<List<Schema>> history = requiredFieldSchemaHistory(table,
scanSchema, scan.snapshot());
+ return !history.isPresent()
+ || requiresMissingRequiredFieldRejection(scanSchema,
projectedFieldIds, history.get());
+ }
+
+ @VisibleForTesting
+ static Set<Integer> projectedFieldIds(
+ Schema scanSchema, List<ConnectorColumnHandle> columns) {
+ Set<Integer> projected = new HashSet<>();
+ if (columns == null || columns.isEmpty()) {
+
projected.addAll(TypeUtil.indexById(scanSchema.asStruct()).keySet());
+ return projected;
+ }
+ Map<Integer, NestedField> fieldsById =
TypeUtil.indexById(scanSchema.asStruct());
+ for (ConnectorColumnHandle column : columns) {
+ IcebergColumnHandle icebergColumn = (IcebergColumnHandle) column;
+ NestedField field =
scanSchema.findField(icebergColumn.getFieldId());
if (field == null) {
continue;
}
- String lower = field.name().toLowerCase(Locale.ROOT);
- if (present.add(lower)) {
- result.add(lower);
+ if (icebergColumn.hasProjectedFieldIds()) {
+ Set<Integer> scopedFieldIds =
icebergColumn.getProjectedFieldIds();
+ Set<Integer> selectableFieldIds = new HashSet<>();
+ selectableFieldIds.add(field.fieldId());
+ if (field.type().isNestedType()) {
+
selectableFieldIds.addAll(TypeUtil.getProjectedIds(field.type()));
+ }
+ for (Integer scopedFieldId : scopedFieldIds) {
+ NestedField scopedField = fieldsById.get(scopedFieldId);
+ if (scopedField == null ||
!selectableFieldIds.contains(scopedFieldId)) {
+ throw new IllegalStateException("Projected Iceberg
field ID " + scopedFieldId
+ + " does not belong to top-level field " +
field.fieldId());
+ }
+ projected.add(scopedFieldId);
+ if (scopedField.type().isNestedType()) {
+ Set<Integer> descendants =
TypeUtil.getProjectedIds(scopedField.type());
+ if (Collections.disjoint(scopedFieldIds, descendants))
{
+ // The access path terminates at this complex
field, so the entire subtree is
+ // projected. An ancestor with a selected
descendant must remain scoped instead.
+ projected.addAll(descendants);
+ }
+ }
+ }
+ continue;
+ }
+ projected.add(field.fieldId());
+ // Iceberg's type visitor returns null for a primitive root;
getProjectedIds(Type) then passes
+ // that null to ImmutableSet.copyOf. The top-level id is already
present, and only nested types
+ // have descendant ids to add.
+ if (field.type().isNestedType()) {
+ projected.addAll(TypeUtil.getProjectedIds(field.type()));
}
}
- return result;
+ return projected;
}
- private static boolean hasEqualityDeletes(Table table) {
- Snapshot snapshot = table.currentSnapshot();
- if (snapshot == null) {
+ private static Optional<List<Schema>> requiredFieldSchemaHistory(
+ Table table, Schema scanSchema, Snapshot selectedSnapshot) {
+ List<Schema> schemas = new ArrayList<>();
+ Set<Integer> schemaIds = new HashSet<>();
+ schemas.add(scanSchema);
+ schemaIds.add(scanSchema.schemaId());
+ Deque<Snapshot> snapshots = new ArrayDeque<>();
+ if (selectedSnapshot != null) {
+ snapshots.add(selectedSnapshot);
+ }
+ Set<Long> visitedSnapshotIds = new HashSet<>();
+ while (!snapshots.isEmpty()) {
Review Comment:
[P2] Avoid walking every retained snapshot on every scan.
`getScanNodeProperties` reaches this loop for ordinary no-equality-delete
queries after the cheap checks, and the loop continues through the entire
parent/source ancestry even when every snapshot has the same schema and no
projected required field can need historical rejection. The equality-delete
carrier has the same snapshot-count walk in the other branch. A streaming table
with 100k retained snapshots therefore pays 100k lookups and visited-ID
allocations per query before split planning. Please bound this by relevant
schema IDs/fields (and stop once they are resolved), with a
many-snapshots/one-schema counter 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]
