hubgeter commented on code in PR #65851:
URL: https://github.com/apache/doris/pull/65851#discussion_r3663577005
##########
be/src/format_v2/table/iceberg_reader.cpp:
##########
@@ -80,61 +90,447 @@ 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();
}
+Status IcebergTableReader::annotate_projected_column(const TFileScanSlotInfo&
slot_info,
+
format::ProjectedColumnBuildContext* context,
+ format::ColumnDefinition*
column) const {
+ RETURN_IF_ERROR(format::TableReader::annotate_projected_column(slot_info,
context, column));
+ DORIS_CHECK(context != nullptr);
+ DORIS_CHECK(column != nullptr);
+ if (!supports_iceberg_scan_semantics_v2(context->scan_params)) {
+ return Status::OK();
+ }
+ if (!context->schema_column.has_value()) {
+ return Status::OK();
+ }
+
+ auto& schema_column = *context->schema_column;
+ RETURN_IF_ERROR(build_initial_default_exprs(&schema_column));
Review Comment:
Fixed in 09f9bfec3e1. V2 recursive complex-type reconstruction now performs
a complete exact current-name pass before the historical-alias pass when
current Iceberg scan semantics are active; legacy semantics are unchanged.
Added a reordered same-name reuse test where the reused current field is MAP
and the renamed field is ARRAY, and the expanded
TableReader/VIcebergTableWriter BE suite passes 89/89.
--
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]
