github-actions[bot] commented on code in PR #67292:
URL: https://github.com/apache/doris/pull/67292#discussion_r4324631642
##########
be/src/storage/segment/segment_iterator.cpp:
##########
@@ -1912,7 +2021,9 @@ Status SegmentIterator::_init_column_iterators() {
// If the col is predicate column, then should read the
last page to check
// if the column is full dict encoding
.is_predicate_column = _column_states[cid].has_predicate(),
- .file_reader = _file_reader.get(),
+ .file_reader = _use_read_ahead_for_column(cid)
Review Comment:
[P2] Preserve dictionary-page corruption recovery
With read-ahead enabled, a dictionary-encoded scalar column whose dictionary
page is absent from Page Cache but corrupt in File Cache now fails the query
even when the remote object is healthy. `_read_dict_data()` passes this wrapper
to PageIO, whose cache eviction/redownload path requires a
`CachedRemoteFileReader` dynamic cast. That cast fails, and
`_read_data_page()`'s new retry ends before dictionary loading. Preserve the
original corruption-recovery path for dictionary pages too, and cover a cached
dictionary checksum failure.
##########
tools/read_io_trace/analyze.py:
##########
@@ -0,0 +1,342 @@
+#!/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.
+
+"""Offline analysis of opt-in READ_IO_TRACE records. All lengths are bytes."""
+
+import argparse
+import collections
+import gzip
+import itertools
+import json
+import math
+import sys
+
+
+SOURCES = ("foreground", "hole_fill", "other")
+
+
+def distribution(values):
+ """Exact nearest-rank percentiles; an empty sample is missing evidence,
not zero latency."""
+ values = sorted(values)
+ if not values:
+ return {"count": 0}
+ return {"count": len(values), "sum": sum(values), "min": values[0],
+ "p50": values[math.ceil(len(values) * 0.50) - 1],
+ "p95": values[math.ceil(len(values) * 0.95) - 1],
+ "p99": values[math.ceil(len(values) * 0.99) - 1], "max":
values[-1]}
+
+
+def summarize_timings(events, max_examples=5):
+ """Durations are accumulated thread wall time, not query elapsed time or
CPU time."""
+ for event in events:
+ if event["start_ns"] <= 0 or event["time_ns"] < event["start_ns"]:
+ raise ValueError(f"invalid timing interval: {event['event']}")
+ details = collections.defaultdict(list)
+ for event in events:
+ for name, value in event.get("details", {}).items():
+ details[name].append(value)
+ slowest = sorted(events, key=lambda item: item["time_ns"] -
item["start_ns"], reverse=True)
+ fields = ("file", "id", "parent_id", "offset", "size", "start_ns",
"time_ns", "outcome")
+ return {
+ "duration_ns": distribution(event["time_ns"] - event["start_ns"] for
event in events),
+ "details": {name: distribution(values) for name, values in
sorted(details.items())},
+ "outcomes": dict(collections.Counter(event["outcome"] for event in
events)),
+ "slowest": [{**{name: event[name] for name in fields},
+ "details": event.get("details", {})} for event in
slowest[:max_examples]],
+ }
+
+
+def summarize_queue_scans(events, max_examples):
+ summary = summarize_timings(events, max_examples)
+ buckets = collections.defaultdict(list)
+ for event in events:
+ size = event["details"]["queue_size"]
+ bucket = "0-63" if size < 64 else "64-255" if size < 256 else
"256-1023" if size < 1024 else "1024+"
+ buckets[bucket].append(event)
+ summary["by_queue_size"] = {name: summarize_timings(items, 0)
+ for name, items in sorted(buckets.items())}
+ return summary
+
+
+def source_group(event):
+ source = event["source"]
+ return "foreground" if source in ("read_ahead", "sync") else source
+
+
+def load_events(paths):
+ """Read JSONL (or legacy INFO) files and check completeness before query
filtering."""
+ events = {}
+ checkpoints = {}
+ jsonl_processes = set()
+ warnings = []
+ for path in paths:
+ opener = gzip.open if path.endswith(".gz") else open
+ with opener(path, "rt", encoding="utf-8") as stream:
+ for number, line in enumerate(stream, 1):
+ marker = "READ_IO_TRACE "
+ legacy = not line.lstrip().startswith("{")
+ if legacy:
+ if "READ_IO_TRACE_LIMIT " in line:
+ warnings.append("Trace event limit reached; capture is
incomplete")
+ if marker not in line:
+ continue
+ try:
+ event = json.loads(line.split(marker, 1)[1] if legacy else
line)
+ if event.get("kind") == "read_io_trace_status":
+ if event["v"] != 1:
+ raise ValueError("unsupported trace checkpoint
version")
+ previous = checkpoints.get(event["process"], (0, 0))
+ checkpoints[event["process"]] = (
+ max(previous[0], event["written_events"]),
+ max(previous[1], event["dropped_events"]))
+ continue
+ if event["v"] != 1 or source_group(event) not in SOURCES:
+ raise ValueError("unsupported trace version or source")
+ key = (event["process"], event["seq"])
+ if key in events and events[key] != event:
+ raise ValueError("conflicting records for the same
process/sequence")
+ events[key] = event # rotated/symlinked log copies must
not count twice
+ if not legacy:
+ jsonl_processes.add(event["process"])
+ except (ValueError, KeyError) as error:
+ raise ValueError(f"{path}:{number}: {error}") from error
+ sequences = collections.defaultdict(list)
+ for process, sequence in events:
+ sequences[process].append(sequence)
+ for process, values in sequences.items():
+ missing = max(values) - min(values) + 1 - len(values)
+ if missing:
+ warnings.append(f"{process}: {missing} missing trace events inside
capture")
+ for process, (written, dropped) in checkpoints.items():
+ if dropped:
+ warnings.append(f"{process}: trace writer lost {dropped} events;
capture is incomplete")
+ captured = len(sequences[process])
+ if captured != written:
+ warnings.append(f"{process}: checkpoint reports {written} written
events, "
+ f"but files contain {captured}; copy the complete
drained trace")
+ for process in jsonl_processes - checkpoints.keys():
+ warnings.append(f"{process}: missing flush checkpoint; capture may
have an incomplete tail")
+ if not events:
+ warnings.append("No trace events found")
+ return list(events.values()), sorted(set(warnings))
+
+
+def overlap_example(left, right, foreground, hole, lifecycle):
+ if foreground["time_ns"] <= hole["start_ns"]:
+ order = "foreground_completed_before_hole_get"
+ elif hole["time_ns"] <= foreground["start_ns"]:
+ order = "hole_completed_before_foreground_get"
+ else:
+ order = "overlapping_get_lifetimes"
+ range_id, task_id = foreground["parent_id"], hole["parent_id"]
+ relevant = [
+ event for event in lifecycle
+ if (event["event"] == "range_writeback" and range_id and event["id"]
== range_id)
+ or (event["event"] in ("hole_queued", "hole_active", "hole_done")
+ and task_id and event["id"] == task_id)
+ or (event["event"] == "fragment_ignored_active" and task_id
+ and event["id"] == task_id and event["parent_id"] == range_id)
+ ]
+ fields = ("event", "id", "parent_id", "offset", "size", "time_ns",
"outcome")
+ return {
+ "file": foreground["file"],
+ "overlap": [left, right],
+ "bytes": right - left,
+ "order": order,
+ "foreground_get": {key: foreground[key] for key in (
+ "id", "parent_id", "source", "offset", "size", "start_ns",
"time_ns")},
+ "hole_get": {key: hole[key] for key in (
+ "id", "parent_id", "offset", "size", "start_ns", "time_ns")},
+ "lifecycle": [{key: event[key] for key in fields}
+ for event in sorted(relevant, key=lambda item:
(item["time_ns"], item["seq"]))],
+ }
+
+
+def summarize_query(events, max_examples=5):
+ """Sweep interval endpoints: O(N log N), no pairwise range comparisons.
+
+ Duplicate bytes count each extra successful transfer once. Within-source
duplication plus
+ cross-source duplication partitions that total even for three or more
overlapping GETs.
+ """
+ successful = [event for event in events if event["event"] == "s3_get"
+ and event["outcome"] == "success"]
+ failed = [event for event in events if event["event"] == "s3_get"
+ and event["outcome"] != "success"]
+ by_file = collections.defaultdict(list)
+ lifecycle = collections.defaultdict(list)
+ for event in successful:
+ if event["bytes"] != event["size"] or event["size"] <= 0:
+ raise ValueError("success GET must have positive size and a full
response")
+ by_file[event["file"]].append(event)
+ for event in events:
+ if event["event"] != "s3_get":
+ lifecycle[event["file"]].append(event)
Review Comment:
[P3] Join packed-file lifecycle events by their IDs
For packed native segments, `S3FileReader` records GETs under the physical
packed-object URI, while range/writeback/hole events use the logical segment
path retained by `PackedFileReader`. Grouping lifecycle events by `file` here
and later selecting only `lifecycle[file]` for a physical GET excludes those
events before `overlap_example()` can match their range/task IDs. Packed-file
overlap examples therefore have empty lifecycle details despite complete
captured IDs. Join lifecycle records by their process-scoped IDs while
retaining physical-object grouping for byte totals, and cover differing
logical/physical paths.
##########
be/src/io/cache/cached_remote_file_reader.cpp:
##########
@@ -132,6 +135,34 @@
CachedRemoteFileReader::CachedRemoteFileReader(FileReaderSPtr remote_file_reader
}
}
+std::shared_ptr<RangeCacheWriteback>
CachedRemoteFileReader::make_range_cache_writeback(
+ const IOContext& io_context, PartialBlockWritebackManager*
partial_block_manager) const {
+ DORIS_CHECK(_cache != nullptr);
+ DORIS_CHECK(partial_block_manager != nullptr);
+ if (_resolve_cache_write_mode(&io_context) == CacheWriteMode::NO_WRITE ||
+ use_remote_only_on_cache_miss(&io_context)) {
+ return nullptr;
+ }
+ auto* write_manager = _cache->async_write_manager();
+ DORIS_CHECK(write_manager != nullptr);
+ CacheContext cache_context(&io_context);
+ return std::make_shared<RangeCacheWriteback>(RangeCacheWritebackOptions {
+ .write_manager = write_manager,
+ .partial_block_manager = partial_block_manager,
+ .inflight_index =
config::enable_async_file_cache_write_inflight_write_buffer_index
+ ? _cache->inflight_write_buffer_index()
+ : nullptr,
+ // The wrapper closes the inner reader on destruction. Keep the
wrapper as the owner
+ // while background hole fills still read directly through the
inner reader.
+ .source_reader = FileReaderSPtr(shared_from_this(),
_remote_file_reader.get()),
+ .cache_hash = _cache_hash,
+ .file_size = size(),
+ .block_size =
static_cast<size_t>(config::file_cache_each_block_size),
+ .admission_ctx =
CacheAdmissionContext::from_cache_context(cache_context, _tablet_id),
Review Comment:
[P2] Preserve the query cache-write budget for consumed ranges
For a native cloud query with peer reads enabled and a positive
`file_cache_query_limit_bytes`, the original reader selects synchronous writes
and charges new blocks through
`split_range_into_cells()->try_admit_cache_write()`, even when async writes are
enabled globally. Read-ahead instead fetches with `NO_WRITE`, then submits
through this admission snapshot, which drops the limiter and its admission
flag; neither complete-block nor hole-fill writeback charges it. If the limiter
has not tripped before this context is created, a long scalar scan can populate
File Cache beyond its query budget. Preserve that admission policy with
ownership safe for deferred work.
--
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]
