github-actions[bot] commented on code in PR #67675:
URL: https://github.com/apache/doris/pull/67675#discussion_r4012363847
##########
be/src/core/column/variant_v2/column_variant_v2.cpp:
##########
@@ -1037,6 +1038,71 @@ VariantRef ColumnVariantV2::get_value_ref(size_t row)
const {
return {.metadata = {.data = metadata.data, .size = metadata.size}, .value
= value};
}
+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 (is_typed() && right.is_typed() &&
+ (_typed_type == right._typed_type ||
_typed_type->equals(*right._typed_type))) {
+ 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.
+ if (type != TYPE_IPV4 && type != TYPE_IPV6) {
Review Comment:
[P1] Keep out-of-range LARGEINTs on canonical ordering. This branch covers
TYPE_LARGEINT, but its Variant adapter changes representation above precision
38: magnitudes greater than 10^38-1 become STRINGs. For typed -10^38 and 0,
native Int128 comparison puts -10^38 first, whereas canonical comparison puts
the EXACT_INTEGER 0 before the STRING. The same rows therefore change ORDER
BY/TopN order after encoding or when compared across representations, and
compare_internal repeats the bypass. Exclude LARGEINT from the native fast path
(or canonicalize its full domain consistently) and add typed-vs-encoded
boundary coverage around +/-10^38, including spill/merge.
##########
fe/fe-type/src/main/java/org/apache/doris/catalog/Type.java:
##########
@@ -651,8 +651,8 @@ public boolean isOnlyMetricType() {
}
public static final String OnlyMetricTypeErrorMsg =
- "Doris hll, bitmap, array, map, struct, jsonb, variant column must
use with specific function, and don't"
- + " support filter, group by or order by. please run 'help
hll' or 'help bitmap' or 'help array'"
+ "Doris hll, bitmap, array, map, struct, jsonb, variant column must
use with specific function"
Review Comment:
[P2] Update all consumers of the shared diagnostic. This constant is still
emitted by materialized-view validation, but unchanged system tests assert text
removed here: pytest/sys/test_sys_materialized_view_2.py:741 requires `must use
with specific function, and don't support filter`, and
pytest/sys/test_sys_array/test_array_alter.py:858 requires the old `filter or
group by` wording. pytest/lib/util.py checks these as substrings, so those
cases now fail deterministically even though four regression expectations were
updated. Preserve a compatible substring or update every affected expectation
and record the intentional user-visible wording change.
##########
regression-test/suites/variant_p2/run_relational_benchmark.py:
##########
@@ -0,0 +1,196 @@
+#!/usr/bin/env python3
+# 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.
+
+"""Run the regression suite with unrestricted load and verified 8-core
queries."""
+import argparse
+import hashlib
+import json
+import os
+from pathlib import Path
+import signal
+import statistics
+import subprocess
+import time
+
+
+def main():
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("phase", choices=("load", "query"))
+ parser.add_argument("--conf", required=True)
+ parser.add_argument("--cpus", help="Eight comma-separated physical CPU IDs
(query only)")
+ parser.add_argument("--rows", type=int, default=44_273_863)
+ parser.add_argument("--repeats", type=int, default=7)
+ parser.add_argument("--warmups", type=int, default=2)
+ parser.add_argument("--keys", default="actor_login,actor_id")
+ parser.add_argument("--stream-load", action="store_true",
+ help="Load public variant_p2 files over HTTP instead
of authenticated S3")
+ parser.add_argument("--resume-files", default="",
+ help="Comma-separated public variant_p2 files to
append after an interrupted load")
+ parser.add_argument("--spill", action="store_true", help="Separate
forced-spill correctness/stability run")
+ parser.add_argument("--output", required=True, help="New evidence
directory")
+ args = parser.parse_args()
+ if args.phase == "query" and (args.stream_load or args.resume_files):
+ parser.error("--stream-load and --resume-files apply only to the load
phase")
+ if args.resume_files and not args.stream_load:
+ parser.error("--resume-files requires --stream-load")
+ repo = Path(__file__).resolve().parents[3]
+ evidence = Path(args.output).resolve()
+ evidence.mkdir(parents=True, exist_ok=False)
+ processes = {}
+ original = {}
+ for name in ("be", "fe"):
+ pid = int((repo / f"output/{name}/bin/{name}.pid").read_text())
+ command = Path(f"/proc/{pid}/cmdline").read_bytes().replace(b"\0", b"
").decode()
+ if str(repo) not in command:
+ raise RuntimeError(f"Refusing to change unrelated {name} PID
{pid}")
+ processes[name] = pid
+ original[pid] = {int(t.name): os.sched_getaffinity(int(t.name))
+ for t in Path(f"/proc/{pid}/task").iterdir()}
+ cpus = None
+ if args.phase == "query":
+ cpus = {int(cpu) for cpu in (args.cpus or "").split(",") if cpu}
+ topology = {tuple((Path(f"/sys/devices/system/cpu/cpu{cpu}/topology")
/ item)
+ .read_text().strip() for item in
("physical_package_id", "core_id"))
+ for cpu in cpus}
+ if len(cpus) != 8 or len(topology) != 8:
+ raise ValueError("Select exactly eight distinct physical cores")
+ cache = (repo / "be/build_RELEASE/CMakeCache.txt").read_text()
+ if "CMAKE_BUILD_TYPE:STRING=RELEASE" not in cache.upper():
+ raise RuntimeError("A Release build is required for performance
measurements")
+ binary = repo / "output/be/lib/doris_be"
+ running = Path(f"/proc/{processes['be']}/exe")
+ if not os.path.samefile(binary, running):
+ raise RuntimeError("Running BE differs from the worktree output
binary")
+ release_binary = repo / "be/build_RELEASE/src/service/doris_be"
+ with release_binary.open("rb") as built, binary.open("rb") as
installed:
+ if hashlib.file_digest(built, "sha256").digest() !=
hashlib.file_digest(installed, "sha256").digest():
+ raise RuntimeError("Installed BE does not match the Release
build")
+ elif args.cpus:
+ raise ValueError("Do not bind CPUs during ingestion")
+ else:
+ for pid in processes.values():
+ if len(os.sched_getaffinity(pid)) <= 8:
+ raise RuntimeError("Load requires each FE/BE process to be
unrestricted beyond eight CPUs")
+ manifest = dict(vars(args), processes=processes, checkout=str(repo),
+ head=subprocess.check_output(["git", "rev-parse", "HEAD"],
cwd=repo, text=True).strip(),
Review Comment:
[P2] Bind benchmark evidence to the product that served it. manifest.head
currently labels the checkout, but two matching stale BE artifacts pass the
inode/hash checks, FE has no build-identity check, and the arbitrary --conf can
point run-regression-test.sh at another cluster while the script pins and
fingerprints the local PIDs. The resulting evidence can therefore report HEAD
and cluster A while timing cluster B or older binaries. Verify and record the
running FE/BE embedded build hashes against the intended commit, and resolve
the configured endpoints/backend membership to those exact processes before
collecting samples.
##########
be/src/core/column/variant_v2/column_variant_v2.cpp:
##########
@@ -1037,6 +1038,71 @@ VariantRef ColumnVariantV2::get_value_ref(size_t row)
const {
return {.metadata = {.data = metadata.data, .size = metadata.size}, .value
= value};
}
+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 (is_typed() && right.is_typed() &&
+ (_typed_type == right._typed_type ||
_typed_type->equals(*right._typed_type))) {
+ 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.
+ if (type != TYPE_IPV4 && type != TYPE_IPV6) {
+ 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());
+ 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;
+ visit_variant_v2_values(
Review Comment:
[P1] Handle projected shredded values before comparing them. A partial
native-Parquet Variant projection creates complete=false state, and extracting
a retained object/array path can return another shredded ColumnVariantV2 with
that flag. This fallback calls read_view(), which unconditionally calls
materialized_column(); that method deliberately throws because omitted roots
cannot be reconstructed. Equality-key serialization similarly reaches
get_value_ref() and the same throw, despite serialized_column() being able to
produce self-contained bytes for the retained projection. Thus ORDER BY/TopN or
an equality join on an object/array external subpath fails before canonical
comparison. Use the supported retained serialization path, or require complete
materialization when planning a relational key, and add a partial-Parquet
object/array regression.
##########
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] Update the existing coercion unit test with this behavior change.
TypeCoercionUtilsTest.testVariantComparisonRequiresExplicitCast still has three
assertThrows cases that now return here: direct Variant EqualTo, direct
NullSafeEqual, and Variant-subpath EqualTo. Those assertions fail before
checking the old diagnostic, even though the newly added focused test passes.
Change them to assert the predicate is accepted without a cast, retain the
ordering/mixed-type rejection cases, and run the full FE UT suite.
##########
regression-test/suites/variant_p2/run_relational_benchmark.py:
##########
@@ -0,0 +1,196 @@
+#!/usr/bin/env python3
+# 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.
+
+"""Run the regression suite with unrestricted load and verified 8-core
queries."""
+import argparse
+import hashlib
+import json
+import os
+from pathlib import Path
+import signal
+import statistics
+import subprocess
+import time
+
+
+def main():
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("phase", choices=("load", "query"))
+ parser.add_argument("--conf", required=True)
+ parser.add_argument("--cpus", help="Eight comma-separated physical CPU IDs
(query only)")
+ parser.add_argument("--rows", type=int, default=44_273_863)
+ parser.add_argument("--repeats", type=int, default=7)
+ parser.add_argument("--warmups", type=int, default=2)
+ parser.add_argument("--keys", default="actor_login,actor_id")
+ parser.add_argument("--stream-load", action="store_true",
+ help="Load public variant_p2 files over HTTP instead
of authenticated S3")
+ parser.add_argument("--resume-files", default="",
+ help="Comma-separated public variant_p2 files to
append after an interrupted load")
+ parser.add_argument("--spill", action="store_true", help="Separate
forced-spill correctness/stability run")
+ parser.add_argument("--output", required=True, help="New evidence
directory")
+ args = parser.parse_args()
+ if args.phase == "query" and (args.stream_load or args.resume_files):
+ parser.error("--stream-load and --resume-files apply only to the load
phase")
+ if args.resume_files and not args.stream_load:
+ parser.error("--resume-files requires --stream-load")
+ repo = Path(__file__).resolve().parents[3]
+ evidence = Path(args.output).resolve()
+ evidence.mkdir(parents=True, exist_ok=False)
+ processes = {}
+ original = {}
+ for name in ("be", "fe"):
+ pid = int((repo / f"output/{name}/bin/{name}.pid").read_text())
+ command = Path(f"/proc/{pid}/cmdline").read_bytes().replace(b"\0", b"
").decode()
+ if str(repo) not in command:
+ raise RuntimeError(f"Refusing to change unrelated {name} PID
{pid}")
+ processes[name] = pid
+ original[pid] = {int(t.name): os.sched_getaffinity(int(t.name))
+ for t in Path(f"/proc/{pid}/task").iterdir()}
+ cpus = None
+ if args.phase == "query":
+ cpus = {int(cpu) for cpu in (args.cpus or "").split(",") if cpu}
+ topology = {tuple((Path(f"/sys/devices/system/cpu/cpu{cpu}/topology")
/ item)
+ .read_text().strip() for item in
("physical_package_id", "core_id"))
+ for cpu in cpus}
+ if len(cpus) != 8 or len(topology) != 8:
+ raise ValueError("Select exactly eight distinct physical cores")
+ cache = (repo / "be/build_RELEASE/CMakeCache.txt").read_text()
+ if "CMAKE_BUILD_TYPE:STRING=RELEASE" not in cache.upper():
+ raise RuntimeError("A Release build is required for performance
measurements")
+ binary = repo / "output/be/lib/doris_be"
+ running = Path(f"/proc/{processes['be']}/exe")
+ if not os.path.samefile(binary, running):
+ raise RuntimeError("Running BE differs from the worktree output
binary")
+ release_binary = repo / "be/build_RELEASE/src/service/doris_be"
+ with release_binary.open("rb") as built, binary.open("rb") as
installed:
+ if hashlib.file_digest(built, "sha256").digest() !=
hashlib.file_digest(installed, "sha256").digest():
+ raise RuntimeError("Installed BE does not match the Release
build")
+ elif args.cpus:
+ raise ValueError("Do not bind CPUs during ingestion")
+ else:
+ for pid in processes.values():
+ if len(os.sched_getaffinity(pid)) <= 8:
+ raise RuntimeError("Load requires each FE/BE process to be
unrestricted beyond eight CPUs")
+ manifest = dict(vars(args), processes=processes, checkout=str(repo),
+ head=subprocess.check_output(["git", "rev-parse", "HEAD"],
cwd=repo, text=True).strip(),
+ started=time.time(), original_affinity={str(p): {str(t):
sorted(m) for t, m in ts.items()}
+ for p, ts in
original.items()})
+ manifest["harness_sha256"] = {
+ name: hashlib.sha256((Path(__file__).parent /
name).read_bytes()).hexdigest()
+ for name in ("load.groovy", "relational_performance.groovy",
Path(__file__).name)
+ }
+ if cpus:
+ with (repo / "output/be/lib/doris_be").open("rb") as binary:
+ manifest["be_sha256"] = hashlib.file_digest(binary,
"sha256").hexdigest()
+ (evidence / "manifest.json").write_text(json.dumps(manifest, indent=2))
+ environment = dict(os.environ, VARIANT_BENCH_PHASE=args.phase,
+ VARIANT_BENCH_ROWS=str(args.rows),
VARIANT_BENCH_REPEATS=str(args.repeats),
+ VARIANT_BENCH_WARMUPS=str(args.warmups),
+ VARIANT_BENCH_KEYS=args.keys,
+ VARIANT_BENCH_RESULTS=str(evidence / "samples.jsonl"),
+ VARIANT_BENCH_SPILL=str(args.spill).lower(),
+
VARIANT_P2_USE_STREAM_LOAD=str(args.stream_load).lower(),
+ VARIANT_P2_RESUME_FILES=args.resume_files,
+ VARIANT_BENCH_CPUS=args.cpus or "unrestricted")
+ for key in list(environment):
+ if key.lower() in ("http_proxy", "https_proxy", "all_proxy"):
+ del environment[key]
+ environment["NO_PROXY"] = environment["no_proxy"] = "127.0.0.1,localhost"
+ child = None
+ try:
Review Comment:
[P2] Restore benchmark state on SIGTERM. The script changes every FE/BE
thread's affinity and starts the regression child in a separate session, but it
installs no SIGTERM handler; Python's default action terminates the interpreter
without unwinding this finally. A normal CI/job cancellation can therefore
leave the child running and the cluster pinned to eight cores, contaminating
later work. Convert SIGTERM into controlled unwinding (or a checked stop flag),
then terminate/reap the child group and restore affinity in finally; SIGKILL
can remain explicitly uncleanable.
##########
be/src/core/column/variant_v2/column_variant_v2.cpp:
##########
@@ -1037,6 +1038,71 @@ VariantRef ColumnVariantV2::get_value_ref(size_t row)
const {
return {.metadata = {.data = metadata.data, .size = metadata.size}, .value
= value};
}
+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 (is_typed() && right.is_typed() &&
+ (_typed_type == right._typed_type ||
_typed_type->equals(*right._typed_type))) {
+ 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.
+ if (type != TYPE_IPV4 && type != TYPE_IPV6) {
+ 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());
+ 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;
+ visit_variant_v2_values(
+ *this, n, n + 1, {}, [](size_t) { DCHECK(false); },
+ [&](size_t, VariantRef left_value) {
+ visit_variant_v2_values(
+ right, m, m + 1, {}, [](size_t) { DCHECK(false); },
+ [&](size_t, VariantRef right_value) {
+ result = canonical_compare(left_value,
right_value);
+ });
+ });
+ return result;
+}
+
+void ColumnVariantV2::compare_internal(size_t rhs_row_id, const IColumn& rhs,
+ int nan_direction_hint, int direction,
+ std::vector<uint8_t>& cmp_res,
+ uint8_t* __restrict filter) const {
+ const auto& right = assert_cast<const ColumnVariantV2&,
TypeCheckOnRelease::DISABLE>(rhs);
+ if (is_typed() && right.is_typed() &&
+ (_typed_type == right._typed_type ||
_typed_type->equals(*right._typed_type))) {
+ const PrimitiveType type = _typed_type->get_primitive_type();
+ // ColumnVector::compare_internal does not provide Variant's canonical
NaN ordering, while
+ // IP typed values use a textual Variant ordering that differs from
native address order.
+ if (type != TYPE_FLOAT && type != TYPE_DOUBLE && type != TYPE_IPV4 &&
type != TYPE_IPV6) {
Review Comment:
[P1] Do not bypass Variant string validation. Doris VARCHAR can normally
contain invalid UTF-8, for example unhex('FF'), and scalar-to-Variant CAST
preserves that ColumnString as a typed Variant. This fast path compares two
such rows as raw bytes, while encoding, hashing, or comparison with an encoded
block calls VariantScalarRef::string and throws INVALID_ARGUMENT. Consequently
the same ORDER BY/TopN can succeed or fail depending on block representation.
Validate STRING-family values at typed-Variant construction or exclude them
from both native comparison fast paths, and cover invalid-byte typed-vs-encoded
comparisons.
##########
regression-test/suites/variant_p0/test_variant_ordering_comparison_error.groovy:
##########
@@ -20,33 +20,40 @@ suite("test_variant_ordering_comparison_error",
"p0,nonConcurrent") {
sql "SET enable_nereids_planner = true"
sql "SET enable_fallback_to_original_planner = false"
- test {
- sql """
- SELECT v
+ qt_variant_order """
Review Comment:
[P2] Cover the external and peer-ordering paths. These three-row cases only
exercise in-memory sorting; the later forced-spill queries group/join by
Variant but order final rows by integer IDs, while the p2 ORDER BY uses LIMIT
and therefore TopN. None validates a spilled run merge on a Variant key, even
though this change enables separate get_permutation, sort_column/EqualFlags,
heap, and merge comparison entry points. Add a deterministic forced
external-sort case with canonical-equal representations, SQL/JSON nulls, and an
ID tie-breaker, plus rank or dense_rank coverage for peers.
--
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]
