github-actions[bot] commented on code in PR #65851:
URL: https://github.com/apache/doris/pull/65851#discussion_r3697924550
##########
be/src/format/transformer/vorc_transformer.cpp:
##########
@@ -576,11 +611,22 @@ Status VOrcTransformer::write(const Block& block) {
try {
DataTypeSerDe::FormatOptions options;
options.timezone = &_state->timezone_obj();
+ Columns normalized_columns;
+ normalized_columns.reserve(block.columns());
for (size_t i = 0; i < block.columns(); i++) {
const auto& col = block.get_by_position(i);
- const auto& raw_column = col.column;
- RETURN_IF_ERROR(_resize_row_batch(col.type, *raw_column,
root->fields[i]));
-
RETURN_IF_ERROR(_serdes[i]->write_column_to_orc(_state->timezone(), *raw_column,
+ ColumnPtr raw_column = col.column;
+ if (_iceberg_schema != nullptr) {
+ DORIS_CHECK(i <
_iceberg_schema->root_struct().fields().size());
+ raw_column = raw_column->convert_to_full_column_if_const();
Review Comment:
[P2] Skip ORC normalization for unaffected Iceberg fields
This expands every constant and recursively normalizes every column whenever
an Iceberg schema is present, even if the field subtree contains no UUID or
FIXED type. Ordinary nullable fields clone their full null maps, ARRAY/MAP
fields clone offsets, and complex wrappers are rebuilt on every batch; constant
defaults that the previous SerDe accepted directly are also expanded to `rows`.
Wide, nullable Iceberg ORC writes therefore gain avoidable O(rows x fields) CPU
and memory traffic. Please precompute whether each field subtree actually needs
UUID/FIXED normalization and pass unaffected columns through unchanged,
materializing constants only when conversion requires it.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriteSchemaContext.java:
##########
@@ -0,0 +1,473 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.connector.iceberg;
+
+import org.apache.doris.connector.api.ConnectorColumn;
+import org.apache.doris.connector.api.ConnectorType;
+import org.apache.doris.connector.api.DorisConnectorException;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.io.BaseEncoding;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.PartitionSpecParser;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SchemaParser;
+import org.apache.iceberg.SortField;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.SortOrderParser;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.transforms.Transforms;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.time.LocalDate;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.UUID;
+
+/**
+ * Statement-scoped Iceberg writer metadata.
+ *
+ * <p>The connector resolves this context before the engine expands omitted
columns or DEFAULT. Analysis,
+ * sink construction and commit validation then consume the same
schema/spec/order/format snapshot, preventing
+ * a concurrent metadata change from combining expressions from one schema
with files described by another.
+ * Cached table columns deliberately do not carry Iceberg write defaults; only
{@link #getColumns()} does, so
+ * DESCRIBE and SHOW CREATE remain metadata-only while a write still receives
typed defaults.</p>
+ */
+final class IcebergWriteSchemaContext {
+
+ private final String tableName;
+ private final Schema schema;
+ private final int formatVersion;
+ private final Optional<String> branchName;
+ private final Optional<UUID> tableUuid;
+ private final Optional<String> v1MetadataFileLocation;
+ private final Optional<Long> v1MetadataTimestampMillis;
+ private final String schemaJson;
+ private final Schema mergeSchema;
+ private final String mergeSchemaJson;
+ private final PartitionSpec partitionSpec;
+ private final String partitionSpecJson;
+ private final SortOrder sortOrder;
+ private final String sortOrderJson;
+ private final FileFormat fileFormat;
+ private final MetricsConfig metricsConfig;
+ private final String fileCompression;
+ private final String dataLocation;
+ private final Map<String, String> writerProperties;
+ private final List<ConnectorColumn> columns;
+
+ static IcebergWriteSchemaContext create(Table table, String tableName,
+ Optional<String> branchName, boolean enableMappingVarbinary,
+ boolean enableMappingTimestampTz) {
+ Objects.requireNonNull(table, "table should not be null");
+ Objects.requireNonNull(tableName, "tableName should not be null");
+ Objects.requireNonNull(branchName, "branchName should not be null");
+ // Iceberg schema evolution is table-global. A branch selects the
snapshot lineage that receives
+ // the commit, while files written by that commit use the table's
current schema.
+ Schema schema = table.schema();
+ int formatVersion = IcebergWriterHelper.getFormatVersion(table);
+ TableIdentity identity = pinTableIdentity(table, formatVersion);
+ return new IcebergWriteSchemaContext(
+ tableName, schema, formatVersion, branchName, identity.uuid,
+ identity.v1MetadataFileLocation,
identity.v1MetadataTimestampMillis,
+ bindPartitionSpec(table.spec(), schema, tableName),
+ bindSortOrder(table.sortOrder(), schema, tableName),
+ IcebergWriterHelper.getFileFormat(table),
MetricsConfig.forTable(table),
+ IcebergWritePlanProvider.getFileCompress(table),
+ IcebergWritePlanProvider.dataLocation(table),
+ ImmutableMap.copyOf(table.properties()),
+ enableMappingVarbinary, enableMappingTimestampTz);
+ }
+
+ private IcebergWriteSchemaContext(String tableName, Schema schema, int
formatVersion,
+ Optional<String> branchName, Optional<UUID> tableUuid,
+ Optional<String> v1MetadataFileLocation, Optional<Long>
v1MetadataTimestampMillis,
+ PartitionSpec partitionSpec, SortOrder sortOrder, FileFormat
fileFormat,
+ MetricsConfig metricsConfig, String fileCompression, String
dataLocation,
+ Map<String, String> writerProperties, boolean
enableMappingVarbinary,
+ boolean enableMappingTimestampTz) {
+ this.tableName = Objects.requireNonNull(tableName, "tableName should
not be null");
+ this.schema = Objects.requireNonNull(schema, "schema should not be
null");
+ this.formatVersion = formatVersion;
+ this.branchName = Objects.requireNonNull(branchName, "branchName
should not be null");
+ this.tableUuid = Objects.requireNonNull(tableUuid, "tableUuid should
not be null");
+ this.v1MetadataFileLocation = Objects.requireNonNull(
+ v1MetadataFileLocation, "v1MetadataFileLocation should not be
null");
+ this.v1MetadataTimestampMillis = Objects.requireNonNull(
+ v1MetadataTimestampMillis, "v1MetadataTimestampMillis should
not be null");
+ Preconditions.checkState(v1MetadataFileLocation.isPresent() ==
v1MetadataTimestampMillis.isPresent(),
+ "Iceberg V1 metadata identity must contain both location and
timestamp");
+ Preconditions.checkState(!tableUuid.isPresent() ||
!v1MetadataFileLocation.isPresent(),
+ "Iceberg table identity cannot contain both UUID and V1
metadata");
+ this.schemaJson = SchemaParser.toJson(schema);
+ this.mergeSchema = formatVersion >= 3
+ ? IcebergWriterHelper.appendRowLineageFieldsForV3(schema) :
schema;
+ this.mergeSchemaJson = SchemaParser.toJson(mergeSchema);
+ this.partitionSpec = Objects.requireNonNull(partitionSpec,
"partitionSpec should not be null");
+ this.partitionSpecJson = PartitionSpecParser.toJson(partitionSpec);
+ this.sortOrder = Objects.requireNonNull(sortOrder, "sortOrder should
not be null");
+ this.sortOrderJson = SortOrderParser.toJson(sortOrder);
+ this.fileFormat = Objects.requireNonNull(fileFormat, "fileFormat
should not be null");
+ this.metricsConfig = Objects.requireNonNull(metricsConfig,
"metricsConfig should not be null");
+ this.fileCompression = Objects.requireNonNull(fileCompression,
"fileCompression should not be null");
+ this.dataLocation = Objects.requireNonNull(dataLocation, "dataLocation
should not be null");
+ this.writerProperties = ImmutableMap.copyOf(
+ Objects.requireNonNull(writerProperties, "writerProperties
should not be null"));
+ validateWriterMetadataSources(schema, partitionSpec, sortOrder,
tableName);
+
+ ImmutableList.Builder<ConnectorColumn> columnBuilder =
ImmutableList.builder();
+ for (Types.NestedField field : schema.columns()) {
+ ConnectorType type = IcebergTypeMapping.fromIcebergType(
+ field.type(), enableMappingVarbinary,
enableMappingTimestampTz);
+ ConnectorColumn column = new ConnectorColumn(
+ field.name(), type, field.doc() == null ? "" : field.doc(),
+ field.isOptional(), null,
true).withUniqueId(field.fieldId());
+ if (isTimestampWithZone(field.type())) {
+ column = column.withTimeZone();
+ }
+ String defaultSql = field.writeDefault() == null
+ ? (field.isOptional() ? "NULL" : null)
+ : toDorisSql(field.type(), field.writeDefault(),
+ enableMappingVarbinary, enableMappingTimestampTz);
+ if (defaultSql != null) {
+ column = column.withDefaultValueSql(defaultSql);
+ }
+ columnBuilder.add(column);
+ }
+ this.columns = columnBuilder.build();
+ }
+
+ private static PartitionSpec bindPartitionSpec(
+ PartitionSpec partitionSpec, Schema schema, String tableName) {
+ if (!partitionSpec.isPartitioned()) {
+ return
PartitionSpec.builderFor(schema).withSpecId(partitionSpec.specId()).build();
+ }
+ try {
+ return PartitionSpecParser.fromJson(schema,
PartitionSpecParser.toJson(partitionSpec));
+ } catch (RuntimeException e) {
+ throw new DorisConnectorException("Iceberg partition spec " +
partitionSpec.specId()
+ + " is incompatible with pinned schema " +
schema.schemaId()
+ + " for table " + tableName + ": " + e.getMessage(), e);
+ }
+ }
+
+ private static SortOrder bindSortOrder(SortOrder sortOrder, Schema schema,
String tableName) {
+ if (!sortOrder.isSorted()) {
+ return SortOrder.unsorted();
+ }
+ try {
+ return SortOrderParser.fromJson(schema,
SortOrderParser.toJson(sortOrder));
+ } catch (RuntimeException e) {
+ throw new DorisConnectorException("Iceberg sort order " +
sortOrder.orderId()
+ + " is incompatible with pinned schema " +
schema.schemaId()
+ + " for table " + tableName + ": " + e.getMessage(), e);
+ }
+ }
+
+ private static void validateWriterMetadataSources(
+ Schema schema, PartitionSpec partitionSpec, SortOrder sortOrder,
String tableName) {
+ Map<Integer, Types.NestedField> topLevelFields =
schema.columns().stream()
+
.collect(ImmutableMap.toImmutableMap(Types.NestedField::fieldId, field ->
field));
+ for (PartitionField field : partitionSpec.fields()) {
+ if (!topLevelFields.containsKey(field.sourceId())) {
+ throw new DorisConnectorException("Iceberg partition field " +
field.fieldId()
+ + " references source field " + field.sourceId()
+ + " outside pinned top-level schema " +
schema.schemaId()
+ + " for table " + tableName);
+ }
+ }
+ for (SortField field : sortOrder.fields()) {
+ if (schema.findField(field.sourceId()) == null) {
+ throw new DorisConnectorException("Iceberg sort field
references source field "
+ + field.sourceId() + " outside pinned schema " +
schema.schemaId()
+ + " for table " + tableName);
+ }
+ }
+ }
+
+ void validateCurrentSchema(Table table, boolean
requireCurrentPartitionSpec) {
+ Schema currentSchema = table.schema();
+ int currentFormatVersion = IcebergWriterHelper.getFormatVersion(table);
+ validateTableIdentity(table, currentFormatVersion);
+ if (currentSchema.schemaId() != schema.schemaId() ||
currentFormatVersion != formatVersion) {
+ throw new DorisConnectorException("Iceberg table schema changed
during write planning for "
+ + tableName + ": pinned schema " + schema.schemaId() +
"/format " + formatVersion
+ + ", current schema " + currentSchema.schemaId() +
"/format " + currentFormatVersion
+ + "; retry the statement");
+ }
+ PartitionSpec retainedSpec = table.specs().get(partitionSpec.specId());
+ if (retainedSpec == null
+ ||
!partitionSpecJson.equals(PartitionSpecParser.toJson(retainedSpec))) {
+ throw new DorisConnectorException("Iceberg partition spec changed
during write planning for "
+ + tableName + ": pinned spec " + partitionSpec.specId()
+ + " is not available with the same definition; retry the
statement");
+ }
+ if (requireCurrentPartitionSpec) {
+ PartitionSpec activeSpec = table.spec();
+ if (activeSpec.specId() != partitionSpec.specId()
+ ||
!partitionSpecJson.equals(PartitionSpecParser.toJson(activeSpec))) {
+ throw new DorisConnectorException("Iceberg current partition
spec changed during overwrite "
+ + "planning for " + tableName + ": pinned spec " +
partitionSpec.specId()
+ + ", current spec " + activeSpec.specId() + "; retry
the statement");
+ }
+ }
+ SortOrder retainedSortOrder =
table.sortOrders().get(sortOrder.orderId());
+ if (retainedSortOrder == null
+ ||
!sortOrderJson.equals(SortOrderParser.toJson(retainedSortOrder))) {
+ throw new DorisConnectorException("Iceberg sort order changed
during write planning for "
+ + tableName + ": pinned order " + sortOrder.orderId()
+ + " is not available with the same definition; retry the
statement");
+ }
+ }
+
+ private static TableIdentity pinTableIdentity(Table table, int
formatVersion) {
+ if (table instanceof HasTableOperations) {
+ TableMetadata metadata = Preconditions.checkNotNull(
+ ((HasTableOperations) table).operations().current(),
+ "Iceberg table %s has no current metadata", table.name());
+ if (metadata.uuid() != null) {
+ return TableIdentity.forUuid(UUID.fromString(metadata.uuid()));
+ }
+ Preconditions.checkState(formatVersion == 1,
+ "Iceberg table %s format %s has no table UUID",
table.name(), formatVersion);
+ return TableIdentity.forV1Metadata(
+ Preconditions.checkNotNull(metadata.metadataFileLocation(),
+ "Iceberg V1 table %s has no metadata file
location", table.name()),
+ metadata.lastUpdatedMillis());
+ }
+ return TableIdentity.forUuid(Preconditions.checkNotNull(
+ table.uuid(), "Iceberg table %s does not expose a table UUID",
table.name()));
+ }
+
+ private void validateTableIdentity(Table table, int currentFormatVersion) {
+ if (tableUuid.isPresent()) {
+ if (!tableUuid.equals(pinTableIdentity(table,
currentFormatVersion).uuid)) {
+ throw tableIdentityChanged();
+ }
+ return;
+ }
+ if (!v1MetadataFileLocation.isPresent()) {
+ return;
+ }
+ Preconditions.checkState(table instanceof HasTableOperations,
+ "Iceberg V1 table %s does not expose table operations",
table.name());
+ TableMetadata currentMetadata = Preconditions.checkNotNull(
+ ((HasTableOperations) table).operations().current(),
+ "Iceberg V1 table %s has no current metadata", table.name());
+ boolean sameMetadata =
v1MetadataFileLocation.get().equals(currentMetadata.metadataFileLocation())
+ && v1MetadataTimestampMillis.get() ==
currentMetadata.lastUpdatedMillis();
+ boolean retainedAncestor = currentMetadata.previousFiles().stream()
+ .anyMatch(entry ->
v1MetadataFileLocation.get().equals(entry.file())
+ && v1MetadataTimestampMillis.get() ==
entry.timestampMillis());
+ if (!sameMetadata && !retainedAncestor) {
+ throw tableIdentityChanged();
+ }
+ }
+
+ private DorisConnectorException tableIdentityChanged() {
+ return new DorisConnectorException("Iceberg table identity changed
during write planning for "
+ + tableName + "; the table may have been dropped and
recreated; retry the statement");
+ }
+
+ private static boolean isTimestampWithZone(Type type) {
+ return type.isPrimitiveType() && type.typeId() == Type.TypeID.TIMESTAMP
+ && ((Types.TimestampType) type).shouldAdjustToUTC();
+ }
+
+ private static String toDorisSql(Type type, Object value,
+ boolean enableMappingVarbinary, boolean enableMappingTimestampTz) {
+ if (value == null) {
+ return "NULL";
+ }
+ switch (type.typeId()) {
+ case BOOLEAN:
+ case INTEGER:
+ case LONG:
+ case FLOAT:
+ case DOUBLE:
+ return String.valueOf(value);
+ case DECIMAL:
+ return ((BigDecimal) value).toPlainString();
+ case STRING:
+ return quote((String) value);
+ case UUID:
+ return binarySql(uuidBytes((UUID) value),
enableMappingVarbinary);
+ case FIXED:
+ case BINARY:
+ return binarySql(byteBufferBytes((ByteBuffer) value),
enableMappingVarbinary);
+ case DATE:
+ return quote(LocalDate.ofEpochDay(((Integer)
value).longValue()).toString());
+ case TIMESTAMP:
+ String timestamp =
Transforms.identity(type).toHumanString(type, value).replace('T', ' ');
+ if (((Types.TimestampType) type).shouldAdjustToUTC() &&
!enableMappingTimestampTz) {
+ timestamp =
timestamp.replaceFirst("(Z|[+-]\\d{2}:\\d{2})$", "");
+ }
+ return quote(timestamp);
+ case LIST:
+ Types.ListType listType = (Types.ListType) type;
+ List<String> items = new ArrayList<>();
+ for (Object item : (List<?>) value) {
+ items.add(toDorisSql(listType.elementType(), item,
+ enableMappingVarbinary, enableMappingTimestampTz));
+ }
+ return "array(" + String.join(", ", items) + ")";
+ case MAP:
+ Types.MapType mapType = (Types.MapType) type;
+ List<String> entries = new ArrayList<>();
+ for (Map.Entry<?, ?> entry : ((Map<?, ?>) value).entrySet()) {
+ entries.add(toDorisSql(mapType.keyType(), entry.getKey(),
+ enableMappingVarbinary, enableMappingTimestampTz));
+ entries.add(toDorisSql(mapType.valueType(),
entry.getValue(),
+ enableMappingVarbinary, enableMappingTimestampTz));
+ }
+ return "map(" + String.join(", ", entries) + ")";
+ case STRUCT:
+ Types.StructType structType = (Types.StructType) type;
+ StructLike struct = (StructLike) value;
+ List<String> fields = new ArrayList<>();
+ for (int i = 0; i < structType.fields().size(); i++) {
+ Types.NestedField field = structType.fields().get(i);
+ fields.add(quote(field.name()));
+ fields.add(toDorisSql(field.type(), struct.get(i,
Object.class),
+ enableMappingVarbinary, enableMappingTimestampTz));
+ }
+ return "named_struct(" + String.join(", ", fields) + ")";
+ default:
+ throw new DorisConnectorException("Unsupported Iceberg
write-default type: " + type);
+ }
+ }
+
+ private static String binarySql(byte[] bytes, boolean
enableMappingVarbinary) {
+ String hex = BaseEncoding.base16().encode(bytes);
+ return enableMappingVarbinary ? "X'" + hex + "'" : "UNHEX('" + hex +
"')";
+ }
+
+ private static String quote(String value) {
+ return "'" + value.replace("'", "''") + "'";
Review Comment:
[P1] Preserve backslashes in Iceberg write defaults
`quote()` escapes apostrophes but leaves backslashes verbatim, while every
consumer reparses this text with Nereids. Under the default SQL mode,
`SqlLiteralUtils.unescapeBackSlash()` turns sequences such as `\n` and `\t`
into control characters and drops the slash from unknown escapes, so a valid
Iceberg string default such as the literal characters `C:\new` is silently
written as a different value for omitted columns and explicit `DEFAULT`
(including nested strings). The added test covers only `O'Reilly` and never
round-trips through the parser. Please preserve the typed Iceberg value as the
design requires, or use a session-aware literal encoding that round-trips under
both backslash modes, and add parser/execution coverage for backslash defaults.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTable.java:
##########
@@ -779,6 +779,30 @@ public List<Column> getSyntheticWriteColumns() {
return
ConnectorColumnConverter.convertColumns(fetchSyntheticWriteColumns());
}
+ /**
+ * Resolves request-scoped data columns for write analysis. A connector
may pin remote schema/default
+ * metadata here; an empty result tells the engine to keep using the
cached table schema.
+ */
+ public Optional<List<Column>> resolveWriteColumns(Optional<String>
branchName) {
+ PluginDrivenExternalCatalog pluginCatalog =
(PluginDrivenExternalCatalog) catalog;
+ Connector connector = pluginCatalog.getConnector();
+ if (connector == null) {
+ return Optional.empty();
+ }
+ ConnectorSession session = pluginCatalog.buildConnectorSession();
+ ConnectorMetadata metadata = PluginDrivenMetadata.get(session,
connector);
+ Optional<ConnectorTableHandle> handle =
resolveConnectorTableHandle(session, metadata);
+ if (!handle.isPresent()) {
+ return Optional.empty();
+ }
+ ConnectorWritePlanProvider provider =
connector.getWritePlanProvider(handle.get());
+ if (provider == null) {
+ return Optional.empty();
+ }
+ return provider.getWriteColumns(session, handle.get(), branchName)
Review Comment:
[P1] Pin the plugin loader around getWriteColumns
This new provider callback crosses from fe-core into a child-first directory
plugin with the request thread's existing TCCL. A connector that performs
by-name reflection/service loading here—or creates a worker that inherits the
TCCL—can therefore resolve the app copy of a dependency and fail with
`ClassNotFoundException`/split-brain `ClassCastException`. This violates the
explicit TCCL-at-plugin-boundaries invariant; analogous provider callbacks in
`PluginDrivenSysExternalTable` and `PluginDrivenScanNode` install
`provider.getClass().getClassLoader()` and restore it in `finally`. Please wrap
`getWriteColumns` with the same engine-side guard and add a directory-plugin
test that asserts the callback TCCL.
##########
be/src/format_v2/table/iceberg_reader.cpp:
##########
@@ -80,61 +91,703 @@ static bool is_projected_iceberg_rowid(const
format::ColumnDefinition& column) {
return column.name == BeConsts::ICEBERG_ROWID_COL;
}
+static int iceberg_hex_value(char value) {
+ if (value >= '0' && value <= '9') {
+ return value - '0';
+ }
+ if (value >= 'a' && value <= 'f') {
+ return value - 'a' + 10;
+ }
+ if (value >= 'A' && value <= 'F') {
+ return value - 'A' + 10;
+ }
+ return -1;
+}
+
+static Status decode_iceberg_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 = iceberg_hex_value(encoded[index]);
+ const int low = iceberg_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();
+}
+
+static Status decode_iceberg_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) {
+ return decode_iceberg_hex(encoded, decoded);
+ }
+
+ 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_iceberg_hex(uuid_hex, decoded);
+}
+
+static std::string iceberg_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()};
+}
+
+static void normalize_iceberg_json_timestamp(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);
+ }
+}
+
+static Status build_v2_null_default(const format::ColumnDefinition& field,
+ const DataTypePtr& data_type, Field*
result) {
+ DORIS_CHECK(data_type != nullptr);
+ DORIS_CHECK(result != nullptr);
+ if (field.is_optional.has_value() && !*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();
+}
+
+static const format::ColumnDefinition* find_v2_struct_child(const
format::ColumnDefinition& field,
+ const std::string&
name) {
+ const auto exact_child = std::ranges::find_if(
+ field.children, [&](const auto& candidate) { return
iequal(candidate.name, name); });
+ if (exact_child != field.children.end()) {
+ return &*exact_child;
+ }
+ const auto aliased_child = std::ranges::find_if(field.children, [&](const
auto& candidate) {
+ return std::ranges::any_of(candidate.name_mapping,
+ [&](const auto& alias) { return
iequal(alias, name); });
+ });
+ return aliased_child == field.children.end() ? nullptr : &*aliased_child;
+}
+
+static Status build_v2_initial_default_field(const format::ColumnDefinition&
field,
+ const DataTypePtr& data_type,
+ std::deque<std::string>*
binary_storage,
+ Field* result);
+
+static Status build_v2_json_default_field(const format::ColumnDefinition&
field,
+ const DataTypePtr& data_type,
+ const rapidjson::Value& json_value,
+ std::deque<std::string>*
binary_storage, Field* result);
+
+static Status build_v2_json_struct_default(const format::ColumnDefinition&
field,
+ const DataTypePtr& value_type,
+ const rapidjson::Value& json_value,
+ std::deque<std::string>*
binary_storage, Field* result) {
+ if (!json_value.IsObject()) {
+ 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 = find_v2_struct_child(field,
struct_type.get_element_name(index));
+ if (child == nullptr || !child->has_identifier_field_id()) {
+ return Status::InvalidArgument(
+ "Iceberg struct default for field '{}' has incomplete
child metadata",
+ field.name);
+ }
+
+ const std::string child_id =
std::to_string(child->get_identifier_field_id());
+ const auto member = json_value.FindMember(child_id.c_str());
+ Field child_value;
+ if (member == json_value.MemberEnd()) {
+ RETURN_IF_ERROR(build_v2_initial_default_field(*child,
struct_type.get_element(index),
+ binary_storage,
&child_value));
+ } else {
+ RETURN_IF_ERROR(build_v2_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 child ColumnDefinition, recursively transported from the 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.
+static Status build_v2_json_array_default(const format::ColumnDefinition&
field,
+ const DataTypePtr& value_type,
+ const rapidjson::Value& json_value,
+ std::deque<std::string>*
binary_storage, Field* result) {
+ if (!json_value.IsArray() || field.children.size() != 1) {
+ return Status::InvalidArgument("Invalid Iceberg list default for field
'{}'", 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_v2_json_default_field(field.children.front(),
+
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 child ColumnDefinitions, recursively transported from the 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.
+static Status build_v2_json_map_default(const format::ColumnDefinition& 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.children.size() != 2) {
+ 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& 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_v2_json_default_field(field.children[0],
map_type.get_key_type(),
+ keys[index],
binary_storage, &key_value));
+ RETURN_IF_ERROR(build_v2_json_default_field(field.children[1],
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();
+}
+
+static Status build_v2_json_scalar_default(const format::ColumnDefinition&
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 = iceberg_json_scalar_text(json_value);
+ const bool binary_like =
+ 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_iceberg_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_iceberg_json_timestamp(primitive_type, &serialized_value);
+ RETURN_IF_ERROR(value_type->get_serde()->from_fe_string(serialized_value,
*result));
+ return Status::OK();
+}
+
+static Status build_v2_json_default_field(const format::ColumnDefinition&
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 build_v2_null_default(field, data_type, result);
+ }
+
+ const auto value_type = remove_nullable(data_type);
+ switch (value_type->get_primitive_type()) {
+ case TYPE_STRUCT:
+ return build_v2_json_struct_default(field, value_type, json_value,
binary_storage, result);
+ case TYPE_ARRAY:
+ return build_v2_json_array_default(field, value_type, json_value,
binary_storage, result);
+ case TYPE_MAP:
+ return build_v2_json_map_default(field, value_type, json_value,
binary_storage, result);
+ default:
+ return build_v2_json_scalar_default(field, value_type, json_value,
binary_storage, result);
+ }
+}
+
+static Status build_v2_initial_default_field(const format::ColumnDefinition&
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.initial_default_value.has_value()) {
+ if (field.is_optional.has_value() && !*field.is_optional) {
+ return Status::InvalidArgument(
+ "Required Iceberg field '{}' is missing from the data file
and has no initial "
+ "default",
+ field.name);
+ }
+ return build_v2_null_default(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);
+ }
+ if (primitive_type == TYPE_STRUCT &&
+ (!document.IsObject() || document.MemberCount() != 0)) {
+ return Status::InvalidArgument(
+ "Iceberg struct field '{}' has a non-empty initial
default", field.name);
+ }
+ return build_v2_json_default_field(field, data_type, document,
binary_storage, result);
+ }
+
+ if (field.initial_default_value_is_base64 || primitive_type ==
TYPE_VARBINARY) {
+ 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(
+ "Base64 Iceberg initial default has incompatible Doris
type {} for field {}",
+ data_type->get_name(), field.name);
+ }
+ return Status::OK();
+ }
+
+
RETURN_IF_ERROR(value_type->get_serde()->from_fe_string(*field.initial_default_value,
*result));
+ return Status::OK();
+}
+
+static Status build_initial_default_literal(const format::ColumnDefinition&
table_field,
+ VExprSPtr* literal) {
+ DORIS_CHECK(table_field.type != nullptr);
+ DORIS_CHECK(table_field.initial_default_value.has_value());
+ DORIS_CHECK(literal != nullptr);
+
+ std::deque<std::string> binary_storage;
+ Field initial_default;
+ RETURN_IF_ERROR(build_v2_initial_default_field(table_field,
table_field.type, &binary_storage,
+ &initial_default));
+ // VLiteral inserts the Field into an owning column before binary_storage
is destroyed.
+ *literal = VLiteral::create_shared(table_field.type, initial_default);
+ return Status::OK();
+}
+
+static Status build_initial_default_exprs(format::ColumnDefinition* column) {
+ DORIS_CHECK(column != nullptr);
+ if (column->initial_default_value.has_value()) {
+ VExprSPtr literal;
+ RETURN_IF_ERROR(build_initial_default_literal(*column, &literal));
+ column->default_expr = VExprContext::create_shared(std::move(literal));
Review Comment:
[P1] Build V2 defaults with the projected table type
The production schema carrier deliberately tags every primitive `TField` as
`STRING` only to distinguish leaves (`IcebergSchemaUtils.buildField()`), but
`build_schema_column_from_external_field()` restores that tag as the real type.
This new recursion consequently builds a string `VLiteral` here even when the
projected field is INT, DATE, or VARBINARY, and the missing-column path
installs it without any cast—only nullability is aligned. Reading an old file
that lacks such an optional field therefore places a `ColumnString` under
non-string block metadata, leading to an assertion/crash or invalid downstream
interpretation. The new tests miss this because their `TField`s leave `type`
unset, preserving the fallback slot type. Please construct each default with
the authoritative projected/child type (or carry the real scalar type) and add
a V2 test using the exact connector-shaped `type=STRING` carrier.
--
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]
