github-actions[bot] commented on code in PR #67675:
URL: https://github.com/apache/doris/pull/67675#discussion_r4056146962
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java:
##########
@@ -4102,11 +4101,12 @@ public Index getInvertedIndex(Column column,
List<String> subPath, String analyz
String childName = child.getName();
if (child.getFieldPatternType() ==
PatternType.MATCH_NAME_GLOB) {
try {
- com.google.re2j.Pattern compiled =
GlobRegexUtil.getOrCompilePattern(childName);
- if (compiled.matcher(subPathString).matches()) {
Review Comment:
[P2] Keep the FE and BE field-pattern dialects aligned. This Java matcher
treats braces as alternation, so a persisted pattern such as `a{b,c}` matches
path `ab` here, while the changed BE `fnmatch(..., FNM_PATHNAME)` treats the
braces literally and does not materialize/attach that field-pattern index for
`ab`. The pattern comes from an unrestricted SQL string, so FE can select the
field-specific index/analyzer for a path that BE classified as dynamic. Use one
shared restricted dialect (or reject constructs whose meanings differ) and
cover the same edge-pattern matrix on both sides.
##########
regression-test/suites/variant_p2/load.groovy:
##########
@@ -30,6 +34,8 @@ suite("load_p2", "variant_type,p2"){
set 'max_filter_ratio', '0.1'
file file_name // import json file
time 10000 // limit inflight 10s
+ timeout 60000
Review Comment:
[P2] Validate the byte range before enabling resumable loads. After this
stream has delivered bytes and retries with `Range: bytes=<offset>-`,
`ResumableHttpInputStream` currently accepts a legal `200 OK` and exposes that
full body while pretending it starts at `offset`. An origin/proxy that ignores
Range therefore appends the file again from byte zero, duplicating rows (the
malformed join row may even fit the 10% filter allowance) and only fails after
the full 44M-row load. Require and validate a matching `206 Content-Range`, or
explicitly skip/verify the prefix and representation identity, and cover a
mid-body drop followed by `200`.
##########
regression-test/suites/variant_p2/run_relational_benchmark.py:
##########
@@ -0,0 +1,211 @@
+#!/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 re
+import signal
+import statistics
+import subprocess
+import sys
+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")
+ # Job cancellation sends SIGTERM. Unwind through the finally block below
so the regression
+ # child is stopped and FE/BE thread affinity is restored.
+ signal.signal(signal.SIGTERM, lambda signum, _: sys.exit(128 + signum))
+ 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()}
+ # The regression client must query the FE whose threads are pinned and
fingerprinted here.
+ conf_port = re.search(r"jdbc:mysql://[^:/]+:(\d+)",
Path(args.conf).read_text())
+ fe_port = re.search(r"^\s*query_port\s*=\s*(\d+)",
+ (repo / "output/fe/conf/fe.conf").read_text(),
re.MULTILINE)
+ if conf_port is None or int(conf_port.group(1)) != int(fe_port.group(1) if
fe_port else 9030):
+ raise RuntimeError("--conf does not point at this worktree's FE query
port")
+ 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)
+ }
+ manifest["dirty_checkout"] = bool(subprocess.check_output(
+ ["git", "status", "--porcelain", "--untracked-files=no"], cwd=repo,
text=True).strip())
+ # Hash what the recorded processes run, not only what the checkout
contains.
+ with Path(f"/proc/{processes['be']}/exe").open("rb") as binary:
+ manifest["be_sha256"] = hashlib.file_digest(binary,
"sha256").hexdigest()
+ with (repo / "output/fe/lib/doris-fe.jar").open("rb") as jar:
+ manifest["fe_jar_sha256"] = hashlib.file_digest(jar,
"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:
+ if cpus:
+ # Re-enumerate until every thread has the mask. New threads inherit
+ # the creator's mask; the monitor below rejects any subsequent
drift.
+ for pid in processes.values():
+ for _ in range(10):
+ tids = [int(t.name) for t in
Path(f"/proc/{pid}/task").iterdir()]
+ for tid in tids:
+ try:
+ os.sched_setaffinity(tid, cpus)
+ except ProcessLookupError:
+ pass
+ if all(os.sched_getaffinity(int(t.name)) == cpus
+ for t in Path(f"/proc/{pid}/task").iterdir()):
+ break
+ else:
+ raise RuntimeError("Could not establish CPU affinity")
+ suites = (["load_p2", "variant_relational_performance"]
+ if args.phase == "load" else
["variant_relational_performance"])
+ with (evidence / "host-load.jsonl").open("w") as host:
+ for suite in suites:
+ suite_environment = environment.copy()
+ if args.phase == "load" and suite ==
"variant_relational_performance":
+ suite_environment["VARIANT_BENCH_PHASE"] = "prepare"
+ log_name = ({"load_p2": "load.log"}.get(suite, "prepare.log")
+ if args.phase == "load" else "regression.log")
+ command = [str(repo / "run-regression-test.sh"), "--conf",
args.conf, "--run",
+ "-d", "variant_p2", "-s", suite]
+ with (evidence / log_name).open("w") as log:
+ child = subprocess.Popen(command, cwd=repo,
env=suite_environment, stdout=log,
+ stderr=subprocess.STDOUT,
start_new_session=True)
+ while child.poll() is None:
+ host.write(json.dumps({"time": time.time(),
+ "loadavg":
Path("/proc/loadavg").read_text().strip()}) + "\n")
+ host.flush()
+ if cpus:
+ for pid in processes.values():
+ for thread in
Path(f"/proc/{pid}/task").iterdir():
+ try:
+ if
os.sched_getaffinity(int(thread.name)) != cpus:
+ raise RuntimeError(f"CPU affinity
drift: {thread}")
+ except ProcessLookupError:
+ pass
+ time.sleep(1)
+ if child.returncode:
+ raise RuntimeError(f"Regression failed: inspect {evidence
/ log_name}")
+ child = None
+ finally:
+ if child is not None and child.poll() is None:
+ os.killpg(child.pid, signal.SIGTERM)
+ child.wait()
+ if cpus:
+ for pid, threads in original.items():
+ for _ in range(10):
+ for thread in Path(f"/proc/{pid}/task").iterdir():
Review Comment:
[P2] Restore every surviving process even when one server exits. `original`
visits BE before FE, but this `/proc/<pid>/task` enumeration is outside the
missing-process guard. If BE crashes during the run, `FileNotFoundError`
escapes this `finally` before FE is visited, leaving the surviving FE pinned to
the benchmark CPUs. This is distinct from the earlier SIGTERM entry-path issue:
controlled unwinding now reaches cleanup, but cleanup itself aborts on server
exit. Treat a vanished process as already un-restorable and isolate/collect
per-process restoration errors so all survivors are attempted.
##########
regression-test/suites/variant_p2/relational_performance.groovy:
##########
@@ -0,0 +1,170 @@
+// 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])
+
+ // ORDER BY ... LIMIT is a TopN and never sorts the whole input.
The window forces a
+ // full sort of every row, and its order-sensitive checksum
compares the complete
+ // native order with the CAST order instead of the first rows only.
+ def fullSort = { orderKey -> """SELECT count(*), sum(CAST(id AS
LARGEINT) * rn) FROM (
Review Comment:
[P2] Make the full-sort oracle order-preserving. The harness hashes only
this aggregate row, and `sum(id * rn)` is not injective over permutations:
orders `[1,2,4,3]` and `[1,3,2,4]` both yield count 4 and sum 29. Native and
CAST Variant sorts can therefore differ while every sample hash still matches
and the run records `complete`. Compare the ranked `(id, rn)` relations
bidirectionally or compute a deterministic ordered digest; keep the compact
aggregate only for timing if needed.
##########
be/src/core/column/variant_v2/column_variant_v2.cpp:
##########
@@ -1465,6 +1621,29 @@ void ColumnVariantV2::pop_back(size_t length) {
_check_invariants();
}
+void ColumnVariantV2::erase(size_t start, size_t length) {
+ DORIS_CHECK_LE(start, size()) << "erase start exceeds the column size";
+ DORIS_CHECK_LE(length, size() - start) << "erase range exceeds the column
size";
+ if (length == 0) {
+ return;
+ }
+ if (_shredded) {
+ ensure_encoded();
+ }
+ if (_typed) {
+ mutate_subcolumn(_typed);
+ _typed->erase(start, length);
+ _check_invariants();
+ return;
+ }
Review Comment:
[P2] Reclaim metadata for erased Variant rows. The analytic sink now calls
`erase(0, remove_rows)` every 256 blocks to bound its persistent key columns,
but this leaves every old metadata blob live. Subsequent `insert_range_from`
calls add each new batch dictionary through a linear scan of all retained
entries, and `byte_size()/allocated_bytes()` keep accounting them. With
evolving object keys, a long window query therefore grows memory and insertion
cost with total input history instead of the live window. Compact/remap the
live metadata IDs (or maintain reusable/ref-counted entries), and exercise
repeated append/erase with distinct dictionaries.
--
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]
