github-actions[bot] commented on code in PR #67675:
URL: https://github.com/apache/doris/pull/67675#discussion_r4024561047
##########
be/src/core/column/variant_v2/column_variant_v2.cpp:
##########
@@ -1037,6 +1038,101 @@ VariantRef ColumnVariantV2::get_value_ref(size_t row)
const {
return {.metadata = {.data = metadata.data, .size = metadata.size}, .value
= value};
}
+std::pair<const ColumnNullable*, const ColumnNullable*>
ColumnVariantV2::_typed_ordering_operands(
+ const ColumnVariantV2& right, bool allow_floating) const {
+ if (!is_typed() || !right.is_typed() ||
+ !(_typed_type == right._typed_type ||
_typed_type->equals(*right._typed_type))) {
+ return {nullptr, nullptr};
+ }
+ const PrimitiveType type = _typed_type->get_primitive_type();
+ // IPv4 and IPv6 typed values use their textual representation in Variant,
whose lexical
+ // ordering differs from the native address ordering. LARGEINT magnitudes
above 10^38 - 1
+ // become Variant strings, so native Int128 order is not the canonical
order. Floating point is
+ // excluded for callers that cannot apply Variant's canonical NaN ordering.
+ if (type == TYPE_IPV4 || type == TYPE_IPV6 || type == TYPE_LARGEINT ||
Review Comment:
[P2] Keep temporal typed values on the canonical validation boundary. This
fast path admits DATE/DATETIME/TIMESTAMPTZ columns and compares their native
payloads directly, but Variant encoding, hashing, spill serialization, and
mixed-representation comparison all pass through `variant_days_since_epoch()`
and reject non-null values for which `is_valid_date()` is false. Such typed
states are accepted by `create_typed()` (and the existing invalid-DATEV2 test
intentionally leaves one intact after `ensure_encoded()` fails), so the same
ORDER BY can succeed in memory and then fail after representation changes.
Validate all temporal values before publishing a typed Variant, or exclude
these types from the native path, and cover comparison versus
hash/serialization for an invalid temporal row.
##########
be/src/core/column/variant_v2/column_variant_v2.cpp:
##########
@@ -1037,6 +1038,101 @@ VariantRef ColumnVariantV2::get_value_ref(size_t row)
const {
return {.metadata = {.data = metadata.data, .size = metadata.size}, .value
= value};
}
+std::pair<const ColumnNullable*, const ColumnNullable*>
ColumnVariantV2::_typed_ordering_operands(
+ const ColumnVariantV2& right, bool allow_floating) const {
+ if (!is_typed() || !right.is_typed() ||
+ !(_typed_type == right._typed_type ||
_typed_type->equals(*right._typed_type))) {
+ return {nullptr, nullptr};
+ }
+ const PrimitiveType type = _typed_type->get_primitive_type();
+ // IPv4 and IPv6 typed values use their textual representation in Variant,
whose lexical
+ // ordering differs from the native address ordering. LARGEINT magnitudes
above 10^38 - 1
+ // become Variant strings, so native Int128 order is not the canonical
order. Floating point is
+ // excluded for callers that cannot apply Variant's canonical NaN ordering.
+ if (type == TYPE_IPV4 || type == TYPE_IPV6 || type == TYPE_LARGEINT ||
+ (!allow_floating && (type == TYPE_FLOAT || type == TYPE_DOUBLE))) {
+ return {nullptr, nullptr};
+ }
+ return {&assert_cast<const ColumnNullable&,
TypeCheckOnRelease::DISABLE>(typed_column()),
+ &assert_cast<const ColumnNullable&,
TypeCheckOnRelease::DISABLE>(right.typed_column())};
+}
+
+int ColumnVariantV2::compare_at(size_t n, size_t m, const IColumn& rhs,
+ int nan_direction_hint) const {
+ const auto& right = assert_cast<const ColumnVariantV2&,
TypeCheckOnRelease::DISABLE>(rhs);
+ DCHECK_LT(n, size());
+ DCHECK_LT(m, right.size());
+
+ if (const auto [left_nullable, right_nullable] =
_typed_ordering_operands(right, true);
+ left_nullable != nullptr) {
+ if (left_nullable->is_null_at(n)) {
+ return right_nullable->is_null_at(m) ? 0 : -1;
+ }
+ if (right_nullable->is_null_at(m)) {
+ return 1;
+ }
+ return left_nullable->get_nested_column().compare_at(
+ n, m, right_nullable->get_nested_column(), nan_direction_hint);
+ }
+
+ int result = 0;
+ // Both sides are typed but the fast path declined them (different typed
types, or a type whose
+ // native order is not the canonical one). Compare the scalars directly:
encoding them into
+ // Variant bytes first would allocate a scratch buffer per comparison, and
a sort does this
+ // O(n log n) times. A null row becomes the canonical null scalar, which
sorts smallest - the
+ // same order the typed fast path above applies.
+ if (is_typed() && right.is_typed()) {
+ const auto& left_nullable =
+ assert_cast<const ColumnNullable&,
TypeCheckOnRelease::DISABLE>(typed_column());
+ const auto& right_nullable =
+ assert_cast<const ColumnNullable&,
TypeCheckOnRelease::DISABLE>(
+ right.typed_column());
+ visit_typed_scalar_column(
+ left_nullable, _typed_type->get_primitive_type(),
_typed_type->get_scale(), n,
+ n + 1, [&](size_t, const VariantScalarRef& left_scalar) {
+ visit_typed_scalar_column(
+ right_nullable,
right._typed_type->get_primitive_type(),
+ right._typed_type->get_scale(), m, m + 1,
+ [&](size_t, const VariantScalarRef& right_scalar) {
+ result = canonical_compare(left_scalar,
right_scalar);
+ });
+ });
+ return result;
+ }
+ // Neither side is typed, so both rows already hold canonical Variant
bytes and can be compared
+ // without the row visitor's revalidation and buffers.
+ if (!is_typed() && !right.is_typed()) {
+ return canonical_compare(get_value_ref(n), right.get_value_ref(m));
+ }
+ // One side typed, one encoded: the typed side still has to be encoded
once.
+ visit_variant_v2_values(
Review Comment:
[P2] Avoid allocating once per mixed-representation comparison. When one
operand is typed and the other encoded, this calls `visit_variant_v2_values()`
for a single row; its typed branch creates and resizes a fresh
`DorisVector<char>` before every `compare_at()`. The generic equality loops
call `compare_at()` once per row (and external merge calls it repeatedly), so a
typed `CAST(... AS VARIANT)` compared with an encoded Variant key incurs
O(rows) heap allocations and a mixed merge can allocate on every comparator
invocation. Compare `VariantScalarRef` directly with the encoded root, or
otherwise reuse/hoist the scratch representation, and add a large
mixed-representation benchmark.
##########
regression-test/suites/variant_p2/relational_performance.groovy:
##########
@@ -0,0 +1,160 @@
+// 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.
+
+import groovy.json.JsonOutput
+import java.security.MessageDigest
+
+suite("variant_relational_performance", "p2,nonConcurrent") {
+ def env = System.getenv()
+ // This benchmark is driven by run_relational_benchmark.py. A plain
variant_p2 run has no
+ // prepared dimension tables, so the suite only runs when a phase is set
explicitly.
+ def phase = env.get("VARIANT_BENCH_PHASE")
+ if (phase == null) {
+ log.info("Skip variant_relational_performance: VARIANT_BENCH_PHASE is
not set")
+ return
+ }
+ long expectedRows = env.getOrDefault("VARIANT_BENCH_ROWS",
"44273863").toLong()
+ int repeats = env.getOrDefault("VARIANT_BENCH_REPEATS", "7").toInteger()
+ int warmups = env.getOrDefault("VARIANT_BENCH_WARMUPS", "2").toInteger()
+ if (!(phase in ["prepare", "query"]) || expectedRows < 1 || repeats < 1 ||
warmups < 0) {
+ throw new IllegalArgumentException("Invalid VARIANT_BENCH
configuration")
+ }
+ // These paths come directly from variant_p2/sql and the GitHub Events
schema.
+ def keys = [
+ actor_login: [column: "actor", path: "login", type: "STRING"],
+ repo_name: [column: "repo", path: "name", type: "STRING"],
+ payload_action: [column: "payload", path: "action", type: "STRING"],
+ actor_id: [column: "actor", path: "id", type: "BIGINT"]
+ ]
+ def requestedKeys = env.getOrDefault("VARIANT_BENCH_KEYS",
"actor_login,actor_id").split(",") as Set
+ if (!keys.keySet().containsAll(requestedKeys)) {
+ throw new IllegalArgumentException("Unknown VARIANT_BENCH_KEYS:
${requestedKeys - keys.keySet()}")
+ }
+ keys = keys.findAll { key, ignored -> requestedKeys.contains(key) }
+
+ def actualRows = (sql("SELECT count(*) FROM
github_events"))[0][0].toString().toLong()
+ assertEquals(expectedRows, actualRows)
+ if (phase == "prepare") {
+ sql "SET default_variant_max_subcolumns_count = 0"
+ keys.each { key, spec ->
+ def nativeKey = "${spec.column}['${spec.path}']"
+ sql "DROP TABLE IF EXISTS variant_relational_dim_${key}"
+ sql """CREATE TABLE variant_relational_dim_${key} (id BIGINT,
k VARIANT)
+ DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 16
+ PROPERTIES("replication_num"="1")"""
+ sql """INSERT INTO variant_relational_dim_${key}
+ SELECT min(id), ${nativeKey} FROM github_events
+ WHERE ${nativeKey} IS NOT NULL GROUP BY ${nativeKey}"""
+ }
+ return
+ }
+
+ sql "SET parallel_pipeline_task_num = 8"
+ sql "SET enable_sql_cache = false"
+ sql "SET enable_query_cache = false"
+ if (env.getOrDefault("VARIANT_BENCH_SPILL", "false").toBoolean()) {
+ sql "SET enable_spill = true"
+ sql "SET enable_force_spill = true"
+ sql "SET spill_min_revocable_mem = 1"
+ }
+ def output = new File(env.getOrDefault("VARIANT_BENCH_RESULTS",
+ "tmp/variant-relational-results.jsonl"))
+ output.parentFile.mkdirs()
+ def record = { event -> output.append(JsonOutput.toJson(event) + "\n")
}
+ def fingerprint = { result ->
+ MessageDigest.getInstance("SHA-256").digest(JsonOutput.toJson(
+ result.collect { row -> row.collect { value -> value == null ?
null : value.toString() } }
+ ).getBytes("UTF-8")).encodeHex().toString()
+ }
+ def query = { statement ->
+ long start = System.nanoTime()
+ def result = sql(statement)
+ [ms: (System.nanoTime() - start) / 1e6, hash: fingerprint(result),
resultRows: result.size()]
+ }
+ record([event: "start", rows: actualRows, repeats: repeats, warmups:
warmups,
+ parallelPipelineTasks: 8, keys: keys.keySet(), cpus:
env.get("VARIANT_BENCH_CPUS"),
+ spill: env.get("VARIANT_BENCH_SPILL"), time: new
Date().toString()])
+ keys.each { key, spec ->
+ def nativeKey = "${spec.column}['${spec.path}']"
+ def castKey = "CAST(${nativeKey} AS ${spec.type})"
+ def nativeGroups = """SELECT ${castKey} k, min(id) first_id,
count(*) n
+ FROM github_events GROUP BY ${nativeKey}"""
+ def castGroups = """SELECT ${castKey} k, min(id) first_id,
count(*) n
+ FROM github_events GROUP BY ${castKey}"""
+ assertEquals(0, (sql("""SELECT count(*) FROM (
+ (${nativeGroups}) EXCEPT (${castGroups}))
difference"""))[0][0].toString().toInteger())
+ assertEquals(0, (sql("""SELECT count(*) FROM (
+ (${castGroups}) EXCEPT (${nativeGroups}))
difference"""))[0][0].toString().toInteger())
+ record([event: "full_group_correctness", key: key])
+
+ def queries = [
+ group: [
+ native: "SELECT count(*), sum(n*n), sum(first_id) FROM
(${nativeGroups}) g",
+ cast: "SELECT count(*), sum(n*n), sum(first_id) FROM
(${castGroups}) g"
+ ],
+ order: [
+ native: "SELECT id FROM github_events ORDER BY
${nativeKey} NULLS FIRST, id LIMIT 1000",
+ cast: "SELECT id FROM github_events ORDER BY ${castKey}
NULLS FIRST, id LIMIT 1000"
+ ]
+ ]
+ queries.each { operation, pair ->
+ def oracle = query(pair.cast).hash
+ assertEquals(oracle, query(pair.native).hash)
+ pair.each { mode, statement ->
+ record([event: "plan", key: key, operation: operation,
mode: mode,
+ plan: sql("EXPLAIN ${statement}")])
+ }
+ for (int round = -warmups; round < repeats; ++round) {
+ def modes = round % 2 == 0 ? ["native", "cast"] : ["cast",
"native"]
+ modes.each { mode ->
+ def sample = query(pair[mode])
+ assertEquals(oracle, sample.hash)
+ record(sample + [event: "sample", key: key, operation:
operation,
+ mode: mode, round: round, sql: pair[mode]])
+ }
+ }
+ }
+
+ def leftKey = "l.${spec.column}['${spec.path}']"
+ def joinOracle = query("""SELECT count(*), sum(l.id) FROM
github_events l
Review Comment:
[P2] Bind the join benchmark to the current source data. The query phase
checks only `github_events`' row count; these dimension tables are rebuilt in a
separate prepare invocation and are never validated here. Because the CAST
oracle and every native/CAST sample join the same persisted table, an empty
dimension produces matching `(0, NULL)` results and a stale subset produces
matching subset results, yet the run still emits `complete`. Rebuild the
dimensions in the query invocation or prove bidirectionally that each dimension
equals the current grouped source, and record source/dimension counts or a
stable fingerprint in the evidence.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/util/TypeCoercionUtils.java:
##########
@@ -1449,6 +1449,11 @@ private static Expression
processComparisonPredicateInternal(ComparisonPredicate
boolean leftIsVariant = left.getDataType().isVariantType();
boolean rightIsVariant = right.getDataType().isVariantType();
+ // V2 equality is shared by scalar predicates and canonical hash join
keys. Keep
+ // ordering and mixed Variant/scalar comparisons on their existing
coercion paths.
+ if (leftIsVariant && rightIsVariant && comparisonPredicate instanceof
EqualPredicate) {
Review Comment:
[P2] Accept an untyped NULL in Variant equality. `v <=> NULL` reaches this
method with the literal still typed as `NullType`, so this two-Variant return
is missed and the mixed-type guard below throws before normal NULL widening or
the later `NullSafeEqualToEqual` rewrite can turn it into `IS NULL`. The
changed regression only succeeds by spelling `CAST(NULL AS VARIANT)`, which is
not the ordinary SQL form; `v = NULL` and `v != NULL` are rejected for the same
reason. Cast a `NullType` peer to the other operand's Variant type (while
retaining rejection for non-null mixed scalar operands), and cover
scalar/filter equality with an uncast NULL literal.
--
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]
