github-actions[bot] commented on code in PR #65851:
URL: https://github.com/apache/doris/pull/65851#discussion_r3703585304
##########
docs/iceberg-v3-default-values-design.md:
##########
@@ -0,0 +1,221 @@
+# Iceberg V3 Default Values in Doris
+
+## Background
+
+Iceberg V3 defines two different default-value contracts:
+
+- `initial-default` is a read-time value. A reader uses it when a selected
data file was written
+ before a field was added and therefore does not contain that field ID.
+- `write-default` is a write-time value. A writer uses it when a known field
is omitted or when a
+ statement explicitly requests `DEFAULT`.
+
+Adding a field with a default initially records the same value in both
properties. A later default
+update changes only `write-default`. Doris must therefore keep the two
concepts separate instead
+of representing both as one generic column default.
+
+The implementation has three ordered phases: read `initial-default`, consume
`write-default`, and
+author/evolve default metadata. The current pull request combines the first
two execution phases so
+that Doris can both read old files and write new rows with the correct
defaults. Metadata authoring
+remains a later pull request.
+
+## Spark-Iceberg reference model
+
+The executable reference is Spark 4.0.0 with Iceberg 1.10.1, matching the
repository Docker
+fixture. Iceberg keeps defaults as typed values in `NestedField`; its Spark
Parquet reader converts
+the typed `initialDefault` to Spark's internal representation when a projected
field is absent.
+The Iceberg schema-update API also preserves the distinction between adding a
field with its
+initial default and later updating only its write default.
+
+Iceberg 1.10.1's ORC projection still rejects a physically missing field whose
`initialDefault` is
+non-NULL. That is an upstream implementation gap, not the V3 contract Doris
should reproduce.
+Doris therefore uses Spark-Iceberg as the reference for typed default
conversion and implements the
+same field-ID/default semantics independently in both its Parquet and ORC scan
paths.
+
+Doris follows that typed-constant model, with an explicit FE/BE transport
boundary:
+
+1. FE reads `NestedField.initialDefault()` from the query-bound Iceberg schema.
+2. FE serializes primitive non-binary values through Iceberg's type-aware
identity transform and
+ complex values through Iceberg `SingleValueParser` JSON. UUID, FIXED, and
BINARY use Base64 as
+ direct defaults because a Thrift string cannot safely carry arbitrary
bytes; recursive metadata
+ also retains their Iceberg type identity for binary values inside complex
JSON.
+3. Each scanner has its own recursive parser and constructs an owning typed
constant before
+ materializing rows. V1 and V2 do not delegate missing-column
materialization to each other.
+
+This is intentionally different from feeding metadata text to the SQL parser.
In particular,
+strings containing quotes, binary zero bytes, decimals, dates, and timestamps
retain their typed
+value rather than being reinterpreted as SQL syntax.
+
+## Implementation roadmap
+
+### Phase 1: read `initial-default`
+
+The read phase adds support for every Iceberg type already mapped by Doris. It
does not add any new
+Iceberg-to-Doris type mapping.
+
+The reader selects defaults from the schema bound to the query:
+
+| Query | Schema used for `initial-default` |
+| --- | --- |
+| Ordinary current-table read | `table.schema()` |
+| Explicit snapshot/time travel | Schema recorded by the selected snapshot |
+| Explicit branch | Current table schema over the branch-head snapshot lineage
|
+| Explicit tag/ref snapshot | Schema recorded by the selected snapshot |
+
+Using `currentSnapshot.schemaId()` for an ordinary read is incorrect because a
schema-only commit
+can update `table.schema()` without creating a snapshot.
+
+FE transports default metadata recursively by Iceberg field ID through the
existing external
+schema Thrift structure. The metadata includes the serialized value, a
lossless Base64 marker for
+UUID/FIXED/BINARY, and the Iceberg optional/required flag. Nested defaults do
not depend on
+`Column.defaultValue`. The read phase deliberately leaves that generic Doris
field unset for
+Iceberg schema columns so existing INSERT/MERGE code cannot mistake
`initial-default` for
+`write-default`. The write phase uses a separate schema-pinned write-default
path in the same pull
+request. Consequently, `DESC` and `SHOW CREATE TABLE` do not present either
Iceberg value as a
+generic Doris column default.
+
+File Scanner V1 and File Scanner V2 consume the same metadata contract but
implement missing-field
+materialization independently:
+
+- V1 applies defaults in its Parquet and ORC schema-change readers.
+- V2 applies defaults through `ColumnDefinition`, `TableColumnMapper`, and
`TableReader`.
+
+Both implementations follow the same rules:
+
+1. A physically present field is read as stored, including an explicit NULL.
+2. A physically absent field with `initial-default` is materialized as a typed
constant.
+3. A physically absent optional field without a default is NULL.
+4. A physically absent required field without a default is an error.
+5. A NULL parent struct, list element, or map value remains NULL. A nested
default is applied only
+ inside a present parent value.
+6. A non-NULL struct default is the Iceberg V3 empty-object sentinel `{}`; its
effective value is
+ built recursively from the struct fields' own `initial-default` metadata.
List and map defaults
+ use Iceberg's single-value JSON (`[...]` and `{"keys": [...], "values":
[...]}`), including nested
+ primitive and complex values.
+
+Complex-column coverage includes defaults on newly added children of structs,
list-element
+structs, and map-value structs, plus recursive decoding of a whole
complex-field default. Iceberg
+Java 1.10.1's public `Types.NestedField`/`UpdateSchema` path rejects every
non-NULL nested-type
+default, even though the V3 format specification permits the empty struct
sentinel. The Docker
+fixture therefore uses only metadata the official API can commit: it covers
child defaults under
+physical complex parents and wholly missing optional complex parents with NULL
parent defaults.
+Focused V1 and V2 unit tests cover the spec-level `{}`/list/map JSON decoder.
The fixture never
+hand-edits table metadata to bypass Iceberg validation.
+
+Spark-Iceberg Docker generates the fixtures. It writes Parquet and ORC files
using the old schema,
+then uses the Iceberg Java API to add defaulted fields. The fixture changes
`write-default` after
+the add so an old file returning `initial-default` proves that the read path
did not accidentally
+consume the current write default. Regression tests force all four paths:
+
+- File Scanner V1 with Parquet
+- File Scanner V1 with ORC
+- File Scanner V2 with Parquet
+- File Scanner V2 with ORC
+
+The type matrix is exactly the set already supported before this change:
BOOLEAN, INTEGER, LONG, FLOAT,
+DOUBLE, DECIMAL, DATE, TIMESTAMP with and without zone, STRING, UUID, FIXED,
BINARY, and their
+existing ARRAY/MAP/STRUCT compositions. TIME, TIMESTAMP_NANO, VARIANT, and any
other currently
+unsupported mapping remain out of scope.
+
+The read phase verifies predicates for fields absent from old files in
forced-path regression
+tests. Focused V1 and V2 BE tests verify equality-delete keys, including a key
that is retained only
+in schema history after being dropped. Iceberg 1.10.1 cannot plan that
dropped-key sequence end to
+end; its planner fix is upstream Iceberg #15268, so this change does not
broaden scope with an
+Iceberg dependency upgrade. Binary-like defaults are checked byte-for-byte,
and regression output
+is generated only by the Doris regression test runner.
+
+### Phase 2: consume `write-default`
+
+The write phase adds write-time consumption without changing Iceberg default
metadata. It
+introduces a statement-scoped, field-ID keyed context pinned to the target
schema ID and format
+version.
+
+The following write paths consume the statement-pinned `write-default`:
+
+- omitted columns in INSERT
+- explicit `DEFAULT`
+- reordered and multi-row VALUES
+- MERGE `NOT MATCHED INSERT`
+- `DEFAULT(column)` in INSERT, UPDATE, and matched MERGE
+
+Explicit NULL and explicit values are never replaced. A bare `DEFAULT` in an
INSERT position uses
+that destination field's write default, while `DEFAULT(column)` resolves the
referenced field by
+Iceberg field ID. Analyzer projection and the schema sent to the Iceberg sink
must use the same
+schema ID; schema skew fails before data is dispatched.
+
+The write phase does not add CREATE/ALTER syntax and does not author or evolve
default metadata.
+
+### Later PR: author and evolve default metadata
+
+The later PR adds DDL support after the read and write execution paths are
stable:
+
+- CREATE COLUMN DEFAULT records the appropriate initial write default for a
new table schema.
+- ADD COLUMN DEFAULT records equal `initial-default` and `write-default`
values.
+- Adding a required field succeeds only with a valid non-NULL initial default.
+- `ALTER COLUMN ... SET DEFAULT` changes only `write-default`.
+- `ALTER COLUMN ... DROP DEFAULT` removes only `write-default`.
+
+DDL validates the Iceberg format version, converts Doris constant expressions
to typed Iceberg
+literals, and exposes current write defaults through user-visible schema
output. Nested ALTER
+syntax is not expanded beyond syntax Doris already supports when the later DDL
PR is implemented.
+
+## Read-path acceptance criteria
+
+1. Every Iceberg type mapped by Doris before this change has an
`initial-default` read test, and the
+ mapping switch itself is unchanged.
+2. Struct children, list-element struct children, and map-value struct
children receive defaults by
+ field ID in both scanners; focused tests also cover `{}` struct, list, and
map default decoding.
+3. Parent NULL values and explicitly stored child NULL values remain NULL.
+4. UUID, FIXED, and BINARY defaults match their exact bytes in both mapping
modes.
+5. Ordinary reads observe schema-only evolution, while explicit historical
reads use their bound
+ historical schema.
+6. Parquet and ORC fixtures are generated through Spark-Iceberg Docker without
hand-edited Iceberg
+ metadata.
+7. V1 and V2 each pass independent unit tests and forced-path regression tests.
+8. Regression predicates and focused V1/V2 equality-delete tests agree with
the value materialized
+ for an absent field; dropped equality-key metadata is recovered from schema
history.
+9. The read implementation never consumes `write-default` and never stores
either Iceberg default
+ in generic `Column.defaultValue`. DDL implementation, dependency upgrades,
and new type mappings
+ remain out of scope.
+
+## Write-path acceptance criteria
+
+1. Omitted INSERT columns, explicit `DEFAULT`, reordered/multi-row VALUES,
supported
+ `DEFAULT(column)`, and MERGE `NOT MATCHED INSERT` consume the typed
`write-default` from the
+ statement-pinned current table schema. A branch target selects the snapshot
lineage that receives
+ the commit; schema evolution remains table-global, so columns added or
renamed after the branch
Review Comment:
[P2] Align the branch-schema contract with the implementation
The current provider now pins `SnapshotUtil.schemaFor(table, branchHead)`,
and the changed regression intentionally rejects current-only `zone` until the
first successful old-schema branch write advances the ref. These criteria
instead promise that post-branch additions/renames are immediately writable and
that branch choice affects only the commit target. Please update this section
and the read matrix/release note to state the branch-head schema behavior;
otherwise this durable design documents the opposite of the behavior the PR
implements and tests.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -1728,57 +1759,405 @@ private static List<String>
requestedLowerNames(List<ConnectorColumnHandle> colu
return names;
}
+ private static boolean mayHaveEqualityDeletes(Snapshot snapshot) {
+ if (snapshot == null) {
+ return false;
+ }
+ String equalityDeletes =
snapshot.summary().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 Schema 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;
+ }
+
+ 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 new Schema(scanSchema.schemaId(), 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;
+ }
+ for (ConnectorColumnHandle column : columns) {
+ NestedField field = scanSchema.findField(((IcebergColumnHandle)
column).getFieldId());
if (field == null) {
continue;
}
- String lower = field.name().toLowerCase(Locale.ROOT);
- if (present.add(lower)) {
- result.add(lower);
+ 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()));
Review Comment:
[P1] Keep the smooth-upgrade gate projection-aware
`PluginDrivenScanNode` reduces each selected slot to an
`IcebergColumnHandle` containing only the top-level field ID; it drops the
slot's pruned type/access path. This line then expands that field to every
descendant, so `SELECT s.a` is rejected whenever an unselected sibling `s.b`
has an initial default and any source BE remains, even though BE never reads
`b`. The legacy scan path was already fixed for this availability regression.
Please carry the resolved nested projection into this provider and add a
planner-level unselected-sibling test.
##########
be/src/format/table/iceberg_default_value.h:
##########
@@ -0,0 +1,493 @@
+// 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());
Review Comment:
[P1] Send a complete typed complex-default carrier
This decoder assumes the parent `initial_default_value` is
`SingleValueParser` JSON, but FE still sends every non-binary value through
`identity(...).toHumanString()`, so complex defaults do not reach it in the
required form. Legacy STRING/CHAR mapping also decodes binary leaves only when
their child `TField` has `initial_default_value_is_base64`, while FE sets that
flag only if the child itself has a default; list/map leaves normally do not.
The new BE test manually sets the flag, masking production. Please serialize
complex parents as Iceberg JSON, mark every UUID/FIXED/BINARY schema leaf by
type, and add an FE-to-reader 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]
