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

test(promql): keep multi-chunk coverage after chain-atomic merge

Browse Source 
#10055 made the partition merge chain-atomic, so a host's chain no longer
arrives as a fixed number of chunks plus one close chunk. Resolve the test
against that semantics: drop the exact chunk-count assertions and keep the
coverage that matters here - every host arrives with all 8,192 grid
positions in order and closes its chain exactly once.

Signed-off-by: discord9 <discord9@163.com>
This commit is contained in:
discord9
2026-11-03 14:07:36 +08:00
parent ce3ef72faf f73a27d779
commit 683b6a672f
17 changed files with 2963 additions and 286 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mito2/AGENTS.md
+1
View File
@@ -69,6 +69,7 @@ filtered `RecordBatch` stream.
- **Series format PoC**: samples are rebuildable local SST sidecars. Optional writer interfaces (`format.rs`) integrate with reconciliation, catalog publication, and deferred purge. Defaults retain the original Parquet writer and disable samples. The opt-in columnar backend (`series_index/columnar.rs`) implements streaming OpenDAL writes and pinned format reads; its wire contract is in `docs/formats/mito-series.md`. Artifact objects record paths and sizes, including the manifest, so reads need no stat. Codecs currently run inline pending scheduling benchmarks. Columnar `SeriesScan` raw reads use scanner-wide preparation in `read/series_scan/prepared.rs`; ordinary record-batch scans retain their SST path. Native reader metadata shares the decoded SST cache reservation; maintenance readers bypass this cache. The paired `experimental_series_index_format` setting selects legacy or columnar artifacts.
- **Series-index coverage** (`series_index/catalog.rs`): `SeriesIndexEntry` stores
compaction-window width and SST summaries keyed by aligned start in both catalogs and Parquet footers.
- **Local index cache**: startup checks a root layout version before workers start; missing or incompatible versions reset the whole rebuildable cache, including catalogs and range indexes. Samples use one generation directory; source SST identity stays in metadata.
- **Columnar samples**: `series_index/samples.rs` implements opt-in per-SST sidecars,
including nullable Float64 fields and preserved sequence/delete rows. Its wire
contract is `docs/formats/mito-samples.md`; time/ID bounds may overlap across
src/mito2/src/series_index/builder.rs
+2 -5
View File
@@ -277,11 +277,8 @@ pub(crate) async fn build_samples(
)?;
let compat = FlatCompatBatch::try_new(&mapper, context.read_format(), false)?;
let generation = store_api::storage::FileId::random();
let prefix = format!(
"{}/samples/{}/{generation}/",
region.region_id.as_u64(),
file_id.file_id()
);
let prefix =
crate::series_index::samples::samples_prefix(region.region_id.as_u64(), generation);
let mut writer = backend
.writer(
version.metadata.clone(),
src/mito2/src/series_index/samples.rs
+6 -4
View File
@@ -172,10 +172,7 @@ impl Manifest {
}
fn prefix(&self) -> String {
format!(
"{}/samples/{}/{}/",
self.region_id, self.source_file_id, self.generation
)
samples_prefix(self.region_id, self.generation)
}
fn path(&self) -> String {
format!("{}manifest.json", self.prefix())
@@ -233,3 +230,8 @@ fn valid_rows(rows: usize, fields: usize) -> bool {
&& (1..=MAX_FIELDS).contains(&fields)
&& rows * (17 + 8 * fields) + fields * rows.div_ceil(8) <= BLOCK_BYTES
}
/// Immutable samples output directory, independent of the source SST identity.
pub(crate) fn samples_prefix(region_id: u64, generation: FileId) -> String {
format!("{region_id}/samples/{generation}/")
}
src/mito2/src/series_index/samples/reader.rs
+2 -2
View File
@@ -103,8 +103,8 @@ impl SamplesReader {
"samples source handle mismatch",
)?;
let path = format!(
"{region}/samples/{}/{}/manifest.json",
entry.file_id, artifact.generation
"{}manifest.json",
crate::series_index::samples::samples_prefix(region, artifact.generation)
);
let mut objects = HashMap::new();
for object in &artifact.objects {
src/mito2/src/series_index/samples/tests.rs
+4 -7
View File
@@ -126,10 +126,7 @@ async fn build(
) -> SamplesFileHandle {
let file_id = FileId::random();
let generation = FileId::random();
let prefix = format!(
"{}/samples/{file_id}/{generation}/",
meta.region_id.as_u64()
);
let prefix = format!("{}/samples/{generation}/", meta.region_id.as_u64());
let mut writer = format
.writer(meta.clone(), store.clone(), &prefix, file_id, generation)
.await
@@ -433,7 +430,7 @@ async fn failed_writer_and_abort_remove_outputs() {
for drop_writer in [false, true] {
let id = FileId::random();
let generation = FileId::random();
let prefix = format!("{}/samples/{id}/{generation}/", m.region_id.as_u64());
let prefix = format!("{}/samples/{generation}/", m.region_id.as_u64());
let mut writer = format
.writer(m.clone(), store.clone(), &prefix, id, generation)
.await
@@ -934,7 +931,7 @@ async fn wide_schema_byte_bound_and_unsupported_input() {
));
let id = FileId::random();
let generation = FileId::random();
let prefix = format!("{}/samples/{id}/{generation}/", m.region_id.as_u64());
let prefix = format!("{}/samples/{generation}/", m.region_id.as_u64());
assert!(
format
.writer(wrong, store.clone(), &prefix, id, generation)
@@ -1019,7 +1016,7 @@ async fn output_failures_and_cancelled_finish_clean_up() {
.unwrap();
let id = FileId::random();
let generation = FileId::random();
let prefix = format!("{}/samples/{id}/{generation}/", m.region_id.as_u64());
let prefix = format!("{}/samples/{generation}/", m.region_id.as_u64());
let mut writer = format
.writer(
m.clone(),
src/mito2/src/worker.rs
+102 -1
View File
@@ -526,7 +526,44 @@ async fn series_index_store_from_config(
}
let root = Path::new(data_home).join("series_index");
new_fs_cache_store(&root.to_string_lossy()).await.map(Some)
let store = new_fs_cache_store(&root.to_string_lossy()).await?;
initialize_series_index_cache(&store).await?;
Ok(Some(store))
}
/// Bump when the rebuildable local index layout becomes incompatible.
const SERIES_INDEX_CACHE_VERSION: &str = "1";
/// Resets incompatible caches before any worker or reader can use their catalogs.
async fn initialize_series_index_cache(store: &ObjectStore) -> Result<()> {
let version = match store.read("version").await {
Ok(bytes) => Some(bytes.to_bytes()),
Err(error) if error.kind() == object_store::ErrorKind::NotFound => None,
Err(error) => return Err(error).context(crate::error::OpenDalSnafu),
};
if version.as_deref() == Some(SERIES_INDEX_CACHE_VERSION.as_bytes()) {
return Ok(());
}
common_telemetry::info!("Resetting series-index cache: missing or incompatible layout version");
// Preserve the cache root, which may be a mount point.
let entries = store.list("/").await.context(crate::error::OpenDalSnafu)?;
for entry in entries {
// Listings can include the directory itself.
if entry.path() == "/" {
continue;
}
store
.delete_with(entry.path())
.recursive(true)
.await
.context(crate::error::OpenDalSnafu)?;
}
store
.write("version", SERIES_INDEX_CACHE_VERSION)
.await
.context(crate::error::OpenDalSnafu)?;
common_telemetry::info!("Series-index cache reset complete");
Ok(())
}
pub async fn write_cache_from_config(
@@ -1699,6 +1736,70 @@ mod tests {
use super::*;
use crate::test_util::TestEnv;
#[tokio::test]
async fn test_series_index_cache_layout_reset() {
let parent = common_test_util::temp_dir::create_temp_dir("series-cache-version");
let root = parent.path().join("series_index");
let sentinel = parent.path().join("sst");
tokio::fs::write(&sentinel, b"source data").await.unwrap();
let store = crate::access_layer::new_fs_cache_store(&root.to_string_lossy())
.await
.unwrap();
// Keep the original directory alive so its inode cannot be reused after deletion.
#[cfg(unix)]
let root_handle = tokio::fs::File::open(&root).await.unwrap();
for version in [None, Some("0"), Some("invalid")] {
if let Some(version) = version {
store.write("version", version).await.unwrap();
}
store.write("stale", "old").await.unwrap();
store
.write("42/samples/old/source/part-0.samples", "old")
.await
.unwrap();
store
.write("42/samples.json", "stale references")
.await
.unwrap();
super::initialize_series_index_cache(&store).await.unwrap();
assert!(!store.exists("stale").await.unwrap());
assert_eq!(
store.read("version").await.unwrap().to_bytes().as_ref(),
SERIES_INDEX_CACHE_VERSION.as_bytes()
);
#[cfg(unix)]
{
use std::os::unix::fs::MetadataExt;
let original = root_handle.metadata().await.unwrap();
let current = tokio::fs::metadata(&root).await.unwrap();
assert_eq!(
(original.dev(), original.ino()),
(current.dev(), current.ino())
);
}
assert!(!store.exists("42/samples.json").await.unwrap());
assert!(
!store
.exists("42/samples/old/source/part-0.samples")
.await
.unwrap()
);
store
.write("42/series/current/manifest.json", "current")
.await
.unwrap();
super::initialize_series_index_cache(&store).await.unwrap();
assert!(
store
.exists("42/series/current/manifest.json")
.await
.unwrap()
);
assert_eq!(tokio::fs::read(&sentinel).await.unwrap(), b"source data");
}
}
#[test]
fn test_region_id_to_index() {
let num_workers = 4;
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
@@ -1534,9 +1534,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()