github-actions[bot] commented on code in PR #66206:
URL: https://github.com/apache/doris/pull/66206#discussion_r3680657792
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -1200,6 +1200,9 @@ private List<Split> doGetSplits(int numBackends) throws
UserException {
return doGetSystemTableSplits();
}
+ boolean projectsVariant = !isTableLevelCountStarPushdown() &&
projectsVariant(desc);
+ checkVariantBackendCompatibility(projectsVariant,
backendPolicy.getBackends());
Review Comment:
[P1] Apply this compatibility gate to batch split generation.
This check is only reached through `doGetSplits()`. In batch mode,
`FileQueryScanNode` instead creates `SplitAssignment`; `init()` calls this
node's `startSplit()`, and `doStartSplit()` enumerates and assigns the lazy
splits without ever calling `doGetSplits()`. A large-table Variant scan can
therefore still be assigned to a `isSmoothUpgradeSrc()` backend and hit the
unsupported old-BE path that this gate is meant to prevent. Please run the
check in a shared path before the batch/non-batch branch (or synchronously in
`startSplit()` as well), and cover mixed-version batch mode in the FE test.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/AccessPathExpressionCollector.java:
##########
@@ -137,6 +137,15 @@ public Void visitSlotReference(SlotReference
slotReference, CollectorContext con
path, context.bottomFilter, ColumnAccessPathType.DATA));
return null;
}
+ if (dataType instanceof VariantType) {
Review Comment:
[P2] Preserve access paths that reach Variant through a complex slot.
The new handling only special-cases a root `VariantType` slot. For
expressions such as `info.payload['x']`, `events[1]['kind']`, or
`attrs['primary']['enabled']`, FE first records a STRUCT/ARRAY/MAP-rooted path;
`DataTypeAccessTree.setAccessByPath()` then reaches the nested Variant with a
key still remaining and throws `unsupported data type`. `rewriteRoot()` catches
that and returns the original plan, dropping nested pruning for the whole
query. The BE recursive parser also has no nested `TYPE_VARIANT` case, so
correctness survives only by reading the complete container and Variant
subtree. Please carry a Variant terminal through the FE and BE recursive paths,
and add an access-path/profile assertion for the nested regression rather than
checking values alone.
##########
be/src/format_v2/column_mapper.cpp:
##########
@@ -1784,15 +1785,126 @@ static bool table_root_is_map(const ColumnMapping&
mapping) {
return remove_nullable(mapping.table_type)->get_primitive_type() ==
TYPE_MAP;
}
+static const ColumnDefinition* find_file_child_by_name(
+ const std::vector<ColumnDefinition>& children, std::string_view name) {
+ const auto child_it = std::ranges::find_if(
+ children, [name](const ColumnDefinition& child) { return
child.name == name; });
+ return child_it == children.end() ? nullptr : &*child_it;
+}
+
+static bool variant_leaf_type_preserves_physical_identity(const
ColumnDefinition& leaf) {
+ if (!leaf.children.empty() || leaf.type == nullptr) {
+ return false;
+ }
+ // ColumnDefinition does not transport Parquet's raw-binary/UUID and
timestamp-unit tags.
+ // Limit direct leaves to identities fully described by the Doris scalar
type; every ambiguous
+ // identity must retain the complete wrapper so reconstruction can inspect
its physical schema.
+ switch (remove_nullable(leaf.type)->get_primitive_type()) {
+ case TYPE_BOOLEAN:
+ case TYPE_TINYINT:
+ case TYPE_SMALLINT:
+ case TYPE_INT:
+ case TYPE_BIGINT:
+ case TYPE_FLOAT:
+ case TYPE_DOUBLE:
+ case TYPE_DECIMAL128I:
+ case TYPE_DATEV2:
+ return true;
+ default:
+ return false;
+ }
+}
+
+static bool build_variant_leaf_path_projection(const ColumnMapping& mapping,
+ const std::vector<std::string>&
path,
+ LocalColumnIndex*
root_projection) {
+ DORIS_CHECK(root_projection != nullptr);
+ if (path.size() != 1 || path[0].empty() || path[0] == "NULL" ||
+ path[0].find('.') != std::string::npos ||
+ std::ranges::all_of(path[0], [](unsigned char value) { return
std::isdigit(value); }) ||
Review Comment:
[P1] Reject signed numeric selectors before retaining a leaf projection.
FE serializes an integer Variant selector with `String.valueOf(intValue())`,
so `v[-1]` and the object-key access `v['-1']` both arrive here as `"-1"`
(large integer narrowing can produce the same signed form). This predicate
rejects only strings made entirely of digits, so `"-1"` is treated as a
lossless object key. On a fully shredded object with a typed field named `-1`,
the mapper can retain only that leaf; execution still knows `v[-1]` is an
`ARRAY_INDEX`, cannot use the object leaf, and then tries to materialize the
intentionally incomplete state instead of returning the normal missing-path
result. Please reject signed numeric strings too (or preserve segment
kind/range end to end), and add a differential `v[-1]` versus `v['-1']` test.
##########
be/src/format_v2/parquet/reader/variant_column_reader.cpp:
##########
@@ -0,0 +1,1058 @@
+// 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.
+
+#include "format_v2/parquet/reader/variant_column_reader.h"
+
+#include <algorithm>
+#include <array>
+#include <cmath>
+#include <cstdint>
+#include <cstring>
+#include <limits>
+#include <mutex>
+#include <optional>
+#include <string_view>
+
+#include "common/exception.h"
+#include "core/assert_cast.h"
+#include "core/column/column_array.h"
+#include "core/column/column_decimal.h"
+#include "core/column/column_map.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_struct.h"
+#include "core/column/column_vector.h"
+#include "core/column/variant_v2/column_variant_v2.h"
+#include "core/column/variant_v2/column_variant_v2_typed_column.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_variant_v2.h"
+#include "core/value/variant/variant_batch_builder.h"
+#include "core/value/variant/variant_metadata.h"
+#include "format_v2/parquet/parquet_column_schema.h"
+
+namespace doris::format::parquet {
+namespace {
+
+struct Cell {
+ const IColumn* column = nullptr;
+ bool is_null = false;
+};
+
+Cell cell_at(const IColumn& column, size_t row) {
+ if (row >= column.size()) {
+ throw Exception(ErrorCode::CORRUPTION, "Parquet Variant row {} exceeds
column size {}", row,
+ column.size());
+ }
+ if (const auto* nullable = check_and_get_column<ColumnNullable>(column)) {
+ return {.column = &nullable->get_nested_column(),
+ .is_null = nullable->get_null_map_data()[row] != 0};
+ }
+ return {.column = &column, .is_null = false};
+}
+
+const ParquetColumnSchema* find_child(const ParquetColumnSchema& schema,
std::string_view name,
+ size_t* index) {
+ for (size_t i = 0; i < schema.children.size(); ++i) {
+ if (schema.children[i]->name == name) {
+ if (index != nullptr) {
+ *index = i;
+ }
+ return schema.children[i].get();
+ }
+ }
+ return nullptr;
+}
+
+Cell struct_child_at(const ParquetColumnSchema& schema, const IColumn&
physical, size_t row,
+ std::string_view name, const ParquetColumnSchema**
child_schema) {
+ const auto& structure = assert_cast<const ColumnStruct&>(physical);
+ size_t index = 0;
+ const auto* child = find_child(schema, name, &index);
+ if (child == nullptr || index >= structure.tuple_size()) {
+ throw Exception(ErrorCode::CORRUPTION, "Parquet Variant {} has no
physical child {}",
+ schema.name, name);
+ }
+ if (child_schema != nullptr) {
+ *child_schema = child;
+ }
+ return cell_at(structure.get_column(index), row);
+}
+
+uint8_t decimal_width(int precision) {
+ if (precision <= 0 || precision > 38) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Parquet Variant decimal precision {} is outside [1,
38]", precision);
+ }
+ return precision <= 9 ? 4 : (precision <= 18 ? 8 : 16);
+}
+
+uint8_t integer_width(const ParquetColumnSchema& schema, PrimitiveType type) {
+ if (schema.type_descriptor.is_unsigned_integer) {
+ throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR,
+ "Unsigned integers are not valid Parquet Variant typed
values");
+ }
+ if (schema.type_descriptor.integer_bit_width > 0) {
+ switch (schema.type_descriptor.integer_bit_width) {
+ case 8:
+ return 1;
+ case 16:
+ return 2;
+ case 32:
+ return 4;
+ case 64:
+ return 8;
+ default:
+ throw Exception(ErrorCode::CORRUPTION, "Invalid Parquet Variant
integer width {}",
+ schema.type_descriptor.integer_bit_width);
+ }
+ }
+ switch (type) {
+ case TYPE_TINYINT:
+ return 1;
+ case TYPE_SMALLINT:
+ return 2;
+ case TYPE_INT:
+ return 4;
+ case TYPE_BIGINT:
+ return 8;
+ default:
+ throw Exception(ErrorCode::CORRUPTION, "Invalid Parquet Variant
integer type {}", type);
+ }
+}
+
+void append_typed_scalar(const ParquetColumnSchema& schema, const IColumn&
column, size_t row,
+ VariantBatchBuilder::Row& builder) {
+ const PrimitiveType type =
remove_nullable(schema.type)->get_primitive_type();
+ switch (type) {
+ case TYPE_BOOLEAN:
+ builder.add_bool(assert_cast<const
ColumnUInt8&>(column).get_data()[row] != 0);
+ return;
+ case TYPE_TINYINT:
+ builder.add_scalar(
+ VariantScalarRef::integer(assert_cast<const
ColumnInt8&>(column).get_data()[row],
+ integer_width(schema, type)));
+ return;
+ case TYPE_SMALLINT:
+ builder.add_scalar(
+ VariantScalarRef::integer(assert_cast<const
ColumnInt16&>(column).get_data()[row],
+ integer_width(schema, type)));
+ return;
+ case TYPE_INT:
+ builder.add_scalar(
+ VariantScalarRef::integer(assert_cast<const
ColumnInt32&>(column).get_data()[row],
+ integer_width(schema, type)));
+ return;
+ case TYPE_BIGINT:
+ builder.add_scalar(
+ VariantScalarRef::integer(assert_cast<const
ColumnInt64&>(column).get_data()[row],
+ integer_width(schema, type)));
+ return;
+ case TYPE_FLOAT:
+ builder.add_float(assert_cast<const
ColumnFloat32&>(column).get_data()[row]);
+ return;
+ case TYPE_DOUBLE:
+ builder.add_double(assert_cast<const
ColumnFloat64&>(column).get_data()[row]);
+ return;
+ case TYPE_DECIMAL128I: {
+ const auto value = assert_cast<const
ColumnDecimal128V3&>(column).get_data()[row].value;
+ builder.add_decimal(value,
static_cast<uint8_t>(schema.type_descriptor.decimal_scale),
+
decimal_width(schema.type_descriptor.decimal_precision));
+ return;
+ }
+ case TYPE_TIMEV2: {
+ const double seconds = assert_cast<const
ColumnTimeV2&>(column).get_data()[row];
+ if (!std::isfinite(seconds) ||
+ std::abs(seconds) >
static_cast<double>(std::numeric_limits<int64_t>::max()) / 1e6) {
+ throw Exception(ErrorCode::CORRUPTION, "Invalid Parquet Variant
TIME value");
+ }
+ builder.add_time_ntz_micros(static_cast<int64_t>(std::llround(seconds
* 1e6)));
+ return;
+ }
+ case TYPE_DATETIMEV2: {
+ if (schema.type_descriptor.time_unit == ParquetTimeUnit::NANOS) {
+ // Native DATETIMEV2 is microsecond based. Reject before returning
a silently truncated
+ // value; a raw INT64 nanos decoder can be added independently.
+ throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR,
+ "Parquet Variant TIMESTAMP(NANOS) is not
supported");
+ }
+ const auto& value = assert_cast<const
ColumnDateTimeV2&>(column).get_data()[row];
+ builder.add_timestamp_micros(
+ variant_timestamp_micros(value, row, "Parquet Variant
TIMESTAMP"),
+ schema.type_descriptor.timestamp_is_adjusted_to_utc);
+ return;
+ }
+ case TYPE_TIMESTAMPTZ: {
+ if (schema.type_descriptor.time_unit == ParquetTimeUnit::NANOS) {
+ throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR,
+ "Parquet Variant TIMESTAMP(NANOS) is not
supported");
+ }
+ const auto& value = assert_cast<const
ColumnTimeStampTz&>(column).get_data()[row];
+ builder.add_timestamp_micros(
+ variant_timestamp_micros(value, row, "Parquet Variant
TIMESTAMP"), true);
+ return;
+ }
+ case TYPE_VARBINARY: {
+ const StringRef value = column.get_data_at(row);
+ if (!schema.type_descriptor.is_uuid) {
+ builder.add_binary(value);
+ return;
+ }
+ if (value.size != 16) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Parquet Variant UUID has {} bytes instead of 16",
value.size);
+ }
+ std::array<uint8_t, 16> uuid {};
+ std::memcpy(uuid.data(), value.data, uuid.size());
+ builder.add_uuid(uuid);
+ return;
+ }
+ case TYPE_STRING: {
+ const StringRef value = column.get_data_at(row);
+ if (schema.type_descriptor.is_uuid) {
+ if (value.size != 16) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Parquet Variant UUID has {} bytes instead of
16", value.size);
+ }
+ std::array<uint8_t, 16> uuid {};
+ std::memcpy(uuid.data(), value.data, uuid.size());
+ builder.add_uuid(uuid);
+ } else if (schema.type_descriptor.is_string_annotation) {
+ builder.add_string(value);
+ } else {
+ builder.add_binary(value);
+ }
+ return;
+ }
+ default:
+ if (!is_supported_variant_typed_identity(type)) {
+ throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR,
+ "Parquet Variant typed value {} is not supported",
+ remove_nullable(schema.type)->get_name());
+ }
+ dispatch_variant_typed_column(
+ column, type, [&]<PrimitiveType Type>(const auto&
typed_column) {
+ with_variant_typed_scalar<Type>(
+ typed_column, row,
+
static_cast<uint8_t>(remove_nullable(schema.type)->get_scale()),
+ [&](const VariantScalarRef& scalar) {
builder.add_scalar(scalar); });
+ });
+ }
+}
+
+enum class WrapperContext { ROOT, ARRAY_ELEMENT, OBJECT_FIELD };
+
+bool append_wrapper(const ParquetColumnSchema& schema, const IColumn& wrapper,
size_t row,
+ VariantMetadataRef metadata, VariantBatchBuilder::Row&
builder,
+ WrapperContext context);
+
+void append_typed_value(const ParquetColumnSchema& schema, const IColumn&
column, size_t row,
+ VariantMetadataRef metadata, const VariantRef*
residual,
+ VariantBatchBuilder::Row& builder) {
+ switch (schema.kind) {
+ case ParquetColumnSchemaKind::PRIMITIVE:
+ if (static_cast<bool>(residual)) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Parquet Variant scalar typed_value cannot have
residual value bytes");
+ }
+ append_typed_scalar(schema, column, row, builder);
+ return;
+ case ParquetColumnSchemaKind::STRUCT: {
+ if (static_cast<bool>(residual) && residual->basic_type() !=
VariantBasicType::OBJECT) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Parquet Variant object typed_value has non-object
residual value");
+ }
+ const auto& structure = assert_cast<const ColumnStruct&>(column);
+ if (structure.tuple_size() != schema.children.size()) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Parquet Variant object {} physical field count
mismatch", schema.name);
+ }
+ auto object = builder.start_object();
+ if (static_cast<bool>(residual)) {
+ for (uint32_t i = 0; i < residual->num_elements(); ++i) {
+ uint32_t field_id = 0;
+ const VariantRef child = residual->object_value_at(i,
&field_id);
+ object.add_key(residual->metadata.key_at(field_id));
+ builder.add_value(child);
+ }
+ }
+ for (size_t i = 0; i < schema.children.size(); ++i) {
+ const auto& child_schema = *schema.children[i];
+ const Cell child = cell_at(structure.get_column(i), row);
+ if (child.is_null) {
+ // Shredded object fields are optional wrapper groups. A
missing group means the
+ // key is absent, which differs from a present wrapper
encoding a Variant null.
+ continue;
+ }
+ // A null/null wrapper means this object field is absent. Delay
add_key until its
+ // presence is known so absent shredded fields do not turn into
Variant nulls.
+ size_t value_index = 0;
+ const auto* value_schema = find_child(child_schema, "value",
&value_index);
+ const auto& child_struct = assert_cast<const
ColumnStruct&>(*child.column);
+ const bool value_present = value_schema != nullptr &&
+
!cell_at(child_struct.get_column(value_index), row).is_null;
+ size_t typed_index = 0;
+ const auto* typed_schema = find_child(child_schema, "typed_value",
&typed_index);
+ const bool typed_present = typed_schema != nullptr &&
+
!cell_at(child_struct.get_column(typed_index), row).is_null;
+ if (!value_present && !typed_present) {
+ continue;
+ }
+ object.add_key(StringRef(child_schema.name));
+ (void)append_wrapper(child_schema, *child.column, row, metadata,
builder,
+ WrapperContext::OBJECT_FIELD);
+ }
+ object.finish();
+ return;
+ }
+ case ParquetColumnSchemaKind::LIST: {
+ if (static_cast<bool>(residual)) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Parquet Variant array typed_value cannot have
residual value bytes");
+ }
+ if (schema.children.size() != 1) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Parquet Variant array {} has invalid element
schema", schema.name);
+ }
+ const auto& array = assert_cast<const ColumnArray&>(column);
+ const size_t begin = array.offset_at(static_cast<ssize_t>(row));
+ const size_t end = array.get_offsets()[row];
+ auto scope = builder.start_array();
+ for (size_t element = begin; element < end; ++element) {
+ const Cell cell = cell_at(array.get_data(), element);
+ if (cell.is_null) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Parquet Variant shredded array element
wrapper is null");
+ }
+ (void)append_wrapper(*schema.children[0], *cell.column, element,
metadata, builder,
+ WrapperContext::ARRAY_ELEMENT);
+ }
+ scope.finish();
+ return;
+ }
+ case ParquetColumnSchemaKind::MAP:
+ case ParquetColumnSchemaKind::VARIANT:
+ throw Exception(ErrorCode::CORRUPTION, "Invalid Parquet Variant
typed_value schema {}",
+ schema.name);
+ }
+}
+
+bool append_wrapper(const ParquetColumnSchema& schema, const IColumn& wrapper,
size_t row,
+ VariantMetadataRef metadata, VariantBatchBuilder::Row&
builder,
+ WrapperContext context) {
+ Cell value;
+ if (find_child(schema, "value", nullptr) != nullptr) {
+ value = struct_child_at(schema, wrapper, row, "value", nullptr);
+ } else {
+ value.is_null = true;
+ }
+ const ParquetColumnSchema* typed_schema = nullptr;
+ Cell typed;
+ if (find_child(schema, "typed_value", nullptr) != nullptr) {
+ typed = struct_child_at(schema, wrapper, row, "typed_value",
&typed_schema);
+ } else {
+ typed.is_null = true;
+ }
+
+ if (find_child(schema, "value", nullptr) == nullptr && typed_schema ==
nullptr) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Parquet Variant wrapper {} has neither value nor
typed_value",
+ schema.name);
+ }
+ if (value.is_null && typed.is_null) {
+ if (context == WrapperContext::OBJECT_FIELD) {
+ return false;
+ }
+ if (context == WrapperContext::ARRAY_ELEMENT) {
+ throw Exception(ErrorCode::CORRUPTION, "Parquet Variant array
element is missing");
+ }
+ builder.add_null();
+ return true;
+ }
+
+ VariantRef residual {.metadata = metadata, .value = {}};
+ if (!value.is_null) {
+ residual.value = value.column->get_data_at(row);
+ }
+ if (typed.is_null) {
+ builder.add_value(residual);
+ return true;
+ }
+ append_typed_value(*typed_schema, *typed.column, row, metadata,
+ value.is_null ? nullptr : &residual, builder);
+ return true;
+}
+
+void encode_variant_range(const ParquetColumnSchema& schema, const IColumn&
wrapper,
+ const ColumnNullable* outer_nullable, size_t begin,
size_t end,
+ ColumnVariantV2& variants) {
+ try {
+ VariantBatchBuilder builder(VariantBatchBuilder::ReserveHint {.rows =
end - begin});
+ for (size_t row = begin; row < end; ++row) {
+ auto output_row = builder.begin_row();
+ if (outer_nullable != nullptr &&
outer_nullable->get_null_map_data()[row] != 0) {
+ output_row.add_null();
+ output_row.finish();
+ continue;
+ }
+ const Cell metadata_cell = struct_child_at(schema, wrapper, row,
"metadata", nullptr);
+ if (metadata_cell.is_null) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Parquet Variant {} has null metadata at row
{}", schema.name, row);
+ }
+ const StringRef metadata_bytes =
metadata_cell.column->get_data_at(row);
+ const VariantMetadataRef metadata {metadata_bytes.data,
metadata_bytes.size};
+ metadata.validate();
+ (void)append_wrapper(schema, wrapper, row, metadata, output_row,
WrapperContext::ROOT);
+ output_row.finish();
+ }
+ VariantBatchBuilder batch = builder.finish_batch();
+ variants.insert_encoded_batch(batch);
+ } catch (...) {
+ if (end - begin <= 1) {
+ throw;
+ }
+ // A single builder has one metadata dictionary. If heterogeneous file
rows cannot fit in
+ // that dictionary, split without changing the destination column's
already-valid batches.
+ // Corrupt input still reaches a one-row range and propagates its
original exception.
+ const size_t middle = begin + (end - begin) / 2;
+ encode_variant_range(schema, wrapper, outer_nullable, begin, middle,
variants);
+ encode_variant_range(schema, wrapper, outer_nullable, middle, end,
variants);
+ }
+}
+
+ColumnVariantV2::MutablePtr encode_variant_column(const ParquetColumnSchema&
schema,
+ const IColumn& physical) {
+ if (schema.kind != ParquetColumnSchemaKind::VARIANT) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT, "Parquet column {} is not
Variant",
+ schema.name);
+ }
+ const auto* outer_nullable =
check_and_get_column<ColumnNullable>(physical);
+ const IColumn& wrapper =
+ outer_nullable == nullptr ? physical :
outer_nullable->get_nested_column();
+ const auto& structure = assert_cast<const ColumnStruct&>(wrapper);
+ if (structure.tuple_size() != schema.children.size()) {
+ throw Exception(ErrorCode::CORRUPTION, "Parquet Variant {} physical
field count mismatch",
+ schema.name);
+ }
+
+ auto variants = ColumnVariantV2::create();
+ constexpr size_t MAX_RECONSTRUCTION_BATCH_ROWS = 4096;
+ for (size_t begin = 0; begin < physical.size(); begin +=
MAX_RECONSTRUCTION_BATCH_ROWS) {
+ encode_variant_range(schema, wrapper, outer_nullable, begin,
+ std::min(physical.size(), begin +
MAX_RECONSTRUCTION_BATCH_ROWS),
+ *variants);
+ }
+ return variants;
+}
+
+std::unique_ptr<ParquetColumnSchema> clone_schema(
+ const ParquetColumnSchema& source, const format::LocalColumnIndex*
projection = nullptr) {
+ auto result = std::make_unique<ParquetColumnSchema>();
+ result->local_id = source.local_id;
+ result->parquet_field_id = source.parquet_field_id;
+ result->name = source.name;
+ result->type = source.type;
+ result->variant_physical_type = source.variant_physical_type;
+ result->leaf_column_id = source.leaf_column_id;
+ result->type_descriptor = source.type_descriptor;
+ result->kind = source.kind;
+ result->max_definition_level = source.max_definition_level;
+ result->max_repetition_level = source.max_repetition_level;
+ result->nullable_definition_level = source.nullable_definition_level;
+ result->definition_level = source.definition_level;
+ result->repetition_level = source.repetition_level;
+ result->repeated_ancestor_definition_level =
source.repeated_ancestor_definition_level;
+ result->repeated_repetition_level = source.repeated_repetition_level;
+ const bool partial = format::is_partial_projection(projection);
+ result->children.reserve(partial ? projection->children.size() :
source.children.size());
+ for (const auto& child : source.children) {
Review Comment:
[P1] Keep the retained schema in the decoded tuple's projection order.
For a partial multi-key read, `AccessPathParser` sorts the Variant paths and
`projected_type()` builds the native `STRUCT` in `projection->children` order,
so the decoded `ColumnStruct` follows that order. This clone instead filters
while iterating `source.children`, preserving the physical file-schema order.
If a shredded object stores fields as `z,a` and the query requests both `a,z`,
the tuple is `[a,z]` but the retained schema is `[z,a]`; `struct_child()`
resolves the schema index and silently returns `z` for `a` (and vice versa).
Please clone partial children in `projection->children` order, or otherwise
make the decoder and state share one canonical ordering, and add a two-leaf
test whose physical order differs from the requested order.
##########
be/src/format_v2/parquet/reader/variant_column_reader.cpp:
##########
@@ -0,0 +1,1058 @@
+// 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.
+
+#include "format_v2/parquet/reader/variant_column_reader.h"
+
+#include <algorithm>
+#include <array>
+#include <cmath>
+#include <cstdint>
+#include <cstring>
+#include <limits>
+#include <mutex>
+#include <optional>
+#include <string_view>
+
+#include "common/exception.h"
+#include "core/assert_cast.h"
+#include "core/column/column_array.h"
+#include "core/column/column_decimal.h"
+#include "core/column/column_map.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_struct.h"
+#include "core/column/column_vector.h"
+#include "core/column/variant_v2/column_variant_v2.h"
+#include "core/column/variant_v2/column_variant_v2_typed_column.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_variant_v2.h"
+#include "core/value/variant/variant_batch_builder.h"
+#include "core/value/variant/variant_metadata.h"
+#include "format_v2/parquet/parquet_column_schema.h"
+
+namespace doris::format::parquet {
+namespace {
+
+struct Cell {
+ const IColumn* column = nullptr;
+ bool is_null = false;
+};
+
+Cell cell_at(const IColumn& column, size_t row) {
+ if (row >= column.size()) {
+ throw Exception(ErrorCode::CORRUPTION, "Parquet Variant row {} exceeds
column size {}", row,
+ column.size());
+ }
+ if (const auto* nullable = check_and_get_column<ColumnNullable>(column)) {
+ return {.column = &nullable->get_nested_column(),
+ .is_null = nullable->get_null_map_data()[row] != 0};
+ }
+ return {.column = &column, .is_null = false};
+}
+
+const ParquetColumnSchema* find_child(const ParquetColumnSchema& schema,
std::string_view name,
+ size_t* index) {
+ for (size_t i = 0; i < schema.children.size(); ++i) {
+ if (schema.children[i]->name == name) {
+ if (index != nullptr) {
+ *index = i;
+ }
+ return schema.children[i].get();
+ }
+ }
+ return nullptr;
+}
+
+Cell struct_child_at(const ParquetColumnSchema& schema, const IColumn&
physical, size_t row,
+ std::string_view name, const ParquetColumnSchema**
child_schema) {
+ const auto& structure = assert_cast<const ColumnStruct&>(physical);
+ size_t index = 0;
+ const auto* child = find_child(schema, name, &index);
+ if (child == nullptr || index >= structure.tuple_size()) {
+ throw Exception(ErrorCode::CORRUPTION, "Parquet Variant {} has no
physical child {}",
+ schema.name, name);
+ }
+ if (child_schema != nullptr) {
+ *child_schema = child;
+ }
+ return cell_at(structure.get_column(index), row);
+}
+
+uint8_t decimal_width(int precision) {
+ if (precision <= 0 || precision > 38) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Parquet Variant decimal precision {} is outside [1,
38]", precision);
+ }
+ return precision <= 9 ? 4 : (precision <= 18 ? 8 : 16);
+}
+
+uint8_t integer_width(const ParquetColumnSchema& schema, PrimitiveType type) {
+ if (schema.type_descriptor.is_unsigned_integer) {
+ throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR,
+ "Unsigned integers are not valid Parquet Variant typed
values");
+ }
+ if (schema.type_descriptor.integer_bit_width > 0) {
+ switch (schema.type_descriptor.integer_bit_width) {
+ case 8:
+ return 1;
+ case 16:
+ return 2;
+ case 32:
+ return 4;
+ case 64:
+ return 8;
+ default:
+ throw Exception(ErrorCode::CORRUPTION, "Invalid Parquet Variant
integer width {}",
+ schema.type_descriptor.integer_bit_width);
+ }
+ }
+ switch (type) {
+ case TYPE_TINYINT:
+ return 1;
+ case TYPE_SMALLINT:
+ return 2;
+ case TYPE_INT:
+ return 4;
+ case TYPE_BIGINT:
+ return 8;
+ default:
+ throw Exception(ErrorCode::CORRUPTION, "Invalid Parquet Variant
integer type {}", type);
+ }
+}
+
+void append_typed_scalar(const ParquetColumnSchema& schema, const IColumn&
column, size_t row,
+ VariantBatchBuilder::Row& builder) {
+ const PrimitiveType type =
remove_nullable(schema.type)->get_primitive_type();
+ switch (type) {
+ case TYPE_BOOLEAN:
+ builder.add_bool(assert_cast<const
ColumnUInt8&>(column).get_data()[row] != 0);
+ return;
+ case TYPE_TINYINT:
+ builder.add_scalar(
+ VariantScalarRef::integer(assert_cast<const
ColumnInt8&>(column).get_data()[row],
+ integer_width(schema, type)));
+ return;
+ case TYPE_SMALLINT:
+ builder.add_scalar(
+ VariantScalarRef::integer(assert_cast<const
ColumnInt16&>(column).get_data()[row],
+ integer_width(schema, type)));
+ return;
+ case TYPE_INT:
+ builder.add_scalar(
+ VariantScalarRef::integer(assert_cast<const
ColumnInt32&>(column).get_data()[row],
+ integer_width(schema, type)));
+ return;
+ case TYPE_BIGINT:
+ builder.add_scalar(
+ VariantScalarRef::integer(assert_cast<const
ColumnInt64&>(column).get_data()[row],
+ integer_width(schema, type)));
+ return;
+ case TYPE_FLOAT:
+ builder.add_float(assert_cast<const
ColumnFloat32&>(column).get_data()[row]);
+ return;
+ case TYPE_DOUBLE:
+ builder.add_double(assert_cast<const
ColumnFloat64&>(column).get_data()[row]);
+ return;
+ case TYPE_DECIMAL128I: {
+ const auto value = assert_cast<const
ColumnDecimal128V3&>(column).get_data()[row].value;
+ builder.add_decimal(value,
static_cast<uint8_t>(schema.type_descriptor.decimal_scale),
+
decimal_width(schema.type_descriptor.decimal_precision));
+ return;
+ }
+ case TYPE_TIMEV2: {
+ const double seconds = assert_cast<const
ColumnTimeV2&>(column).get_data()[row];
+ if (!std::isfinite(seconds) ||
+ std::abs(seconds) >
static_cast<double>(std::numeric_limits<int64_t>::max()) / 1e6) {
+ throw Exception(ErrorCode::CORRUPTION, "Invalid Parquet Variant
TIME value");
+ }
+ builder.add_time_ntz_micros(static_cast<int64_t>(std::llround(seconds
* 1e6)));
+ return;
+ }
+ case TYPE_DATETIMEV2: {
+ if (schema.type_descriptor.time_unit == ParquetTimeUnit::NANOS) {
+ // Native DATETIMEV2 is microsecond based. Reject before returning
a silently truncated
+ // value; a raw INT64 nanos decoder can be added independently.
+ throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR,
+ "Parquet Variant TIMESTAMP(NANOS) is not
supported");
+ }
+ const auto& value = assert_cast<const
ColumnDateTimeV2&>(column).get_data()[row];
+ builder.add_timestamp_micros(
+ variant_timestamp_micros(value, row, "Parquet Variant
TIMESTAMP"),
+ schema.type_descriptor.timestamp_is_adjusted_to_utc);
+ return;
+ }
+ case TYPE_TIMESTAMPTZ: {
+ if (schema.type_descriptor.time_unit == ParquetTimeUnit::NANOS) {
+ throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR,
+ "Parquet Variant TIMESTAMP(NANOS) is not
supported");
+ }
+ const auto& value = assert_cast<const
ColumnTimeStampTz&>(column).get_data()[row];
+ builder.add_timestamp_micros(
+ variant_timestamp_micros(value, row, "Parquet Variant
TIMESTAMP"), true);
+ return;
+ }
+ case TYPE_VARBINARY: {
+ const StringRef value = column.get_data_at(row);
+ if (!schema.type_descriptor.is_uuid) {
+ builder.add_binary(value);
+ return;
+ }
+ if (value.size != 16) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Parquet Variant UUID has {} bytes instead of 16",
value.size);
+ }
+ std::array<uint8_t, 16> uuid {};
+ std::memcpy(uuid.data(), value.data, uuid.size());
+ builder.add_uuid(uuid);
+ return;
+ }
+ case TYPE_STRING: {
+ const StringRef value = column.get_data_at(row);
+ if (schema.type_descriptor.is_uuid) {
+ if (value.size != 16) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Parquet Variant UUID has {} bytes instead of
16", value.size);
+ }
+ std::array<uint8_t, 16> uuid {};
+ std::memcpy(uuid.data(), value.data, uuid.size());
+ builder.add_uuid(uuid);
+ } else if (schema.type_descriptor.is_string_annotation) {
+ builder.add_string(value);
+ } else {
+ builder.add_binary(value);
+ }
+ return;
+ }
+ default:
+ if (!is_supported_variant_typed_identity(type)) {
+ throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR,
+ "Parquet Variant typed value {} is not supported",
+ remove_nullable(schema.type)->get_name());
+ }
+ dispatch_variant_typed_column(
+ column, type, [&]<PrimitiveType Type>(const auto&
typed_column) {
+ with_variant_typed_scalar<Type>(
+ typed_column, row,
+
static_cast<uint8_t>(remove_nullable(schema.type)->get_scale()),
+ [&](const VariantScalarRef& scalar) {
builder.add_scalar(scalar); });
+ });
+ }
+}
+
+enum class WrapperContext { ROOT, ARRAY_ELEMENT, OBJECT_FIELD };
+
+bool append_wrapper(const ParquetColumnSchema& schema, const IColumn& wrapper,
size_t row,
+ VariantMetadataRef metadata, VariantBatchBuilder::Row&
builder,
+ WrapperContext context);
+
+void append_typed_value(const ParquetColumnSchema& schema, const IColumn&
column, size_t row,
+ VariantMetadataRef metadata, const VariantRef*
residual,
+ VariantBatchBuilder::Row& builder) {
+ switch (schema.kind) {
+ case ParquetColumnSchemaKind::PRIMITIVE:
+ if (static_cast<bool>(residual)) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Parquet Variant scalar typed_value cannot have
residual value bytes");
+ }
+ append_typed_scalar(schema, column, row, builder);
+ return;
+ case ParquetColumnSchemaKind::STRUCT: {
+ if (static_cast<bool>(residual) && residual->basic_type() !=
VariantBasicType::OBJECT) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Parquet Variant object typed_value has non-object
residual value");
+ }
+ const auto& structure = assert_cast<const ColumnStruct&>(column);
+ if (structure.tuple_size() != schema.children.size()) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Parquet Variant object {} physical field count
mismatch", schema.name);
+ }
+ auto object = builder.start_object();
+ if (static_cast<bool>(residual)) {
+ for (uint32_t i = 0; i < residual->num_elements(); ++i) {
+ uint32_t field_id = 0;
+ const VariantRef child = residual->object_value_at(i,
&field_id);
+ object.add_key(residual->metadata.key_at(field_id));
+ builder.add_value(child);
+ }
+ }
+ for (size_t i = 0; i < schema.children.size(); ++i) {
+ const auto& child_schema = *schema.children[i];
+ const Cell child = cell_at(structure.get_column(i), row);
+ if (child.is_null) {
+ // Shredded object fields are optional wrapper groups. A
missing group means the
+ // key is absent, which differs from a present wrapper
encoding a Variant null.
+ continue;
+ }
+ // A null/null wrapper means this object field is absent. Delay
add_key until its
+ // presence is known so absent shredded fields do not turn into
Variant nulls.
+ size_t value_index = 0;
+ const auto* value_schema = find_child(child_schema, "value",
&value_index);
+ const auto& child_struct = assert_cast<const
ColumnStruct&>(*child.column);
+ const bool value_present = value_schema != nullptr &&
+
!cell_at(child_struct.get_column(value_index), row).is_null;
+ size_t typed_index = 0;
+ const auto* typed_schema = find_child(child_schema, "typed_value",
&typed_index);
+ const bool typed_present = typed_schema != nullptr &&
+
!cell_at(child_struct.get_column(typed_index), row).is_null;
+ if (!value_present && !typed_present) {
+ continue;
+ }
+ object.add_key(StringRef(child_schema.name));
+ (void)append_wrapper(child_schema, *child.column, row, metadata,
builder,
+ WrapperContext::OBJECT_FIELD);
+ }
+ object.finish();
+ return;
+ }
+ case ParquetColumnSchemaKind::LIST: {
+ if (static_cast<bool>(residual)) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Parquet Variant array typed_value cannot have
residual value bytes");
+ }
+ if (schema.children.size() != 1) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Parquet Variant array {} has invalid element
schema", schema.name);
+ }
+ const auto& array = assert_cast<const ColumnArray&>(column);
+ const size_t begin = array.offset_at(static_cast<ssize_t>(row));
+ const size_t end = array.get_offsets()[row];
+ auto scope = builder.start_array();
+ for (size_t element = begin; element < end; ++element) {
+ const Cell cell = cell_at(array.get_data(), element);
+ if (cell.is_null) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Parquet Variant shredded array element
wrapper is null");
+ }
+ (void)append_wrapper(*schema.children[0], *cell.column, element,
metadata, builder,
+ WrapperContext::ARRAY_ELEMENT);
+ }
+ scope.finish();
+ return;
+ }
+ case ParquetColumnSchemaKind::MAP:
+ case ParquetColumnSchemaKind::VARIANT:
+ throw Exception(ErrorCode::CORRUPTION, "Invalid Parquet Variant
typed_value schema {}",
+ schema.name);
+ }
+}
+
+bool append_wrapper(const ParquetColumnSchema& schema, const IColumn& wrapper,
size_t row,
+ VariantMetadataRef metadata, VariantBatchBuilder::Row&
builder,
+ WrapperContext context) {
+ Cell value;
+ if (find_child(schema, "value", nullptr) != nullptr) {
+ value = struct_child_at(schema, wrapper, row, "value", nullptr);
+ } else {
+ value.is_null = true;
+ }
+ const ParquetColumnSchema* typed_schema = nullptr;
+ Cell typed;
+ if (find_child(schema, "typed_value", nullptr) != nullptr) {
+ typed = struct_child_at(schema, wrapper, row, "typed_value",
&typed_schema);
+ } else {
+ typed.is_null = true;
+ }
+
+ if (find_child(schema, "value", nullptr) == nullptr && typed_schema ==
nullptr) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Parquet Variant wrapper {} has neither value nor
typed_value",
+ schema.name);
+ }
+ if (value.is_null && typed.is_null) {
+ if (context == WrapperContext::OBJECT_FIELD) {
+ return false;
+ }
+ if (context == WrapperContext::ARRAY_ELEMENT) {
+ throw Exception(ErrorCode::CORRUPTION, "Parquet Variant array
element is missing");
+ }
+ builder.add_null();
+ return true;
+ }
+
+ VariantRef residual {.metadata = metadata, .value = {}};
+ if (!value.is_null) {
+ residual.value = value.column->get_data_at(row);
+ }
+ if (typed.is_null) {
+ builder.add_value(residual);
+ return true;
+ }
+ append_typed_value(*typed_schema, *typed.column, row, metadata,
+ value.is_null ? nullptr : &residual, builder);
+ return true;
+}
+
+void encode_variant_range(const ParquetColumnSchema& schema, const IColumn&
wrapper,
+ const ColumnNullable* outer_nullable, size_t begin,
size_t end,
+ ColumnVariantV2& variants) {
+ try {
+ VariantBatchBuilder builder(VariantBatchBuilder::ReserveHint {.rows =
end - begin});
+ for (size_t row = begin; row < end; ++row) {
+ auto output_row = builder.begin_row();
+ if (outer_nullable != nullptr &&
outer_nullable->get_null_map_data()[row] != 0) {
+ output_row.add_null();
+ output_row.finish();
+ continue;
+ }
+ const Cell metadata_cell = struct_child_at(schema, wrapper, row,
"metadata", nullptr);
+ if (metadata_cell.is_null) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Parquet Variant {} has null metadata at row
{}", schema.name, row);
+ }
+ const StringRef metadata_bytes =
metadata_cell.column->get_data_at(row);
+ const VariantMetadataRef metadata {metadata_bytes.data,
metadata_bytes.size};
+ metadata.validate();
+ (void)append_wrapper(schema, wrapper, row, metadata, output_row,
WrapperContext::ROOT);
+ output_row.finish();
+ }
+ VariantBatchBuilder batch = builder.finish_batch();
+ variants.insert_encoded_batch(batch);
+ } catch (...) {
+ if (end - begin <= 1) {
+ throw;
+ }
+ // A single builder has one metadata dictionary. If heterogeneous file
rows cannot fit in
+ // that dictionary, split without changing the destination column's
already-valid batches.
+ // Corrupt input still reaches a one-row range and propagates its
original exception.
+ const size_t middle = begin + (end - begin) / 2;
+ encode_variant_range(schema, wrapper, outer_nullable, begin, middle,
variants);
+ encode_variant_range(schema, wrapper, outer_nullable, middle, end,
variants);
+ }
+}
+
+ColumnVariantV2::MutablePtr encode_variant_column(const ParquetColumnSchema&
schema,
+ const IColumn& physical) {
+ if (schema.kind != ParquetColumnSchemaKind::VARIANT) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT, "Parquet column {} is not
Variant",
+ schema.name);
+ }
+ const auto* outer_nullable =
check_and_get_column<ColumnNullable>(physical);
+ const IColumn& wrapper =
+ outer_nullable == nullptr ? physical :
outer_nullable->get_nested_column();
+ const auto& structure = assert_cast<const ColumnStruct&>(wrapper);
+ if (structure.tuple_size() != schema.children.size()) {
+ throw Exception(ErrorCode::CORRUPTION, "Parquet Variant {} physical
field count mismatch",
+ schema.name);
+ }
+
+ auto variants = ColumnVariantV2::create();
+ constexpr size_t MAX_RECONSTRUCTION_BATCH_ROWS = 4096;
+ for (size_t begin = 0; begin < physical.size(); begin +=
MAX_RECONSTRUCTION_BATCH_ROWS) {
+ encode_variant_range(schema, wrapper, outer_nullable, begin,
+ std::min(physical.size(), begin +
MAX_RECONSTRUCTION_BATCH_ROWS),
+ *variants);
+ }
+ return variants;
+}
+
+std::unique_ptr<ParquetColumnSchema> clone_schema(
+ const ParquetColumnSchema& source, const format::LocalColumnIndex*
projection = nullptr) {
+ auto result = std::make_unique<ParquetColumnSchema>();
+ result->local_id = source.local_id;
+ result->parquet_field_id = source.parquet_field_id;
+ result->name = source.name;
+ result->type = source.type;
+ result->variant_physical_type = source.variant_physical_type;
+ result->leaf_column_id = source.leaf_column_id;
+ result->type_descriptor = source.type_descriptor;
+ result->kind = source.kind;
+ result->max_definition_level = source.max_definition_level;
+ result->max_repetition_level = source.max_repetition_level;
+ result->nullable_definition_level = source.nullable_definition_level;
+ result->definition_level = source.definition_level;
+ result->repetition_level = source.repetition_level;
+ result->repeated_ancestor_definition_level =
source.repeated_ancestor_definition_level;
+ result->repeated_repetition_level = source.repeated_repetition_level;
+ const bool partial = format::is_partial_projection(projection);
+ result->children.reserve(partial ? projection->children.size() :
source.children.size());
+ for (const auto& child : source.children) {
+ const format::LocalColumnIndex* child_projection = nullptr;
+ if (partial) {
+ child_projection = format::find_child_projection(projection,
child->local_id);
+ if (child_projection == nullptr) {
+ continue;
+ }
+ }
+ result->children.push_back(clone_schema(*child, child_projection));
+ }
+ return result;
+}
+
+ColumnPtr unwrap_nullable(ColumnPtr column) {
+ if (const auto* nullable = check_and_get_column<ColumnNullable>(*column)) {
+ return nullable->get_nested_column_ptr();
+ }
+ return column;
+}
+
+ColumnPtr struct_child(const ParquetColumnSchema& schema, ColumnPtr column,
std::string_view name,
+ const ParquetColumnSchema** child_schema) {
+ column = unwrap_nullable(std::move(column));
+ const auto* structure = check_and_get_column<ColumnStruct>(*column);
+ if (structure == nullptr) {
+ return nullptr;
+ }
+ size_t index = 0;
+ const auto* child = find_child(schema, name, &index);
+ if (child == nullptr || index >= structure->tuple_size()) {
+ return nullptr;
+ }
+ if (child_schema != nullptr) {
+ *child_schema = child;
+ }
+ return structure->get_column_ptr(index);
+}
+
+bool has_present_value(const ColumnPtr& column) {
+ if (const auto* nullable = check_and_get_column<ColumnNullable>(*column)) {
+ return std::ranges::any_of(nullable->get_null_map_data(),
+ [](uint8_t is_null) { return is_null == 0;
});
+ }
+ return !column->empty();
+}
+
+bool supports_direct_typed_variant_state(const ParquetColumnSchema& schema) {
+ if (schema.type == nullptr || schema.kind !=
ParquetColumnSchemaKind::PRIMITIVE) {
+ return false;
+ }
+ // ColumnVariantV2 typed state carries only a Doris type. Binary/UUID
annotations, temporal
+ // units, and other Parquet-only identity must therefore reconstruct
canonical Variant bytes.
+ switch (remove_nullable(schema.type)->get_primitive_type()) {
+ case TYPE_BOOLEAN:
+ case TYPE_TINYINT:
+ case TYPE_SMALLINT:
+ case TYPE_INT:
+ case TYPE_BIGINT:
+ case TYPE_FLOAT:
+ case TYPE_DOUBLE:
+ case TYPE_DECIMAL128I:
+ case TYPE_DATEV2:
+ return true;
+ default:
+ return false;
+ }
+}
+
+bool same_data_type(const DataTypePtr& left, const DataTypePtr& right) {
+ return (!left && !right) || (left && right && left->equals(*right));
+}
+
+bool same_type_descriptor(const ParquetTypeDescriptor& left, const
ParquetTypeDescriptor& right) {
+ return same_data_type(left.doris_type, right.doris_type) &&
+ same_data_type(left.physical_doris_type, right.physical_doris_type)
&&
+ left.extra_type_info == right.extra_type_info && left.time_unit ==
right.time_unit &&
+ left.physical_type == right.physical_type &&
+ left.integer_bit_width == right.integer_bit_width &&
+ left.decimal_precision == right.decimal_precision &&
+ left.decimal_scale == right.decimal_scale && left.fixed_length ==
right.fixed_length &&
+ left.is_unsigned_integer == right.is_unsigned_integer &&
+ left.is_decimal == right.is_decimal && left.is_timestamp ==
right.is_timestamp &&
+ left.timestamp_is_adjusted_to_utc ==
right.timestamp_is_adjusted_to_utc &&
+ left.is_string_like == right.is_string_like &&
+ left.is_string_annotation == right.is_string_annotation &&
+ left.is_uuid == right.is_uuid && left.unsupported_reason ==
right.unsupported_reason;
+}
+
+bool same_shredded_schema(const ParquetColumnSchema& left, const
ParquetColumnSchema& right) {
+ if (left.name != right.name || left.kind != right.kind ||
+ !same_data_type(left.type, right.type) ||
+ !same_type_descriptor(left.type_descriptor, right.type_descriptor) ||
+ left.children.size() != right.children.size()) {
+ return false;
+ }
+ for (size_t i = 0; i < left.children.size(); ++i) {
+ if (!same_shredded_schema(*left.children[i], *right.children[i])) {
+ return false;
+ }
+ }
+ return true;
+}
+
+void append_compatible_column(IColumn& output, const IColumn& converted);
+void validate_compatible_column(const IColumn& output, const IColumn&
converted);
+
+class ParquetVariantShreddedState final : public VariantShreddedState {
+public:
+ ParquetVariantShreddedState(const ParquetColumnSchema& schema, ColumnPtr
physical,
+ const format::LocalColumnIndex* projection)
+ : _schema(clone_schema(schema, projection)),
Review Comment:
[P2] Share the immutable projected schema across batches and selections.
Every native read batch constructs this state and recursively clones the
entire retained Variant schema. `filter()`, `select_range()`, and
`select_indices()` then recursively clone the same immutable tree again. For a
wide shredded object, many small batches or late filter/slice operations
therefore add O(schema-node) traversal and heap allocations unrelated to the
selected row count. Please build one projection-aligned immutable schema per
reader/plan and share its ownership with each state and derivative, and add
allocation or benchmark coverage for a wide schema over many batches.
--
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]
