rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-11-14 06:46:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(promql): own raw series identities at the repartition boundary (#10058)

Browse Source 
* feat(promql): own raw series identities at the repartition boundary

A raw frame's is_last closes one producer's stream, not the series: the same
stored primary key may still arrive from another input. Routing such frames
through the evaluated path therefore emits several partial series where a
consumer expects one.

Teach SeriesRepartitionExec to group raw identities. SeriesRouting selects the
grouping key and try_new_with_routing takes an explicit policy, rejecting the
combinations a frame's content cannot honour: an evaluated frame may not use
RawPrimaryKey, a raw frame may not use CanonicalLabels, and a raw frame must
declare explicit labels.

Raw routing keys on the complete stored primary key rather than on labels so
two pieces of one physical key always meet in one owner, where a disagreement
about labels is rejected instead of being split across owners that would each
look consistent on its own.

A raw identity is published only after every input reaches normal EOF, unlike
an evaluated identity, which is complete once each input either contributed it
or ended. merge_raw_identity orders the pieces by timestamp, collapses equal
duplicates, rejects conflicting ones and emits one closing chunk. The
reconstruction's pool charge covers the two parallel row vectors, the
timestamps and the worst-case per-row sample slices that alternating
contributions produce.

This adds the capability only; no node is constructed with raw routing yet.

Signed-off-by: discord9 <discord9@163.com>

* test(promql): cover raw identity merging and routing

source/merge covers the pure merge: interleaved ordering, duplicate collapse
versus bit-level conflict, key and label rejection, histogram property
comparison and shape mismatch.

The router needs its own proof, because the readiness rule and the identity key
are what a consumer depends on. A raw child whose partitions are scripted
delivers them: one gated input that has already handed over every chunk but
whose stream is still open blocks publication until it ends; identical labels
with different primary keys stay apart; pieces of one key that disagree on
labels fail; and one key with conflicting samples at one timestamp surfaces an
error rather than dropping data.

The gate holds back only the end of the stream, so the first test distinguishes
"every input contributed" from "every input ended". Swapping the raw readiness
rule for the evaluated one makes it fail.

Signed-off-by: discord9 <discord9@163.com>

* docs(promql): correct the cross-partition merge boundary

The guide said native execution does not merge one logical series split across
partitions. That stopped being true when the repartition owner began finalizing
an exact identity's fragments: the merge now happens at that boundary, and the
remaining limit is a logical series split across regions.

Signed-off-by: discord9 <discord9@163.com>

* fix(promql): charge raw chunks their slice and publish complete identities

Two problems in the raw path, both about holding memory longer than the data
requires.

A frame batch decodes into many chunks that are slices of the same Arrow
buffers. Charging each chunk the whole allocation charged one batch's timestamps
once per chunk: 64 series of 64 rows retained 4.2 MiB for a 32 KiB timestamp
buffer, 129 times the real size, and the reservations live until the identity is
published. raw_chunk_slice_bytes charges the slice each chunk actually owns,
using the same slice-aware accounting the histogram path already used. The same
fixture now retains 95 KiB.

A raw identity also waited for every input to reach EOF even when every input
had already handed over its complete chain for it. read_raw_chains hands a run
of chunks over atomically once is_last arrives, and an input cannot contribute
the same key twice, so a contribution is final for that input. Readiness is now
the same rule for both contents: every input has contributed the identity or
ended.

The previous test asserted that a fully contributed identity must still wait for
an unrelated input's EOF, which pinned the conservative rule as if it were
required. It is replaced by two tests, one for each direction: an identity whose
inputs all contributed publishes without waiting, and an identity with an input
that has contributed nothing waits. Reverting the readiness rule to the old one
fails the first and leaves the second passing.

Signed-off-by: discord9 <discord9@163.com>

* fix(promql): charge histogram indices and steady the raw accounting tests

A histogram field is an index array naming rows of a shared table. Charging only
the table missed the indices, and a query over few distinct histograms holds
mostly indices: 4096 positions naming one row is 16 KiB of indices against a
table of a few hundred bytes. Both the histogram and the mixed branch now charge
the index array's slice, as the evaluated path already does.

The two accounting tests waited by sampling a fixed number of idle ticks. Under
load the input task may not have run yet, so the sample reads zero and the test
fails for a reason that has nothing to do with what it measures. Withdrawing the
index charge reproduces the failure, so the test still discriminates after the
wait was made deterministic.

Signed-off-by: discord9 <discord9@163.com>

* test(promql): gate the producer so the failure cannot drain unclaimed queues

The retention this test reads is not guaranteed across `start` and `claim`.
`start` spawns the producer before it returns, and a failure drains every queue
nobody has claimed yet, releasing the charges those queues held. A producer that
reaches its error first therefore leaves the test claiming two empty queues, and
"queued output chains retain memory before drain" can never become true.

That is the shape the suite now hits on this machine: the baseline commit fails
this test in about ten seconds when the suite runs concurrently with another
one, while the same test written with the gate passes every time.

The producer now waits on a switch before its error, and the test claims both
queues, waits for the retained charge, and only then opens the switch. The
assertion keeps its meaning because nothing can drain the queues before the test
owns them.

This is a test fix. The production cleanup it exposed is correct behaviour: a
queue no consumer claimed should not keep holding memory.

Signed-off-by: discord9 <discord9@163.com>

* feat(promql): route raw frames by a match key

A vector operation between two raw sources needs the series of one match key in
one partition, and the node could not give it that: raw frames were restricted to
key routing, so `100 * (1 - a / b)` could not be planned at all.

The restriction rested on an inference the code made when it started a read: the
routing policy named the frame content, so only one policy could mean raw. That
inference stops holding once a raw frame may be routed by a match key, and the
content is now read from the child's own declaration instead.

Choosing a destination and deciding a fragment's identity are separate jobs. A
match key picks the partition by hashing the labels the frame already carries --
raw frames are required to declare them -- while the stored key still decides
which identity a fragment belongs to. Both keep one key's fragments in one
partition, because one key always decodes to one label set, so merging is
unaffected. Canonical labels stay refused for raw frames: they include the metric
name, which is not a match key.

Signed-off-by: discord9 <discord9@163.com>

* fix(promql): report a raw chain that cannot pick a destination

Two paths hid a destination they could not compute. A matching-routed chunk
without labels built an error and then discarded it through `unwrap_or_else`,
routing the chunk by its stored key instead; and `output_labels` answered the
stored-key policy with nothing at all, so every chain would have gone to one
partition silently.

Neither is reachable today -- a raw frame is required to carry explicit labels,
and a stored-key-routed frame never asks for a label hash -- but a silent fallback
turns a defect into a wrong answer. Both now say what went wrong.

Signed-off-by: discord9 <discord9@163.com>

* test(promql): cover the histogram properties a duplicate check has to compare

Merging two raw pieces treats a sample seen twice at one timestamp as a duplicate
only when every property of the two matches. The comparison already covered all of
them, but its test changed only the float ones, so dropping `reset_hint`,
`start_timestamp` or `custom_values` from the comparison left every test passing:
conflicting samples would have been discarded as duplicates with no failure.

Each of the three now has a case that differs in exactly that property. Removing
any of the three comparisons fails the test.

The bucket side is deliberately not covered: a legal sample must satisfy
`span_lengths` summing to the bucket count with an in-range cursor, so a sample
differing in one bucket field is a different shape rather than a different value.

Signed-off-by: discord9 <discord9@163.com>

* perf(promql): publish an ordered identity as its inputs pass it, and charge what a merge uses

A key-then-time identity could only be published after every input ended, so the
node buffered a whole region even though the ordering it was protecting is exactly
what lets it publish early. An input emits its keys in order and each chain goes to
one owner, so the keys one owner sees from one input are non-decreasing too: a
greater key from an input proves that input will not contribute a smaller one. The
owner records the greatest key seen per input and publishes the pending identity
with the smallest key once every input has contributed it, ended, or been seen
past it. Readiness is checked in ascending key order, so the output stays ordered.

An identity carried by a single piece needs no reconstruction at all: it is marked
final and published as it is, which is what the evaluated path already did. A
multi-piece identity charges each allocation the reconstruction actually makes —
the two row vectors, the timestamps, and one descriptor per surviving run —
instead of reserving a worst case per row per field that a well-ordered identity
never approaches.

The two tests for the ordered path were written against a fixture where the key
they expected to be held was already complete, so they asserted the opposite of
the rule. They now use fixtures where the held key really has not been passed, and
they collect rows per identity because a frame may split one identity across
chunks.

Signed-off-by: discord9 <discord9@163.com>

* fix(promql): drop the uncharged merge wrapper the charging path replaced

Routing a raw identity charges each allocation before it is made, so
`merge_raw_identity` had no caller left outside the tests and
`cargo clippy -p promql --all-targets -- -D warnings` refused it as dead
code. Its body is now the test helper that calls the charging form with a
no-op reserve, and the production name is gone.

Signed-off-by: discord9 <discord9@163.com>

* fix(promql): stop scanning settled identities, and keep merge errors typed

Three faults in the raw owner, each small:

A key-then-time frame publishes only in key order, so if the smallest pending
identity is not ready no larger one can be: an input reaches a later key only by
passing the earlier one. The rescan applied that predicate to every pending
group anyway, so an input that fell behind could have the owner walk the whole
map after each arriving chain it could not yet settle. Ask the first entry only.

`reserve` returning `ResourcesExhausted` came back through `invalid(...)`, so a
memory-pool rejection reached the caller as a malformed-series error with the
wrong status. A reservation failure is not a claim about the data; hand the
`DataFusionError` through untouched.

The merge reserved its row and selection scratch up front and kept holding both
after `rows` and `selected` were dropped, since the reservation outlives them
until the packer takes the output. Release the pair once the merge is done. The
release amount matches the two reservations, so it can only fail if the
reservation was already reduced elsewhere, which is why the assertion checks it
but the query still succeeds.

Signed-off-by: discord9 <discord9@163.com>

* test(promql): assert the decode a chain-atomic merge produces

This case asserted that unpacking fails with "evaluated series identity changed
before its chain closed". It was written to keep that boundary honest until the
region merge became chain-atomic, and its own comment asks for this change once
that landed - which it has: unpacking now succeeds.

The success it asserts is the same property read the other way. Sixteen hosts,
one chain each, each closing exactly once, each reporting 1.0 - the rate its own
samples produce, since the fixture's value rises one per second. A merge that
started interleaving chains again would cut them, and a cut chain cannot report
one correct value per host.

The case's evaluation grid holds one step, so it says nothing about how a chain
spanning several chunks behaves; that is a property of the grid, not of the
merge, and the comment now says so instead of claiming four batches per chain.

Signed-off-by: discord9 <discord9@163.com>

* test(promql): cover the gaps the review named, and drop two false claims

Four independent gaps in the raw repartition tests, plus one comment that
asserted something the code does not do.

`RawScript` reported `SeriesOrdering::Contiguous` regardless of what its
plan declared. `pack_raw_series` derives both the frame metadata and the
`SequenceValidator` from that one value, so every `KeyThenTime` test packed
`Contiguous` batches and validated them under the `Contiguous` contract —
`SequenceValidator`'s "raw series keys are not ordered" check never ran.
Carry the declared ordering through `RawScripted` and hand it to the
stream, so the batches and the check agree with the plan.

`raw_matching_routing_groups_by_match_key` asserted only that both
identities shared an output, which holds under primary-key routing too
because those two keys collide in a three-output layout. Assert the owner
against `output_labels` of the match key instead, so routing by the wrong
key fails.

The test named for waiting asserted early publication through the
watermark. Rename it, and cover the other side deterministically with
`gate_chunks`: an identity is withheld while another input has delivered
nothing, then publishes once the gate opens.

The two lifecycle tests set `gate_chunks: true`, which blocks the *first*
chunk, so nothing was ever charged and their wait for a reservation could
not be satisfied. Let the chunks through and keep the gate closed at EOF:
the chain is delivered, charged, parked, and then dropped.

`histogram_duplicates_compare_every_property` claimed the bucket side was
unreachable because a legal sample must satisfy layout constraints. Only
the bucket *count* is constrained; bucket values are free. Add a case per
property — schema, count, zero_count, sum, zero_threshold, either side's
bucket value, either side's span offset and span length with equal counts,
plus the existing reset_hint, start_timestamp and custom_values — so
dropping any comparison from `histograms_equal` fails.

`native_prefix_multi_partition_series_chains` said the region merges
chain-atomically. It does not: `CoalescePartitionsExec` still interleaves
partitions batch by batch. The decode passes because #10023 closes a
region's pending raw chunk at package completion, so a chain's value and
its close are no longer latched apart. Say that, drop the unsupported
"four batches per chain" count, and note that a one-point grid gives one
value plus a possible empty close.

Signed-off-by: discord9 <discord9@163.com>

* perf(promql): merge ordered raw pieces without a row index

`merge_raw_identity_charging` rebuilt a `(timestamp, chunk, row)` entry for
every row of every piece and sorted the lot, then walked it again to drop
duplicate timestamps. Reconstructing one series therefore cost
`24 * rows` bytes of transient workspace and an `O(n log n)` sort, even
when the pieces already arrived in order.

The validation above that point has already refused a piece whose own
timestamps are not strictly increasing, so a piece never needs sorting
internally. What is left to order is only the relation between pieces.
When every adjacent pair satisfies `previous.last() < next.first()` the
concatenation is already the merged sequence, and no timestamp can repeat
across pieces, so neither the index nor the duplicate walk is needed.

Keep the general path unchanged for the interleaved and touching cases -
touching pieces share a timestamp, so they still need the duplicate check -
and shrink only what the path actually reserved.

Signed-off-by: discord9 <discord9@163.com>

* test(promql): vary histogram spans one dimension at a time

`histogram_duplicates_compare_every_property` varied `span_offsets` and
`span_lengths` together in the two length cases: the baseline was
`offsets: [0], lengths: [2]` and the cases used
`offsets: [0, 0], lengths: [1, 1]`. The offsets comparison rejected them
first, so removing the lengths comparison left the test green and those two
cases pinned nothing.

The baseline is now two spans in both sides (`offsets: [0, 0]`,
`lengths: [1, 1]`). The offset cases change only the cursors (`[1, 0]`) and
the length cases only the lengths (`[2, 0]`), keeping the cursor list, the
span count and the bucket count fixed, with a zero-length span as the only
difference. Ablating any one of the four comparisons - positive and negative
offsets, positive and negative lengths - now fails the case that covers it.

`single_chunk_uses_ordered_fast_path` is removed. It passed with the fast
path forced off, so it did not pin path selection, and the owner never
reaches the merge with one chunk: `finish_raw_group` returns the chunk after
setting `is_last`. The fast path stays pinned by
`ordered_disjoint_pieces_concatenate_without_row_workspace`, which asserts
the exact reservation and zero shrink for disjoint chunks, and by
`touching_ordered_chunks_use_general_duplicate_path` on the other side.

Signed-off-by: discord9 <discord9@163.com>

---------

Signed-off-by: discord9 <discord9@163.com>
This commit is contained in:
discord9
2026-11-02 15:08:44 +08:00
committed by GitHub
parent a47a61e8fb
commit f73a27d779
9 changed files with 2791 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/promql/AGENTS.md
+5 -4
View File
@@ -23,10 +23,11 @@ functions, vector matching and extension operators. Statement planning lives in
Native PromQL is an experimental, standalone evaluation route alongside the
existing DataFusion PromQL path. It operates on one storage region, which may
supply multiple scan partitions. It can evaluate each independent input series
in those partitions, but does not merge one logical series split across regions
or partitions; cross-region `SeriesMerge`/`Exchange` and global final-label
semantics are not implemented. Do not describe partition-local work as a global
supply multiple scan partitions. Where a boundary must answer one logical
series once, `SeriesRepartitionExec` merges that identity's fragments across
scan partitions and recomputes a single close for it. Cross-region
`SeriesMerge`/`Exchange` are still not implemented, so a logical series split
across regions is not merged. Do not describe partition-local work as a global
series merge.
Vector operations use the existing vector-matching semantics; do not describe
src/promql/src/extension_plan.rs
+1 -1
View File
@@ -57,7 +57,7 @@ pub use range_manipulate::{RangeManipulate, RangeManipulateExec, RangeManipulate
pub use raw_series_source::{RawSeriesSource, RawSeriesSourceExec};
pub use scalar_calculate::ScalarCalculate;
pub use series_divide::{SeriesDivide, SeriesDivideExec, SeriesDivideStream};
pub use series_repartition::SeriesRepartitionExec;
pub use series_repartition::{SeriesRepartitionExec, SeriesRouting};
pub use union_distinct_on::{UnionDistinctOn, UnionDistinctOnExec, UnionDistinctOnStream};
pub type Millisecond = <TimestampMillisecondType as ArrowPrimitiveType>::Native;
src/promql/src/extension_plan/series_repartition.rs
+710 -98
View File
File diff suppressed because it is too large Load Diff
src/promql/src/extension_plan/series_repartition/tests.rs
+1071 -62
View File
File diff suppressed because it is too large Load Diff
src/promql/src/source.rs
+3
View File
@@ -12,9 +12,12 @@
// See the License for the specific language governing permissions and
// limitations under the License.
mod merge;
#[cfg(test)]
mod tests;
pub(crate) use merge::merge_raw_identity_charging;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
src/promql/src/source/merge.rs
+252
View File
@@ -0,0 +1,252 @@
// Copyright 2023 Greptime Team
//
// Licensed 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.
#[cfg(test)]
mod tests;
use datafusion::arrow::buffer::ScalarBuffer;
use datafusion_common::DataFusionError;
use datatypes::error::InvalidSeriesSnafu;
use datatypes::series::histogram::HistogramView;
use datatypes::series::raw::{RawSeriesChunk, RawTimestamps};
use datatypes::series::samples::{SampleRef, Samples, SeriesFields, SeriesSchema};
/// Rebuilds one raw series from every piece of one complete physical identity, reporting each
/// allocation before it happens.
///
/// Reconstruction holds three intermediate vectors whose sizes depend on how the pieces interleave,
/// so what it needs cannot be known up front. The caller charges `reserve` as the work grows
/// instead of reserving a worst case it may never reach.
pub(crate) fn merge_raw_identity_charging(
schema: &SeriesSchema,
chunks: &[RawSeriesChunk],
reserve: &mut dyn FnMut(usize) -> std::result::Result<(), DataFusionError>,
shrink: &mut dyn FnMut(usize),
) -> std::result::Result<RawSeriesChunk, DataFusionError> {
let invalid = |reason: &str| {
DataFusionError::External(Box::new(
InvalidSeriesSnafu {
reason: reason.to_string(),
}
.build(),
))
};
let Some(first) = chunks.first() else {
return Err(invalid("cannot merge an empty raw identity"));
};
if schema.eval_grid.is_some() {
return Err(invalid("raw series schema cannot have an evaluation grid"));
}
let key = first.primary_key.clone();
let labels = first
.metric
.as_ref()
.ok_or_else(|| invalid("raw identity chunk has no explicit labels"))?
.clone();
for chunk in chunks {
if chunk.primary_key != key {
return Err(invalid("raw identity chunks have different primary keys"));
}
if chunk.metric.as_ref() != Some(&labels) {
return Err(invalid(
"raw identity chunks have different or missing labels",
));
}
if chunk.fields.len() != schema.fields.len()
|| chunk
.fields
.iter()
.any(|field| field.len() != chunk.time.len())
{
return Err(invalid("raw identity chunk shape does not match schema"));
}
if chunk
.time
.as_buffer()
.windows(2)
.any(|pair| pair[0] >= pair[1])
{
return Err(invalid(
"raw identity chunk timestamps are not strictly increasing",
));
}
}
let mut row_count = 0usize;
for chunk in chunks.iter() {
row_count = row_count.saturating_add(chunk.time.len());
}
let chunks_are_ordered_and_disjoint = chunks.windows(2).all(|pair| {
pair[0]
.time
.as_buffer()
.last()
.zip(pair[1].time.as_buffer().first())
.is_some_and(|(last, first)| last < first)
});
let mut transient_workspace_bytes = 0usize;
let mut output_fields = Vec::with_capacity(schema.fields.len());
let times = if chunks_are_ordered_and_disjoint {
reserve(row_count.saturating_mul(std::mem::size_of::<i64>()))?;
let mut times = Vec::with_capacity(row_count);
for chunk in chunks {
times.extend_from_slice(chunk.time.as_buffer());
}
for field_index in 0..schema.fields.len() {
let mut parts = Vec::with_capacity(chunks.len());
for chunk in chunks {
let source = chunk
.fields
.get(field_index)
.ok_or_else(|| invalid("field index out of bounds"))?;
reserve(std::mem::size_of::<Samples>())?;
parts.push(
source
.slice(0, chunk.time.len())
.map_err(|error| DataFusionError::External(Box::new(error)))?,
);
}
output_fields.push(
Samples::try_from_chunks(parts)
.map_err(|error| DataFusionError::External(Box::new(error)))?,
);
}
times
} else {
transient_workspace_bytes = row_count
.saturating_mul(std::mem::size_of::<(i64, usize, usize)>())
.saturating_mul(2);
reserve(row_count.saturating_mul(std::mem::size_of::<(i64, usize, usize)>()))?;
let mut rows = Vec::with_capacity(row_count);
for (chunk_index, chunk) in chunks.iter().enumerate() {
for row_index in 0..chunk.time.len() {
rows.push((chunk.time.as_buffer()[row_index], chunk_index, row_index));
}
}
rows.sort_unstable();
reserve(
rows.len()
.saturating_mul(std::mem::size_of::<(i64, usize, usize)>()),
)?;
let mut selected = Vec::with_capacity(rows.len());
let mut index = 0;
while index < rows.len() {
let mut end = index + 1;
while end < rows.len() && rows[end].0 == rows[index].0 {
end += 1;
}
for duplicate in &rows[index + 1..end] {
for field_index in 0..schema.fields.len() {
if !samples_equal(
chunks[rows[index].1]
.fields
.get(field_index)
.ok_or_else(|| invalid("field index out of bounds"))?
.sample(rows[index].2),
chunks[duplicate.1]
.fields
.get(field_index)
.ok_or_else(|| invalid("field index out of bounds"))?
.sample(duplicate.2),
) {
return Err(invalid("duplicate raw timestamp has conflicting samples"));
}
}
}
selected.push(rows[index]);
index = end;
}
reserve(selected.len().saturating_mul(std::mem::size_of::<i64>()))?;
let mut times = Vec::with_capacity(selected.len());
for row in &selected {
times.push(row.0);
}
for field_index in 0..schema.fields.len() {
let mut parts = Vec::new();
let mut start = 0;
while start < selected.len() {
let (chunk_index, row_index) = (selected[start].1, selected[start].2);
let mut end = start + 1;
while end < selected.len()
&& selected[end].1 == chunk_index
&& selected[end].2 == selected[end - 1].2 + 1
{
end += 1;
}
let source = chunks[chunk_index]
.fields
.get(field_index)
.ok_or_else(|| invalid("field index out of bounds"))?;
reserve(std::mem::size_of::<Samples>())?;
parts.push(
source
.slice(row_index, end - start)
.map_err(|error| DataFusionError::External(Box::new(error)))?,
);
start = end;
}
output_fields.push(
Samples::try_from_chunks(parts)
.map_err(|error| DataFusionError::External(Box::new(error)))?,
);
}
times
};
let result = RawSeriesChunk::try_new(
schema,
key,
Some(labels),
RawTimestamps::try_new(ScalarBuffer::from(times))
.map_err(|error| DataFusionError::External(Box::new(error)))?,
SeriesFields::try_from_vec(output_fields)
.map_err(|error| DataFusionError::External(Box::new(error)))?,
true,
)
.map_err(|error| DataFusionError::External(Box::new(error)))?;
// Only the general merge path allocates transient row-index workspaces.
shrink(transient_workspace_bytes);
Ok(result)
}
fn samples_equal(left: SampleRef<'_>, right: SampleRef<'_>) -> bool {
match (left, right) {
(SampleRef::Missing, SampleRef::Missing) => true,
(SampleRef::Float(a), SampleRef::Float(b)) => a.to_bits() == b.to_bits(),
(SampleRef::Histogram(a), SampleRef::Histogram(b)) => histograms_equal(a, b),
_ => false,
}
}
fn histograms_equal(a: HistogramView<'_>, b: HistogramView<'_>) -> bool {
a.schema == b.schema
&& a.zero_threshold.to_bits() == b.zero_threshold.to_bits()
&& a.sum.to_bits() == b.sum.to_bits()
&& a.count.to_bits() == b.count.to_bits()
&& a.zero_count.to_bits() == b.zero_count.to_bits()
&& a.reset_hint == b.reset_hint
&& a.start_timestamp == b.start_timestamp
&& float_slices_equal(a.custom_values, b.custom_values)
&& a.positive.span_offsets == b.positive.span_offsets
&& a.positive.span_lengths == b.positive.span_lengths
&& float_slices_equal(a.positive.buckets, b.positive.buckets)
&& a.negative.span_offsets == b.negative.span_offsets
&& a.negative.span_lengths == b.negative.span_lengths
&& float_slices_equal(a.negative.buckets, b.negative.buckets)
}
fn float_slices_equal(a: &[f64], b: &[f64]) -> bool {
a.len() == b.len() && a.iter().zip(b).all(|(a, b)| a.to_bits() == b.to_bits())
}
src/promql/src/source/merge/tests.rs
+702
View File
@@ -0,0 +1,702 @@
// Copyright 2023 Greptime Team
//
// Licensed 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.
use std::sync::Arc;
use datafusion::arrow::array::{Float64Array, UInt32Array};
use datafusion::arrow::buffer::Buffer;
use datatypes::series::histogram::{BucketSideView, CounterResetHint, HistogramBatchBuilder};
use datatypes::series::raw::{RawSeriesChunk, RawTimestamps};
use datatypes::series::samples::{Samples, SeriesFields, SeriesSchema};
use datatypes::series::{Label, SeriesLabels};
use super::merge_raw_identity_charging;
/// The same merge, with accounting the tests do not measure.
fn merge_raw_identity(
schema: &SeriesSchema,
chunks: &[RawSeriesChunk],
) -> datafusion::common::Result<RawSeriesChunk> {
merge_raw_identity_charging(schema, chunks, &mut |_| Ok(()), &mut |_| {})
}
fn schema(count: usize) -> SeriesSchema {
SeriesSchema {
fields: (0..count)
.map(|i| datatypes::series::samples::FieldSpec {
name: Arc::from(format!("f{i}")),
})
.collect::<Vec<_>>()
.into(),
eval_grid: None,
}
}
fn histogram(sum: f64, threshold: f64) -> Samples {
let mut builder = HistogramBatchBuilder::new();
let side = BucketSideView {
span_offsets: &[],
span_lengths: &[],
buckets: &[],
};
let index = builder
.push(datatypes::series::histogram::HistogramView {
schema: 1,
zero_threshold: threshold,
sum,
count: 1.0,
zero_count: 0.0,
reset_hint: CounterResetHint::Unknown,
start_timestamp: None,
custom_values: &[],
positive: side,
negative: side,
})
.unwrap();
Samples::try_from_histogram(
UInt32Array::from(vec![Some(index)]),
builder.finish().unwrap(),
)
.unwrap()
}
fn labels(value: &str) -> SeriesLabels {
SeriesLabels::try_new(
Some(Arc::from("metric")),
vec![Label {
name: Arc::from("tag"),
value: Arc::from(value),
}],
)
.unwrap()
}
fn chunk(
key: &[u8],
label: &str,
times: Vec<i64>,
fields: Vec<Vec<Option<f64>>>,
) -> RawSeriesChunk {
RawSeriesChunk::try_new(
&schema(fields.len()),
Buffer::from(key.to_vec()),
Some(labels(label)),
RawTimestamps::try_new(times.into()).unwrap(),
SeriesFields::try_from_vec(
fields
.into_iter()
.map(|values| Samples::from_float(Float64Array::from(values)))
.collect(),
)
.unwrap(),
false,
)
.unwrap()
}
#[test]
fn interleaved_pieces_merge_in_time_order() {
let schema = schema(2);
let result = merge_raw_identity(
&schema,
&[
chunk(
b"k",
"v",
vec![1, 3],
vec![vec![Some(10.), Some(30.)], vec![Some(1.), Some(3.)]],
),
chunk(
b"k",
"v",
vec![2, 4],
vec![vec![Some(20.), Some(40.)], vec![Some(2.), Some(4.)]],
),
],
)
.unwrap();
assert_eq!(result.time.as_buffer().as_ref(), &[1, 2, 3, 4]);
assert_eq!(
result
.fields
.get(0)
.unwrap()
.iter()
.map(|s| match s {
datatypes::series::samples::SampleRef::Float(v) => v,
_ => 0.,
})
.collect::<Vec<_>>(),
vec![10., 20., 30., 40.]
);
assert!(result.is_last);
}
#[test]
fn ordered_disjoint_pieces_concatenate_without_row_workspace() {
let schema = schema(1);
let chunks = [
chunk(b"k", "v", vec![1, 2], vec![vec![Some(10.), Some(20.)]]),
chunk(b"k", "v", vec![3, 4], vec![vec![Some(30.), Some(40.)]]),
];
let result = merge_raw_identity(&schema, &chunks).unwrap();
assert_eq!(result.time.as_buffer().as_ref(), &[1, 2, 3, 4]);
assert_eq!(result.time.len(), 4);
assert_eq!(
result
.fields
.get(0)
.unwrap()
.iter()
.map(|sample| match sample {
datatypes::series::samples::SampleRef::Float(value) => value,
_ => 0.,
})
.collect::<Vec<_>>(),
vec![10., 20., 30., 40.]
);
let mut reserved = 0usize;
let mut shrunk = 0usize;
merge_raw_identity_charging(
&schema,
&chunks,
&mut |bytes| {
reserved += bytes;
Ok(())
},
&mut |bytes| shrunk += bytes,
)
.unwrap();
assert_eq!(
reserved,
4 * std::mem::size_of::<i64>() + 2 * std::mem::size_of::<Samples>()
);
assert_eq!(shrunk, 0);
assert!(shrunk <= reserved);
}
#[test]
fn touching_ordered_chunks_use_general_duplicate_path() {
let schema = schema(1);
let chunks = [
chunk(b"k", "v", vec![1, 2], vec![vec![Some(10.), Some(20.)]]),
chunk(b"k", "v", vec![2, 3], vec![vec![Some(20.), Some(30.)]]),
];
let mut reserved = 0usize;
let mut shrunk = 0usize;
let result = merge_raw_identity_charging(
&schema,
&chunks,
&mut |bytes| {
reserved += bytes;
Ok(())
},
&mut |bytes| shrunk += bytes,
)
.unwrap();
assert_eq!(result.time.as_buffer().as_ref(), &[1, 2, 3]);
assert_eq!(shrunk, 2 * 4 * std::mem::size_of::<(i64, usize, usize)>());
assert!(shrunk <= reserved);
}
#[test]
fn equal_duplicates_collapse_but_bit_differences_conflict() {
assert!(
merge_raw_identity(
&schema(1),
&[
chunk(b"k", "v", vec![1], vec![vec![Some(1.)]]),
chunk(b"k", "v", vec![1], vec![vec![Some(1.)]])
]
)
.is_ok()
);
for (left, right) in [
(-0.0, 0.0),
(
f64::from_bits(0x7ff8_0000_0000_0001),
f64::from_bits(0x7ff8_0000_0000_0002),
),
] {
assert!(
merge_raw_identity(
&schema(1),
&[
chunk(b"k", "v", vec![1], vec![vec![Some(left)]]),
chunk(b"k", "v", vec![1], vec![vec![Some(right)]])
]
)
.is_err()
);
}
assert!(
merge_raw_identity(
&schema(1),
&[
chunk(b"k", "v", vec![1], vec![vec![None]]),
chunk(b"k", "v", vec![1], vec![vec![Some(1.)]])
]
)
.is_err()
);
}
#[test]
fn rejects_different_keys_labels_and_internal_duplicates() {
assert!(
merge_raw_identity(
&schema(1),
&[
chunk(b"a", "v", vec![1], vec![vec![Some(1.)]]),
chunk(b"b", "v", vec![2], vec![vec![Some(2.)]])
]
)
.is_err()
);
assert!(
merge_raw_identity(
&schema(1),
&[
chunk(b"a", "v", vec![1], vec![vec![Some(1.)]]),
chunk(b"a", "other", vec![2], vec![vec![Some(2.)]])
]
)
.is_err()
);
assert!(RawTimestamps::try_new(vec![1, 1].into()).is_err());
let mut no_labels = chunk(b"a", "v", vec![1], vec![vec![Some(1.)]]);
no_labels.metric = None;
assert!(merge_raw_identity(&schema(1), &[no_labels]).is_err());
}
#[test]
fn histogram_duplicates_compare_every_float_property() {
let schema = schema(1);
let make = |sample| RawSeriesChunk {
primary_key: Buffer::from(b"k".to_vec()),
metric: Some(labels("v")),
time: RawTimestamps::try_new(vec![1].into()).unwrap(),
fields: SeriesFields::one(sample),
is_last: false,
};
assert!(
merge_raw_identity(
&schema,
&[make(histogram(2.0, 0.5)), make(histogram(2.0, 0.5))]
)
.is_ok()
);
assert!(
merge_raw_identity(
&schema,
&[make(histogram(2.0, 0.5)), make(histogram(3.0, 0.5))]
)
.is_err()
);
assert!(
merge_raw_identity(
&schema,
&[make(histogram(2.0, 0.5)), make(histogram(2.0, -0.0))]
)
.is_err()
);
}
/// Builds a histogram with configurable comparison properties for duplicate-conflict tests.
fn histogram_with(view: datatypes::series::histogram::HistogramView<'_>) -> Samples {
let mut builder = HistogramBatchBuilder::new();
let index = builder.push(view).unwrap();
Samples::try_from_histogram(
UInt32Array::from(vec![Some(index)]),
builder.finish().unwrap(),
)
.unwrap()
}
/// Two histograms are duplicates only when every property matches.
///
/// Bucket values are independently free within a valid layout and are compared directly. The
/// span/cursor layout cannot be varied independently of bucket count: each layout must have
/// nonnegative lengths summing to the bucket count and a valid cursor. The cases below vary each
/// offset and length while preserving valid layouts and equal bucket counts.
#[test]
fn histogram_duplicates_compare_every_property() {
let schema = schema(1);
let make = |sample| RawSeriesChunk {
primary_key: Buffer::from(b"k".to_vec()),
metric: Some(labels("v")),
time: RawTimestamps::try_new(vec![1].into()).unwrap(),
fields: SeriesFields::one(sample),
is_last: false,
};
let positive = BucketSideView {
span_offsets: &[0, 0],
span_lengths: &[1, 1],
buckets: &[1.0, 2.0],
};
let negative = BucketSideView {
span_offsets: &[0, 0],
span_lengths: &[1, 1],
buckets: &[3.0, 4.0],
};
let build = |schema,
count,
zero_count,
sum,
zero_threshold,
positive,
negative,
reset_hint,
start_timestamp,
custom_values| {
histogram_with(datatypes::series::histogram::HistogramView {
schema,
count,
zero_count,
sum,
zero_threshold,
positive,
negative,
reset_hint,
start_timestamp,
custom_values,
})
};
let baseline = || {
build(
1,
1.0,
0.0,
2.0,
0.5,
positive,
negative,
CounterResetHint::Unknown,
None,
&[],
)
};
assert!(
merge_raw_identity(&schema, &[make(baseline()), make(baseline())]).is_ok(),
"identical histograms are duplicates"
);
let positive_buckets_changed = BucketSideView {
buckets: &[1.0, 2.5],
..positive
};
let negative_buckets_changed = BucketSideView {
buckets: &[3.0, 4.5],
..negative
};
// Same span count and lengths as the baseline, different cursors: only the offsets differ.
let positive_offsets_changed = BucketSideView {
span_offsets: &[1, 0],
..positive
};
let negative_offsets_changed = BucketSideView {
span_offsets: &[1, 0],
..negative
};
// Same cursors and same total bucket count (2), different split into spans. A zero-length
// span is a valid layout, so only the lengths differ here.
let positive_lengths_changed = BucketSideView {
span_lengths: &[2, 0],
..positive
};
let negative_lengths_changed = BucketSideView {
span_lengths: &[2, 0],
..negative
};
let cases: Vec<(&str, Samples)> = vec![
(
"schema (1 vs 2)",
build(
2,
1.0,
0.0,
2.0,
0.5,
positive,
negative,
CounterResetHint::Unknown,
None,
&[],
),
),
(
"count (1.0 vs 2.0)",
build(
1,
2.0,
0.0,
2.0,
0.5,
positive,
negative,
CounterResetHint::Unknown,
None,
&[],
),
),
(
"zero_count (0.0 vs 1.0)",
build(
1,
1.0,
1.0,
2.0,
0.5,
positive,
negative,
CounterResetHint::Unknown,
None,
&[],
),
),
(
"sum (2.0 vs 3.0)",
build(
1,
1.0,
0.0,
3.0,
0.5,
positive,
negative,
CounterResetHint::Unknown,
None,
&[],
),
),
(
"zero_threshold (0.5 vs -0.0)",
build(
1,
1.0,
0.0,
2.0,
-0.0,
positive,
negative,
CounterResetHint::Unknown,
None,
&[],
),
),
(
"positive bucket value (2.0 vs 2.5)",
build(
1,
1.0,
0.0,
2.0,
0.5,
positive_buckets_changed,
negative,
CounterResetHint::Unknown,
None,
&[],
),
),
(
"negative bucket value (4.0 vs 4.5)",
build(
1,
1.0,
0.0,
2.0,
0.5,
positive,
negative_buckets_changed,
CounterResetHint::Unknown,
None,
&[],
),
),
(
"positive span offset ([0, 0] vs [1, 0])",
build(
1,
1.0,
0.0,
2.0,
0.5,
positive_offsets_changed,
negative,
CounterResetHint::Unknown,
None,
&[],
),
),
(
"negative span offset ([0, 0] vs [1, 0])",
build(
1,
1.0,
0.0,
2.0,
0.5,
positive,
negative_offsets_changed,
CounterResetHint::Unknown,
None,
&[],
),
),
(
"positive span lengths ([1, 1] vs [2, 0])",
build(
1,
1.0,
0.0,
2.0,
0.5,
positive_lengths_changed,
negative,
CounterResetHint::Unknown,
None,
&[],
),
),
(
"negative span lengths ([1, 1] vs [2, 0])",
build(
1,
1.0,
0.0,
2.0,
0.5,
positive,
negative_lengths_changed,
CounterResetHint::Unknown,
None,
&[],
),
),
(
"reset hint (Unknown vs CounterReset)",
build(
1,
1.0,
0.0,
2.0,
0.5,
positive,
negative,
CounterResetHint::CounterReset,
None,
&[],
),
),
(
"start timestamp (None vs Some(7))",
build(
1,
1.0,
0.0,
2.0,
0.5,
positive,
negative,
CounterResetHint::Unknown,
Some(7),
&[],
),
),
(
"custom values ([] vs [1.5])",
build(
1,
1.0,
0.0,
2.0,
0.5,
positive,
negative,
CounterResetHint::Unknown,
None,
&[1.5],
),
),
];
for (what, changed) in cases {
assert!(
merge_raw_identity(&schema, &[make(baseline()), make(changed)]).is_err(),
"two histograms differing only in {what} are a conflict, not a duplicate"
);
}
}
#[test]
fn rejects_empty_and_shape_mismatch() {
assert!(merge_raw_identity(&schema(1), &[]).is_err());
let valid = chunk(b"k", "v", vec![1], vec![vec![Some(1.)]]);
assert!(merge_raw_identity(&schema(2), &[valid]).is_err());
let malformed = RawSeriesChunk {
primary_key: Buffer::from(b"k".to_vec()),
metric: Some(labels("v")),
time: RawTimestamps::try_new(vec![1].into()).unwrap(),
fields: SeriesFields::one(Samples::from_float(Float64Array::from(vec![
Some(1.),
Some(2.),
]))),
is_last: false,
};
assert!(merge_raw_identity(&schema(1), &[malformed]).is_err());
}
#[test]
fn reservation_error_is_preserved() {
let error = merge_raw_identity_charging(
&schema(1),
&[chunk(b"k", "v", vec![1], vec![vec![Some(1.)]])],
&mut |_| {
Err(datafusion::common::DataFusionError::Execution(
"reservation refused".into(),
))
},
&mut |_| {},
)
.unwrap_err();
assert!(matches!(
error,
datafusion::common::DataFusionError::Execution(_)
));
}
#[test]
fn merge_releases_transient_row_workspace() {
let mut reserved = 0usize;
let mut shrunk = 0usize;
let result = merge_raw_identity_charging(
&schema(1),
&[
chunk(b"k", "v", vec![1, 3], vec![vec![Some(1.), Some(3.)]]),
chunk(b"k", "v", vec![1, 2], vec![vec![Some(1.), Some(2.)]]),
],
&mut |bytes| {
reserved = reserved.saturating_add(bytes);
Ok(())
},
&mut |bytes| shrunk = shrunk.saturating_add(bytes),
)
.unwrap();
assert_eq!(result.time.len(), 3);
assert_eq!(shrunk, 2 * 4 * std::mem::size_of::<(i64, usize, usize)>());
let retained = reserved - shrunk;
assert!(retained < reserved);
assert!(retained >= result.time.len() * std::mem::size_of::<i64>());
}
src/promql/src/vector_match.rs
+2 -2
View File
@@ -1528,9 +1528,9 @@ fn same_grid(lhs: &SeriesSchema, rhs: &SeriesSchema) -> Result<Arc<EvalGrid>> {
/// Label pairs identifying a match group or a result; empty values are absent, as in
/// Prometheus.
type GroupKey = Vec<(Arc<str>, Arc<str>)>;
pub(crate) type GroupKey = Vec<(Arc<str>, Arc<str>)>;
fn group_key(metric: &SeriesLabels, labels: &MatchingLabels) -> GroupKey {
pub(crate) fn group_key(metric: &SeriesLabels, labels: &MatchingLabels) -> GroupKey {
let mut key: GroupKey = match labels {
MatchingLabels::On(names) => names
.iter()
tests-integration/tests/native_prefix.rs
+45 -15
View File
@@ -55,13 +55,15 @@ fn evaluated_frame_schema(tree: &NativeTree) -> datafusion::common::DFSchemaRef
)
}
// Reproduction from the review: 16 hosts × 20k samples, `rate[10s]`, and 16 DN
// partitions produces four batches per chain and fails while unpacking because
// `CoalescePartitionsExec` interleaves their batches, so a chain is cut and the frames no
// longer decode. This test asserts that failure, which is why it runs rather than being
// ignored: it keeps the boundary honest until the frontend region merge becomes
// chain-atomic, and it is the test to change when that fix lands. The parallelism-one
// path is exercised by `native_prefix_executes_over_real_mito_region`.
// Sends `rate[10s]` on a one-point evaluation grid for 16 hosts over 16 region-scan
// partitions to the datanode's remote-read entry point, then checks that the evaluated frames
// decode and each host reports the rate its samples produce. The former decode failure no longer
// reproduces because #10023 makes region package completion close its pending final raw chunk,
// so a chain's value and close are no longer latched apart in separate batches by the
// batch-interleaving `CoalescePartitionsExec`. That coalescing is still not chain-atomic: a chain
// spanning batches can still be cut. This one-point grid gives each host one evaluated value plus
// a possible empty closing chunk, so it does not exercise a chain spanning many batches. The
// parallelism-one path is exercised by `native_prefix_executes_over_real_mito_region`.
#[tokio::test(flavor = "multi_thread")]
async fn native_prefix_multi_partition_series_chains() {
let instance = GreptimeDbStandaloneBuilder::new("native_prefix_partitioned_chains")
@@ -205,20 +207,48 @@ async fn native_prefix_multi_partition_series_chains() {
}
let schema = batches.first().expect("evaluated frames").schema.clone();
let batches = RecordBatches::try_new(schema, batches).unwrap();
let error = common_query::series_frame::unpack_evaluated_series(
// Package completion closes each region's pending final raw chunk, so the evaluated frames
// decode; ordinary partition coalescing still does not keep chains atomic. Check that every
// host's chain closes exactly once and carries the rate its own samples produce.
let decoded = common_query::series_frame::unpack_evaluated_series(
Box::pin(DfRecordBatchStreamAdapter::new(batches.as_stream())),
None,
)
.unwrap()
.try_collect::<Vec<_>>()
.await
.unwrap_err();
assert!(
error
.to_string()
.contains("evaluated series identity changed before its chain closed"),
"{error}"
);
.expect("a chain-atomic merge decodes every chain");
let mut closed = std::collections::BTreeMap::<String, (usize, Vec<f64>)>::new();
for (series, is_last) in decoded {
let host = series
.metric
.get("host")
.unwrap_or_else(|| panic!("a decoded series keeps its host label: {:?}", series.metric))
.to_string();
let entry = closed.entry(host).or_default();
if is_last {
entry.0 += 1;
}
for field in series.fields.iter() {
entry
.1
.extend(field.iter().filter_map(|sample| match sample {
datatypes::series::samples::SampleRef::Float(value) => Some(value),
datatypes::series::samples::SampleRef::Histogram(_)
| datatypes::series::samples::SampleRef::Missing => None,
}));
}
}
assert_eq!(16, closed.len(), "one chain per host: {closed:?}");
for (host, (closes, values)) in &closed {
assert_eq!(1, *closes, "`{host}` closes its chain exactly once");
assert_eq!(
vec![1.0],
*values,
"`{host}` reports the rate of its own samples, which a cut chain would lose"
);
}
}
#[tokio::test(flavor = "multi_thread")]