github-actions[bot] commented on code in PR #66302:
URL: https://github.com/apache/doris/pull/66302#discussion_r3687839410
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalIcebergMergeSink.java:
##########
@@ -69,6 +70,9 @@ public LogicalIcebergMergeSink(IcebergExternalDatabase
database,
this.targetTable = Objects.requireNonNull(targetTable, "targetTable !=
null in LogicalIcebergMergeSink");
this.targetIcebergTable = Objects.requireNonNull(
targetIcebergTable, "targetIcebergTable != null in
LogicalIcebergMergeSink");
+ // UPDATE and MERGE construct this sink directly and rewrite every
base column; validate at
+ // the shared sink boundary so an unchanged Variant cannot enter the
unsupported writer.
+ IcebergUtils.validateWriteSchema(this.targetTable.getFullSchema());
Review Comment:
[P2] Keep delete-only MERGE available for Variant tables
This rejects every MERGE on a Variant target, including a statement whose
only action is `WHEN MATCHED THEN DELETE`. That plan cannot emit an insert:
`VIcebergMergeSink::write()` reduces delete rows to `_row_id` and calls only
the position-delete writer, while the table writer is invoked only under
`has_insert`—the same reason ordinary `DELETE` remains allowed. Please
apply the Variant gate only when a MERGE can emit update/insert rows (or
route a pure-delete MERGE through the delete sink), and add a delete-only
Variant MERGE regression.
##########
be/src/format_v2/parquet/parquet_statistics.cpp:
##########
@@ -452,6 +455,322 @@ std::optional<format::LocalColumnId>
file_column_id_by_block_position(
return std::nullopt;
}
+enum class VariantComparisonOp { EQ, NE, LT, LE, GT, GE };
+
+struct VariantShreddedPredicate {
+ int slot_index = -1;
+ std::vector<std::string> path;
+ DataTypePtr comparison_type;
+ DataTypePtr literal_type;
+ Field literal;
+ VariantComparisonOp op = VariantComparisonOp::EQ;
+};
+
+std::string callable_name(const VExprSPtr& expr) {
+ if (const auto function =
std::dynamic_pointer_cast<VectorizedFnCall>(expr);
+ function != nullptr) {
+ return function->function_name();
+ }
+ return expr == nullptr ? std::string {} : expr->expr_name();
+}
+
+std::optional<VariantComparisonOp> variant_comparison_op(std::string_view
name) {
+ if (name == "eq") {
+ return VariantComparisonOp::EQ;
+ }
+ if (name == "ne") {
+ return VariantComparisonOp::NE;
+ }
+ if (name == "lt") {
+ return VariantComparisonOp::LT;
+ }
+ if (name == "le") {
+ return VariantComparisonOp::LE;
+ }
+ if (name == "gt") {
+ return VariantComparisonOp::GT;
+ }
+ if (name == "ge") {
+ return VariantComparisonOp::GE;
+ }
+ return std::nullopt;
+}
+
+VariantComparisonOp reverse_variant_comparison(VariantComparisonOp op) {
+ switch (op) {
+ case VariantComparisonOp::EQ:
+ case VariantComparisonOp::NE:
+ return op;
+ case VariantComparisonOp::LT:
+ return VariantComparisonOp::GT;
+ case VariantComparisonOp::LE:
+ return VariantComparisonOp::GE;
+ case VariantComparisonOp::GT:
+ return VariantComparisonOp::LT;
+ case VariantComparisonOp::GE:
+ return VariantComparisonOp::LE;
+ }
+ __builtin_unreachable();
+}
+
+std::optional<std::pair<Field, DataTypePtr>> variant_literal(const VExprSPtr&
expr) {
+ const auto literal = std::dynamic_pointer_cast<VLiteral>(expr);
+ if (literal == nullptr || !literal->get_column_ptr() ||
literal->get_column_ptr()->empty()) {
+ return std::nullopt;
+ }
+ Field value;
+ literal->get_column_ptr()->get(0, value);
+ if (value.is_null()) {
+ return std::nullopt;
+ }
+ return std::make_pair(std::move(value), literal->get_data_type());
+}
+
+std::optional<VariantShreddedPredicate> extract_variant_shredded_predicate(
+ const VExprContextSPtr& conjunct) {
+ if (conjunct == nullptr || conjunct->root() == nullptr ||
+ conjunct->root()->get_num_children() != 2) {
+ return std::nullopt;
+ }
+ auto op = variant_comparison_op(callable_name(conjunct->root()));
+ if (!op.has_value()) {
+ return std::nullopt;
+ }
+
+ VExprSPtr value_expr;
+ std::optional<std::pair<Field, DataTypePtr>> literal;
+ if ((literal =
variant_literal(conjunct->root()->get_child(1))).has_value()) {
+ value_expr = conjunct->root()->get_child(0);
+ } else if ((literal =
variant_literal(conjunct->root()->get_child(0))).has_value()) {
+ value_expr = conjunct->root()->get_child(1);
+ op = reverse_variant_comparison(*op);
+ } else {
+ return std::nullopt;
+ }
+
+ const auto comparison_type = value_expr->data_type();
+ while (value_expr->node_type() == TExprNodeType::CAST_EXPR &&
+ value_expr->get_num_children() == 1) {
+ if (!expr_zonemap::data_types_compatible(value_expr->data_type(),
comparison_type)) {
+ // Every removed cast must preserve the comparison domain.
Otherwise bounds for the
+ // raw typed leaf could skip rows whose value changes in an
intermediate narrowing cast.
+ return std::nullopt;
+ }
+ value_expr = value_expr->get_child(0);
+ }
+
+ std::vector<std::string> reverse_path;
+ while (callable_name(value_expr) == "element_at" &&
value_expr->get_num_children() == 2) {
+ const auto key = variant_literal(value_expr->get_child(1));
+ if (!key.has_value() || key->first.get_type() != TYPE_STRING) {
+ // Repeated array shredding has no single scalar page range, so
only object keys are
+ // eligible for this file-level optimization.
+ return std::nullopt;
+ }
+ reverse_path.push_back(key->first.get<TYPE_STRING>());
+ value_expr = value_expr->get_child(0);
+ }
+ const auto slot = std::dynamic_pointer_cast<VSlotRef>(value_expr);
+ if (slot == nullptr || reverse_path.empty() || comparison_type == nullptr
||
+ remove_nullable(slot->data_type())->get_primitive_type() !=
TYPE_VARIANT ||
+ !expr_zonemap::data_types_compatible(comparison_type,
literal->second)) {
+ return std::nullopt;
+ }
+ std::ranges::reverse(reverse_path);
+ return VariantShreddedPredicate {.slot_index = slot->column_id(),
+ .path = std::move(reverse_path),
+ .comparison_type = comparison_type,
+ .literal_type = literal->second,
+ .literal = std::move(literal->first),
+ .op = *op};
+}
+
+bool has_variant_shredded_filter(const format::FileScanRequest& request) {
+ return std::ranges::any_of(request.conjuncts, [](const auto& conjunct) {
+ return extract_variant_shredded_predicate(conjunct).has_value();
+ });
+}
+
+const ParquetColumnSchema* child_named(const ParquetColumnSchema& parent,
std::string_view name) {
+ const auto it = std::ranges::find_if(parent.children, [&](const auto&
child) {
+ return child != nullptr && child->name == name;
+ });
+ return it == parent.children.end() ? nullptr : it->get();
+}
+
+struct ResolvedVariantShredding {
+ const ParquetColumnSchema* fallback_value = nullptr;
+ const ParquetColumnSchema* typed_value = nullptr;
+};
+
+bool metadata_cast_is_order_preserving(const DataTypePtr& source, const
DataTypePtr& target) {
+ if (expr_zonemap::data_types_compatible(source, target)) {
+ return true;
+ }
+ const auto source_type = remove_nullable(source);
+ const auto target_type = remove_nullable(target);
+ const auto source_primitive = source_type->get_primitive_type();
+ const auto target_primitive = target_type->get_primitive_type();
+ // Metadata bounds may cross only exact widening domains. This mirrors the
residual CAST while
+ // excluding rounding, overflow, and narrowing cases that could reverse a
pruning decision.
+ if (source_primitive == TYPE_FLOAT && target_primitive == TYPE_DOUBLE) {
+ return true;
+ }
+ if (is_int(source_primitive) && source_primitive != TYPE_LARGEINT &&
+ is_decimalv3(target_primitive)) {
+ const uint32_t required_integer_digits = source_primitive ==
TYPE_TINYINT ? 3
+ : source_primitive ==
TYPE_SMALLINT ? 5
+ : source_primitive ==
TYPE_INT ? 10
+
: 19;
+ return target_type->get_precision() >= target_type->get_scale() &&
+ target_type->get_precision() - target_type->get_scale() >=
required_integer_digits;
+ }
+ if (is_decimalv3(source_primitive) && is_decimalv3(target_primitive)) {
+ const uint32_t source_integer_digits =
+ source_type->get_precision() - source_type->get_scale();
+ const uint32_t target_integer_digits =
+ target_type->get_precision() - target_type->get_scale();
+ return target_integer_digits >= source_integer_digits &&
+ target_type->get_scale() >= source_type->get_scale();
+ }
+ return false;
+}
+
+std::optional<Field> cast_metadata_field(const Field& value, const
DataTypePtr& source,
+ const DataTypePtr& target) {
+ if (expr_zonemap::data_types_compatible(source, target)) {
+ return value;
+ }
+ const auto source_type = remove_nullable(source);
+ const auto target_type = remove_nullable(target);
+ if (source_type->get_primitive_type() == TYPE_FLOAT &&
+ target_type->get_primitive_type() == TYPE_DOUBLE) {
+ return
Field::create_field<TYPE_DOUBLE>(static_cast<double>(value.get<TYPE_FLOAT>()));
+ }
+ try {
+ auto source_column = source_type->create_column();
+ source_column->insert(value);
+ DataTypeSerDe::FormatOptions options =
DataTypeSerDe::get_default_format_options();
+ options.converted_from_string = true;
+ std::string text = source_type->to_string(*source_column, 0, options);
+ StringRef input(text.data(), text.size());
+ auto target_column = target_type->create_column();
+ if (!target_type->get_serde()
+ ->from_string_strict_mode(input, *target_column, options)
+ .ok() ||
+ target_column->size() != 1) {
+ return std::nullopt;
+ }
+ Field result;
+ target_column->get(0, result);
+ return result;
+ } catch (...) {
+ return std::nullopt;
+ }
+}
+
+std::optional<ParquetColumnStatistics> normalize_variant_statistics(
+ const VariantShreddedPredicate& predicate, const ParquetColumnSchema&
typed_value,
+ const ParquetColumnStatistics& statistics) {
+ if (!statistics.has_min_max ||
+ expr_zonemap::data_types_compatible(typed_value.type,
predicate.comparison_type)) {
+ return statistics;
+ }
+ auto min_value =
+ cast_metadata_field(statistics.min_value, typed_value.type,
predicate.comparison_type);
+ auto max_value =
+ cast_metadata_field(statistics.max_value, typed_value.type,
predicate.comparison_type);
+ if (!min_value.has_value() || !max_value.has_value()) {
+ return std::nullopt;
+ }
+ auto normalized = statistics;
+ normalized.min_value = std::move(*min_value);
+ normalized.max_value = std::move(*max_value);
+ return normalized;
+}
+
+std::optional<ResolvedVariantShredding> resolve_variant_shredding(
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
+ const format::FileScanRequest& request, const
VariantShreddedPredicate& predicate) {
+ const auto local_id = file_column_id_by_block_position(request,
predicate.slot_index);
+ if (!local_id.has_value() || local_id->value() < 0 ||
+ local_id->value() >= static_cast<int>(file_schema.size())) {
+ return std::nullopt;
+ }
+ const ParquetColumnSchema* wrapper = file_schema[local_id->value()].get();
+ if (wrapper == nullptr || wrapper->kind !=
ParquetColumnSchemaKind::VARIANT) {
+ return std::nullopt;
+ }
+ for (const auto& component : predicate.path) {
+ const auto* typed_object = child_named(*wrapper, "typed_value");
+ if (typed_object == nullptr || typed_object->kind !=
ParquetColumnSchemaKind::STRUCT) {
+ return std::nullopt;
+ }
+ wrapper = child_named(*typed_object, component);
+ if (wrapper == nullptr || wrapper->kind !=
ParquetColumnSchemaKind::STRUCT) {
+ return std::nullopt;
+ }
+ }
+ const auto* fallback = child_named(*wrapper, "value");
+ const auto* typed = child_named(*wrapper, "typed_value");
+ const auto typed_primitive = typed == nullptr || typed->type == nullptr
+ ? INVALID_TYPE
+ :
remove_nullable(typed->type)->get_primitive_type();
+ if (fallback == nullptr || typed == nullptr ||
+ fallback->kind != ParquetColumnSchemaKind::PRIMITIVE ||
+ typed->kind != ParquetColumnSchemaKind::PRIMITIVE ||
typed->max_repetition_level != 0 ||
+ // Parquet float statistics do not prove that a page contains no NaN.
Min/max pruning in
+ // the presence of NaN is not order preserving, so keep those pages
until such proof exists.
+ typed_primitive == TYPE_FLOAT || typed_primitive == TYPE_DOUBLE ||
+ !metadata_cast_is_order_preserving(typed->type,
predicate.comparison_type) ||
Review Comment:
[P1] Keep Variant statistics in the residual comparison domain
This type-only check also admits UUID and unannotated binary leaves when
`enable_mapping_varbinary` is false, because both are exposed as Doris
`STRING`. Their footer/page bounds are still raw physical bytes, however,
while `CAST(v['path'] AS STRING)` reconstructs UUID/BINARY Variant values
and renders them through the JSON fallback. An equality predicate can
therefore match the residual expression but fall outside the raw-byte
min/max interval, causing both Row Group and page pruning to discard
matching rows. Please reject identities such as `is_uuid` and raw binary
here unless their bounds and literal are normalized through the exact
Variant cast semantics, and cover both pruning levels with a differential
test.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java:
##########
@@ -736,6 +736,7 @@ private Plan
bindIcebergTableSink(MatchingContext<UnboundIcebergTableSink<Plan>>
Pair<IcebergExternalDatabase, IcebergExternalTable> pair =
bind(ctx.cascadesContext, sink);
IcebergExternalDatabase database = pair.first;
IcebergExternalTable table = pair.second;
+ IcebergUtils.validateWriteSchema(table.getFullSchema());
Review Comment:
[P1] Validate the pinned write-target schema
`getFullSchema()` reads the table-scoped snapshot currently in
`StatementContext`. Because relations bind before this sink, a self-insert
whose source uses `FOR VERSION AS OF` can leave an old Variant-free
generation there, pass this check, and then bind/write the latest
Variant-bearing schema loaded at lines 741-750. Please load the target
snapshot first and validate the exact `targetSchema` used for the sink, and
add a historical self-insert regression case.
--
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]
