github-actions[bot] commented on code in PR #65851:
URL: https://github.com/apache/doris/pull/65851#discussion_r3699219186
##########
be/src/format/table/iceberg_reader.cpp:
##########
@@ -268,74 +521,79 @@ Status
IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
all_file_columns_have_field_ids = false;
} else {
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
Review Comment:
[P1] Detect descendant IDs before choosing hidden-key mapping
This mode flag inspects only top-level Parquet fields. With an ID-less
`payload` wrapper whose renamed child still has field ID 2, ordinary projection
enters ID mode via `parquet_subtree_has_field_id()` and correctly binds the
wrapper through that descendant, but the equality-key path enters name mode. It
then misses the old child name (or an authoritative empty name mapping) and
synthesizes NULL/default instead of reading the physical key, so an applicable
equality delete is missed or the scan rejects a required field. Please make the
hidden-key decision recursive like `BuildTableInfoUtil` and add a V1 Parquet
equality-delete test with an ID-less root and renamed ID-bearing child.
##########
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();
Review Comment:
[P1] Keep branch writes on the branch-head schema
This always pins `table.schema()`, but Iceberg 1.10.1's Spark connector
exposes `SnapshotUtil.schemaFor(table, branch)` as the target table schema, and
Doris branch reads do the same. If branch `b` is still on schema A after main
advances to B, Doris now rejects valid A-shaped writes and accepts B-only
columns/defaults that Spark does not expose for that branch. The regression was
flipped to enforce this divergence even though the earlier current-schema
request was explicitly withdrawn after the Spark contract was checked. Please
restore branch-head schema resolution and keep the current-schema compatibility
checks at commit time.
##########
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));
+ }
+ for (auto& child : column->children) {
+ RETURN_IF_ERROR(build_initial_default_exprs(&child));
+ }
+ return Status::OK();
+}
+
static Status build_missing_equality_delete_key_expr(const
format::ColumnDefinition& table_field,
const DataTypePtr&
delete_key_type,
+ bool
require_complete_metadata,
VExprSPtr* key_expr) {
DORIS_CHECK(delete_key_type != nullptr);
DORIS_CHECK(key_expr != nullptr);
if (!table_field.initial_default_value.has_value()) {
+ if (require_complete_metadata && !table_field.is_optional.has_value())
{
+ return Status::InvalidArgument(
+ "Iceberg equality delete field '{}' is missing optionality
metadata",
+ table_field.name);
+ }
+ if (table_field.is_optional.has_value() && !*table_field.is_optional) {
+ return Status::InvalidArgument("Missing required field: {}",
table_field.name);
+ }
// A newly added optional field without an initial default is
logically NULL in older
// files. EqualityDeletePredicate treats NULL == NULL as a match.
*key_expr = VLiteral::create_shared(make_nullable(delete_key_type),
Field());
return Status::OK();
}
VExprSPtr literal;
- if (table_field.initial_default_value_is_base64 ||
- table_field.type->get_primitive_type() == TYPE_VARBINARY) {
- // New FE versions mark every Iceberg UUID/BINARY/FIXED default as
Base64 regardless of its
- // Doris mapping. Keep the VARBINARY fallback for scan descriptors
produced before that
- // marker existed. Decode before parsing so STRING/CHAR and VARBINARY
all compare against
- // the raw bytes stored in equality-delete files.
- std::string decoded_default;
- if (!base64_decode(*table_field.initial_default_value,
&decoded_default)) {
- return Status::InvalidArgument("Invalid Base64 Iceberg initial
default for field {}",
- table_field.name);
- }
- if (table_field.type->get_primitive_type() == TYPE_VARBINARY) {
- const auto initial_default =
-
Field::create_field<TYPE_VARBINARY>(StringView(decoded_default));
- // VLiteral must copy the borrowed StringView while
decoded_default is alive; UUID and
- // long FIXED defaults otherwise retain a pointer into freed
decode storage.
- literal = VLiteral::create_shared(table_field.type,
initial_default);
- } else {
-
DORIS_CHECK(is_string_type(table_field.type->get_primitive_type()));
- literal = VLiteral::create_shared(table_field.type,
-
Field::create_field<TYPE_STRING>(decoded_default));
- }
- } else {
- // An added field's initial default is its logical value in every
older data file that lacks
- // the physical column. FE normalizes the string for the current Doris
table type.
- Field initial_default;
- RETURN_IF_ERROR(table_field.type->get_serde()->from_fe_string(
- *table_field.initial_default_value, initial_default));
- literal = VLiteral::create_shared(table_field.type, initial_default);
- }
-
- DORIS_CHECK(literal != nullptr);
+ RETURN_IF_ERROR(build_initial_default_literal(table_field, &literal));
if (table_field.type->equals(*delete_key_type)) {
*key_expr = std::move(literal);
return Status::OK();
}
auto cast_expr = Cast::create_shared(delete_key_type);
- cast_expr->add_child(std::move(literal));
+ cast_expr->add_child(literal);
*key_expr = std::move(cast_expr);
return Status::OK();
}
+static bool find_equality_delete_column_path(const
std::vector<format::ColumnDefinition>& fields,
+ int32_t field_id,
+ std::vector<const
format::ColumnDefinition*>* path) {
+ DORIS_CHECK(path != nullptr);
+ for (const auto& field : fields) {
+ path->push_back(&field);
+ if (field.has_identifier_field_id() && field.get_identifier_field_id()
== field_id) {
+ return true;
+ }
+ if (find_equality_delete_column_path(field.children, field_id, path)) {
+ return true;
+ }
+ path->pop_back();
+ }
+ return false;
+}
+
+class NestedStructFieldExpr final : public VExpr {
+public:
+ NestedStructFieldExpr(DataTypePtr data_type, std::vector<size_t>
child_indexes,
+ std::string expr_name)
+ : VExpr(std::move(data_type), false),
+ _child_indexes(std::move(child_indexes)),
+ _expr_name(std::move(expr_name)) {
+ _node_type = TExprNodeType::FUNCTION_CALL;
+ }
+
+ Status prepare(RuntimeState* state, const RowDescriptor& row_desc,
+ VExprContext* context) override {
+ RETURN_IF_ERROR_OR_PREPARED(VExpr::prepare(state, row_desc, context));
+ _prepare_finished = true;
+ return Status::OK();
+ }
+
+ Status open(RuntimeState* state, VExprContext* context,
+ FunctionContext::FunctionStateScope scope) override {
+ RETURN_IF_ERROR_OR_PREPARED(VExpr::open(state, context, scope));
+ _open_finished = true;
+ return Status::OK();
+ }
+
+ void close(VExprContext* context, FunctionContext::FunctionStateScope
scope) override {
+ VExpr::close(context, scope);
+ }
+
+ Status execute_column_impl(VExprContext* context, const Block* block,
const Selector* selector,
+ size_t count, ColumnPtr& result_column) const
override {
+ DORIS_CHECK(_children.size() == 1);
+ ColumnPtr current;
+ RETURN_IF_ERROR(
+ _children.front()->execute_column(context, block, selector,
count, current));
+
+ std::vector<const NullMap*> ancestor_null_maps;
+ for (const size_t child_index : _child_indexes) {
+ if (const auto* nullable =
check_and_get_column<ColumnNullable>(*current);
Review Comment:
[P1] Materialize missing struct defaults before nested lookup
When the whole top-level struct is absent from an old data file, `data_path`
is empty and `missing_root_expr` is a `VLiteral`. `VLiteral` returns a
`ColumnConst`, but this loop tests it directly as
`ColumnNullable`/`ColumnStruct`; the nullable check misses the wrapper and the
following `DORIS_CHECK(struct_column != nullptr)` aborts. This is reachable for
an added optional `payload STRUCT<k INT>` followed by an equality delete on
`payload.k`, for both a NULL struct and a non-NULL complex initial default.
Existing tests cover only a physically present parent with a missing child.
Please materialize `current` before traversing it and add forced V2 Parquet/ORC
coverage where the entire struct is absent.
--
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]
