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): carry the native plan with serde instead of a hand-written codec

Browse Source 
`native/wire.rs` encoded `NativeTree` by hand: fixed node tags, fixed field
layouts, both directions written out, and a `check_shape` that narrowed the
transport to `Input -> Instant|Window -> Map(Math*)`. The serialization layer
therefore decided which plans could exist. `abs(ceil(x))` compiled but could
not be transported, and every new node kind cost a second codec path plus a
new whitelist entry. Widening the whitelist to nested maps maintained that
interpreter rather than removing it.

The file is deleted. The plan types derive serde and the codec is postcard
1.1.3 -- bincode is unusable here because RUSTSEC-2025-0141 marks the whole
crate unmaintained with no patched version. postcard is maintained, has a
documented stable wire format since 1.0, and writes an f64 as its bit pattern
in eight little-endian bytes, so NaN payloads, signed zero, infinities and
subnormals survive exactly.

What remains hand-written is framing only: magic `NPRQ`, a version, a payload
length. The version is rejected before the payload is deserialized, so an
unknown version is a clean error instead of a silent misparse, and oversize,
trailing bytes and truncation are separate errors. Decoding ends at
`NativeTree::try_new`, so a tree the compiler would not build cannot be
decoded.

Three types keep written impls because they hold invariants. `EvalGrid`
serializes its `GridSpec` and rebuilds through `try_deferred` with an empty
cache -- the grid is never serialized and never materialized on decode.
`SeriesLabels` rebuilds through `try_new`, keeping the sorting and the
duplicate / `__name__`-as-a-tag rejection. `LabelRewrite` travels as its
parameters with the regex recompiled, and fails to decode if the pattern no
longer compiles; compiled internals never travel.

`FusedSeries` now carries its output schema and accepts `PlainRows` and
`partial_fold`, which it used to refuse. A carried schema is only honest if it
is checked, so decode validates every column role, the exact time type and
nullability, missing or extra value columns, and the scalar-root tag rule; a
`SeriesFrame` schema must equal the frame layout the tree derives, and a
partial export must match `state_schema(binding.tags)` including tag order.

Identity stops being display text. `FusedSeries` compared `tree.to_string()`
with a field patch, `AggregateFinalize` hashed the rendering, and
`LabelRewrite` compared its own output; all three are structural now, with
floats compared by `to_bits()` so that signed zero and NaN payloads stay
reflexive under `Eq`. `same_binding` compared `Arc::ptr_eq`, which put an
allocation address into a transportable plan; it compares the ordered fields
and the optional grid spec now, distinguishing raw from evaluated.

Tests assert the properties rather than the byte layout: a non-canonical NaN
payload round-trips by bits, an unknown version is rejected before decode even
when the payload is garbage, non-prefix trees (binary, set, aggregate, scalar,
evaluated input) round-trip, thirteen malformed schemas are each rejected with
their own message, and a multi-chunk chain over a real region is checked
against its exact sample count and contiguous timestamps.

The multi-partition integration test is split in two and labelled honestly. It
used to assert a failure, on the theory that coalescing interleaved multi-chunk
chains; its fixture's grid has one position, so no chain has more than one
chunk and that assertion was already failing at the base commit. One test now
covers the single-position shape and says it does not exercise interleaving;
the other covers a 8192-position grid that does.

Deserialization is bounded by depth, not only by bytes. The derived recursion
walks boxed nodes, and a payload under the 1 MiB limit can nest far deeper than
the stack allows, which aborts the process instead of returning an error. A
thread-local budget guards `NativeNode`'s deserializer, so a hostile payload is
rejected before a complete tree exists. `NativeTree::Deserialize` also routes
through `try_new`, so no path can materialize a tree the constructor would
refuse. Measured under postcard with no `RUST_MIN_STACK` override: 128 nodes
decode, 129 return a decode error. Postcard collapses custom serde errors to
its unit variant, so that rejection does not name the nesting limit; the other
envelope failures (version, size, trailing bytes, truncation) each keep their
own message.

Signed-off-by: discord9 <discord9@163.com>
This commit is contained in:
discord9
2026-10-31 01:21:05 +08:00
parent e7044931fb
commit abc5868073
24 changed files with 1615 additions and 1542 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/datatypes/Cargo.toml
+1 -1
View File
@@ -33,7 +33,7 @@ parquet-variant-compute.workspace = true
parquet-variant-json.workspace = true
paste.workspace = true
regex.workspace = true
serde.workspace = true
serde = { workspace = true, features = ["rc"] }
serde_json.workspace = true
snafu.workspace = true
sqlparser.workspace = true
src/datatypes/src/series.rs
+21 -3
View File
@@ -36,7 +36,7 @@ use crate::error::{InvalidSeriesSnafu, Result};
/// evaluated streams use current name/label identity, not storage primary keys or object address/hash.
/// This ordering
/// is not a query-global ordering, partition-ownership rule, or finality guarantee.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize, Hash)]
pub enum SeriesOrdering {
/// Series chunks may interleave; positions within each series remain strictly increasing.
Interleaved,
@@ -47,7 +47,7 @@ pub enum SeriesOrdering {
}
/// A label with shared UTF-8 name and value strings.
#[derive(Clone, Debug, PartialEq, Eq)]
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize, Hash)]
pub struct Label {
/// Label name.
pub name: Arc<str>,
@@ -61,12 +61,30 @@ pub struct Label {
/// The metric name is only the `__name__` value (see [`Self::name`]); this type is
/// the whole label set a PromQL series shows. It is neither a canonical
/// matching/grouping identity nor a storage identity (read key / primary key).
#[derive(Clone, Debug, PartialEq, Eq)]
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub struct SeriesLabels {
name: Option<Arc<str>>,
tags: Arc<[Label]>,
}
impl serde::Serialize for SeriesLabels {
fn serialize<S: serde::Serializer>(
&self,
serializer: S,
) -> std::result::Result<S::Ok, S::Error> {
(&self.name, self.tags.as_ref()).serialize(serializer)
}
}
impl<'de> serde::Deserialize<'de> for SeriesLabels {
fn deserialize<D: serde::Deserializer<'de>>(
deserializer: D,
) -> std::result::Result<Self, D::Error> {
let (name, tags): (Option<Arc<str>>, Vec<Label>) =
serde::Deserialize::deserialize(deserializer)?;
Self::try_new(name, tags).map_err(serde::de::Error::custom)
}
}
impl SeriesLabels {
/// Creates a metric name, sorting labels and rejecting duplicates or `__name__` tags.
pub fn try_new(name: Option<Arc<str>>, mut tags: Vec<Label>) -> Result<Self> {
src/datatypes/src/series/samples.rs
+2 -2
View File
@@ -664,12 +664,12 @@ impl SeriesFields {
}
/// Shared name and position of each output field.
#[derive(Clone, Debug)]
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, PartialEq, Hash, Eq)]
pub struct FieldSpec {
pub name: Arc<str>,
}
/// Shared schema describing series fields.
#[derive(Clone, Debug)]
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, PartialEq, Hash, Eq)]
pub struct SeriesSchema {
pub fields: Arc<[FieldSpec]>,
/// Shared grid for regular series; its `GridSpec` must match each series grid axis. Explicit axes require `None`.
src/datatypes/src/series/time.rs
+30
View File
@@ -90,6 +90,36 @@ pub struct EvalGrid {
spec: GridSpec,
timestamps: OnceCell<ScalarBuffer<i64>>,
}
impl serde::Serialize for EvalGrid {
fn serialize<S: serde::Serializer>(
&self,
serializer: S,
) -> std::result::Result<S::Ok, S::Error> {
self.spec.serialize(serializer)
}
}
impl<'de> serde::Deserialize<'de> for EvalGrid {
fn deserialize<D: serde::Deserializer<'de>>(
deserializer: D,
) -> std::result::Result<Self, D::Error> {
let spec = GridSpec::deserialize(deserializer)?;
Self::try_deferred(spec).map_err(serde::de::Error::custom)
}
}
impl PartialEq for EvalGrid {
fn eq(&self, other: &Self) -> bool {
self.spec == other.spec
}
}
impl Eq for EvalGrid {}
impl std::hash::Hash for EvalGrid {
fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
self.spec.start_ms.hash(state);
self.spec.step_ms.hash(state);
self.spec.len.hash(state)
}
}
impl EvalGrid {
/// Allocates the grid timestamps once for sharing across result series.
src/promql/Cargo.toml
+2
View File
@@ -28,6 +28,8 @@ prometheus.workspace = true
prost.workspace = true
regex.workspace = true
snafu.workspace = true
serde = { workspace = true, features = ["rc"] }
postcard.workspace = true
store-api.workspace = true
table.workspace = true
tokio.workspace = true
src/promql/src/extension_plan/aggregate_finalize.rs
+14 -20
View File
@@ -226,10 +226,7 @@ impl AggregateFinalize {
)?))
}
/// The comparable form of the split this node carries, for its plan line and its identity.
///
/// [`BoundAggregate`] carries no `Eq`/`Hash` - a quantile's parameter is an `f64` - so its
/// `Debug` rendering stands in, the way [`FusedSeries`] renders the tree it runs.
/// The readable form of the split, for its plan line only.
fn key(&self) -> String {
format!("{:?} {:?} {:?}", self.op.kind, self.op.group, self.op.stage)
}
@@ -284,12 +281,10 @@ impl UserDefinedLogicalNodeCore for AggregateFinalize {
impl PartialEq for AggregateFinalize {
fn eq(&self, other: &Self) -> bool {
// `NativeTree` carries no structural equality, so the rendered suffix stands in for it, as
// [`FusedSeries`] compares the tree it runs.
self.promql == other.promql
&& self.key() == other.key()
&& self.op == other.op
&& self.binding == other.binding
&& self.suffix.to_string() == other.suffix.to_string()
&& self.suffix == other.suffix
&& self.input == other.input
&& self.output_schema == other.output_schema
}
@@ -300,9 +295,9 @@ impl Eq for AggregateFinalize {}
impl Hash for AggregateFinalize {
fn hash<H: Hasher>(&self, state: &mut H) {
self.promql.hash(state);
self.key().hash(state);
self.op.hash(state);
self.binding.hash(state);
self.suffix.to_string().hash(state);
self.suffix.hash(state);
self.input.hash(state);
self.output_schema.hash(state);
}
@@ -310,16 +305,15 @@ impl Hash for AggregateFinalize {
impl PartialOrd for AggregateFinalize {
fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
// The rendered forms stand in for the trees, as they do in equality.
let mine = (self.promql.to_string(), self.key(), self.suffix.to_string());
let theirs = (
other.promql.to_string(),
other.key(),
other.suffix.to_string(),
);
match mine.partial_cmp(&theirs) {
Some(Ordering::Equal) => {}
ord => return ord,
match self.promql.cmp(&other.promql) {
Ordering::Equal => {}
ord => return Some(ord),
}
if self.op != other.op {
return None;
}
if self.suffix != other.suffix {
return None;
}
match self.binding.partial_cmp(&other.binding) {
Some(Ordering::Equal) => {}
src/promql/src/extension_plan/fused_series.rs
+163 -94
View File
@@ -238,19 +238,9 @@ pub struct FusedSeries {
impl FusedSeries {
/// Serializes the native tree and the binding required to reconstruct this plan node.
/// The PromQL text is included for diagnostics only and never drives execution; children
/// remain ordinary Substrait inputs. Version 1 supports only SeriesFrame without partial folds:
/// decoding must rebuild its output schema, and only SeriesFrame derives that schema from the tree.
/// remain ordinary Substrait inputs. The versioned envelope carries the complete output schema;
/// PlainRows and partial-state layouts are validated against their tree and binding on decode.
pub fn serialize(&self) -> DataFusionResult<Vec<u8>> {
if self.format != FusedOutputFormat::SeriesFrame {
return Err(DataFusionError::NotImplemented(
"FusedSeries serialization does not support PlainRows output schema".into(),
));
}
if self.partial_fold.is_some() {
return Err(DataFusionError::NotImplemented(
"FusedSeries serialization does not support partial_fold".into(),
));
}
fn text(out: &mut Vec<u8>, s: &str) -> DataFusionResult<()> {
let n = u32::try_from(s.len())
.map_err(|_| DataFusionError::Plan("FusedSeries string too long".into()))?;
@@ -258,10 +248,10 @@ impl FusedSeries {
out.extend(s.as_bytes());
Ok(())
}
let tree = crate::native::wire::encode_prefix(&self.tree)
let tree = crate::native::plan::encode_tree(&self.tree)
.map_err(|e| DataFusionError::Plan(format!("cannot encode FusedSeries tree: {e}")))?;
let mut out = Vec::new();
out.extend(1u32.to_le_bytes());
out.extend(2u32.to_le_bytes());
out.extend(
u32::try_from(tree.len())
.map_err(|_| DataFusionError::Plan("FusedSeries tree too large".into()))?
@@ -294,12 +284,33 @@ impl FusedSeries {
for tag in &self.binding.tags {
text(&mut out, tag)?;
}
out.push(1); // explicit stable tag: SeriesFrame
out.push(match self.format {
FusedOutputFormat::PlainRows => 0,
FusedOutputFormat::SeriesFrame => 1,
});
text(&mut out, &self.promql)?;
match &self.partial_fold {
Some(op) => {
out.push(1);
let encoded =
{ postcard::to_stdvec(op).map_err(|e| DataFusionError::Plan(e.to_string()))? };
out.extend((encoded.len() as u32).to_le_bytes());
out.extend(encoded);
}
None => out.push(0),
}
// Preserve the complete output schema. DataFusion's schema serialization is the canonical
// schema representation used for this envelope.
let schema = postcard::to_stdvec(self.output_schema.as_arrow()).map_err(|e| {
DataFusionError::Plan(format!("cannot serialize FusedSeries schema: {e}"))
})?;
out.extend((schema.len() as u32).to_le_bytes());
out.extend(schema);
Ok(out)
}
/// Rebuilds a serialized SeriesFrame node. Substrait attaches its child plans separately.
/// Rebuilds a serialized FusedSeries node, validating its carried schema against the tree,
/// output format, and optional partial fold. Substrait attaches children separately.
pub fn deserialize(bytes: &[u8]) -> DataFusionResult<Self> {
struct Reader<'a> {
b: &'a [u8],
@@ -340,13 +351,13 @@ impl FusedSeries {
}
let mut r = Reader { b: bytes, p: 0 };
let version = r.u32()?;
if version != 1 {
if version != 2 {
return Err(DataFusionError::Plan(format!(
"unsupported FusedSeries envelope version {version}"
)));
}
let n = r.u32()? as usize;
let tree = crate::native::wire::decode_prefix(r.take(n)?)
let tree = crate::native::plan::decode_tree(r.take(n)?)
.map_err(|e| DataFusionError::Plan(format!("cannot decode FusedSeries tree: {e}")))?;
let grid = Arc::new(
EvalGrid::try_deferred(
@@ -376,6 +387,7 @@ impl FusedSeries {
tags.push(r.text()?)
}
let format = match r.u8()? {
0 => FusedOutputFormat::PlainRows,
1 => FusedOutputFormat::SeriesFrame,
x => {
return Err(DataFusionError::Plan(format!(
@@ -384,6 +396,27 @@ impl FusedSeries {
}
};
let promql = Arc::<str>::from(r.text()?);
let partial_fold = match r.u8()? {
0 => None,
1 => {
let n = r.u32()? as usize;
Some(
postcard::from_bytes(r.take(n)?)
.map_err(|e| DataFusionError::Plan(e.to_string()))?,
)
}
x => {
return Err(DataFusionError::Plan(format!(
"invalid partial fold tag {x}"
)));
}
};
let schema_len = r.u32()? as usize;
let schema_bytes = r.take(schema_len)?;
let schema: datafusion::arrow::datatypes::Schema = postcard::from_bytes(schema_bytes)
.map_err(|e| {
DataFusionError::Plan(format!("cannot deserialize FusedSeries schema: {e}"))
})?;
if r.p != bytes.len() {
return Err(DataFusionError::Plan(
"trailing bytes in FusedSeries payload".into(),
@@ -399,27 +432,52 @@ impl FusedSeries {
// A frame output declares the layout of the tree's evaluated root, which the tree itself
// carries. `try_new_with_format` compares the two, so passing an empty schema here would
// make every serialized frame node impossible to rebuild.
let output_schema = if format == FusedOutputFormat::SeriesFrame {
let evaluated = evaluated_output(&tree)?;
Arc::new(datafusion::common::DFSchema::try_from(
series_frame_output_schema(evaluated.as_ref())?,
)?)
} else {
Arc::new(datafusion::common::DFSchema::empty())
let binding = FusedSeriesBinding {
grid,
metric_name,
time_index,
tags,
};
Self::try_new_with_format(
promql,
tree,
FusedSeriesBinding {
grid,
metric_name,
time_index,
tags,
},
children,
output_schema,
format,
)
let declared_schema = Arc::new(datafusion::common::DFSchema::try_from(schema)?);
if binding.tags.iter().any(|tag| tag == &binding.time_index) {
return Err(DataFusionError::Plan(
"tag column cannot be the time column".into(),
));
}
if let Some(op) = partial_fold {
if format != FusedOutputFormat::PlainRows {
return Err(DataFusionError::Plan(
"partial FusedSeries must use PlainRows".into(),
));
}
let expected = Arc::new(datafusion::common::DFSchema::try_from(state_schema(
&binding.tags,
))?);
if declared_schema.as_arrow() != expected.as_arrow() {
return Err(DataFusionError::Plan(
"partial FusedSeries output schema does not match state_schema(binding.tags)"
.into(),
));
}
return Self::try_new_partial(promql, tree, binding, children, op);
}
let output_schema = if format == FusedOutputFormat::SeriesFrame {
let expected = series_frame_output_schema(evaluated_output(&tree)?.as_ref())?;
if declared_schema.as_arrow() != expected.as_ref() {
return Err(DataFusionError::Plan(
"FusedSeries SeriesFrame schema mismatch".into(),
));
}
declared_schema
} else {
validate_plain_output_schema(
&Arc::new(declared_schema.as_arrow().clone()),
&tree,
&binding.time_index,
)?;
declared_schema
};
Self::try_new_with_format(promql, tree, binding, children, output_schema, format)
}
/// The name this node is shown under.
@@ -587,19 +645,11 @@ impl FusedSeries {
let Some(root) = self.tree.root().output.samples() else {
return Vec::new();
};
self.output_schema
.fields()
.iter()
.map(|field| field.name())
.filter(|name| {
*name != &self.binding.time_index
&& !root
.fields
.iter()
.any(|field| field.name.as_ref() == name.as_str())
})
.cloned()
.collect()
output_tags_from_schema(
&Arc::new(self.output_schema.as_arrow().clone()),
&root.fields,
&self.binding.time_index,
)
}
}
@@ -657,17 +707,12 @@ impl UserDefinedLogicalNodeCore for FusedSeries {
impl PartialEq for FusedSeries {
fn eq(&self, other: &Self) -> bool {
// `NativeTree` carries no structural equality, so the rendered tree stands in for it. The
// renderer names every node with its bound operation and grids, which is what makes two
// nodes the same computation; comparing the text alone would call two different trees
// equal whenever their PromQL happens to read the same.
self.promql == other.promql
&& self.tree.to_string() == other.tree.to_string()
&& tree_input_fields(&self.tree) == tree_input_fields(&other.tree)
&& self.tree == other.tree
&& self.binding == other.binding
&& self.children == other.children
&& self.output_schema == other.output_schema
&& partial_key(self.partial_fold.as_ref()) == partial_key(other.partial_fold.as_ref())
&& self.partial_fold == other.partial_fold
&& self.format == other.format
}
}
@@ -680,7 +725,8 @@ impl Hash for FusedSeries {
self.binding.hash(state);
self.children.hash(state);
self.output_schema.hash(state);
partial_key(self.partial_fold.as_ref()).hash(state);
self.tree.hash(state);
self.partial_fold.hash(state);
self.format.hash(state);
}
}
@@ -699,24 +745,16 @@ impl PartialOrd for FusedSeries {
Some(Ordering::Equal) => {}
ord => return ord,
}
match self.tree.to_string().partial_cmp(&other.tree.to_string()) {
Some(Ordering::Equal) => {}
ord => return ord,
if self.tree != other.tree {
return None;
}
match partial_key(self.partial_fold.as_ref())
.partial_cmp(&partial_key(other.partial_fold.as_ref()))
{
Some(Ordering::Equal) => {}
ord => return ord,
if self.partial_fold != other.partial_fold {
return None;
}
match self.format.cmp(&other.format) {
Ordering::Equal => {}
ord => return Some(ord),
}
match tree_input_fields(&self.tree).cmp(&tree_input_fields(&other.tree)) {
Ordering::Equal => {}
ord => return Some(ord),
}
if self.output_schema == other.output_schema {
Some(Ordering::Equal)
} else {
@@ -738,29 +776,6 @@ fn render_partial_fold(op: &BoundAggregate) -> String {
format!("{kind} {group} {:?}", op.stage)
}
/// A stable comparable representation for an optional partial fold.
fn partial_key(partial: Option<&BoundAggregate>) -> Option<String> {
partial.map(render_partial_fold)
}
/// The raw field names bound to each input leaf, retaining source association and leaf order.
fn tree_input_fields(tree: &NativeTree) -> Vec<(usize, Vec<String>)> {
tree_inputs(tree)
.into_iter()
.map(|input| {
let fields = match &input.output {
NativeType::Samples(schema) => schema
.fields
.iter()
.map(|field| field.name.to_string())
.collect(),
NativeType::Scalar(_) | NativeType::AggregateState(_) => Vec::new(),
};
(input.source_index, fields)
})
.collect()
}
/// The schemas read by distinct raw sources, in dense source-index order.
fn collect_source_schemas(tree: &NativeTree) -> DataFusionResult<Vec<Arc<SeriesSchema>>> {
let mut schemas: Vec<(usize, Arc<SeriesSchema>)> = Vec::new();
@@ -904,6 +919,49 @@ pub(crate) fn tree_source_fields(tree: &NativeTree) -> DataFusionResult<Vec<Vec<
.collect())
}
fn output_tags_from_schema(schema: &SchemaRef, fields: &[FieldSpec], time: &str) -> Vec<String> {
schema
.fields()
.iter()
.filter(|field| {
field.name().as_str() != time && !fields.iter().any(|f| f.name.as_ref() == field.name())
})
.map(|field| field.name().clone())
.collect()
}
fn validate_plain_output_schema(
schema: &SchemaRef,
tree: &NativeTree,
time: &str,
) -> DataFusionResult<()> {
let fields = match &tree.root().output {
NativeType::Samples(s) => s.fields.to_vec(),
NativeType::Scalar(_) => {
let non_time_columns = schema
.fields()
.iter()
.filter(|field| field.name().as_str() != time)
.count();
if non_time_columns != 1 {
return Err(DataFusionError::Plan(format!(
"FusedSeriesExec scalar root cannot produce {} tag columns",
non_time_columns.saturating_sub(1)
)));
}
scalar_root_fields(schema, &[], time)?
}
NativeType::AggregateState(_) => {
return Err(DataFusionError::Plan(
"PlainRows does not support aggregate-state roots".into(),
));
}
};
let tags = output_tags_from_schema(schema, &fields, time);
classify_output(schema, &fields, &tags, time)?;
Ok(())
}
fn validate_physical_tree(tree: &NativeTree) -> DataFusionResult<Vec<Arc<SeriesSchema>>> {
let schemas = collect_source_schemas(tree)?;
if tree.root().output.grid().is_none() {
@@ -963,6 +1021,17 @@ pub(crate) fn classify_output(
tags: &[String],
time_column: &str,
) -> DataFusionResult<Vec<OutputColumn>> {
if output_schema
.fields()
.iter()
.filter(|field| field.name().as_str() == time_column)
.count()
> 1
{
return Err(DataFusionError::Plan(format!(
"duplicate time column {time_column:?} in FusedSeriesExec output schema"
)));
}
for (index, tag) in tags.iter().enumerate() {
if tags[..index].contains(tag) {
return Err(DataFusionError::Plan(format!(
src/promql/src/extension_plan/fused_series/tests.rs
+324 -3
View File
@@ -943,12 +943,42 @@ fn fused_series_envelope_round_trip_and_rejects_unsupported_payloads() {
let bytes = node.serialize().unwrap();
let decoded = FusedSeries::deserialize(&bytes).unwrap();
assert_eq!(
crate::native::wire::encode_prefix(decoded.tree()).unwrap(),
crate::native::wire::encode_prefix(node.tree()).unwrap()
crate::native::plan::encode_tree(decoded.tree()).unwrap(),
crate::native::plan::encode_tree(node.tree()).unwrap()
);
assert_eq!(decoded.binding(), node.binding());
assert_eq!(decoded.schema(), node.schema());
assert!(fused(rate_tree()).serialize().is_err());
let rows = fused(rate_tree());
let row_decoded = FusedSeries::deserialize(&rows.serialize().unwrap()).unwrap();
assert_eq!(row_decoded.schema(), rows.schema());
assert_eq!(row_decoded.tree(), rows.tree());
// The output schema owns the label layout; the source binding may have no tags or different
// labels after a tree operation such as label rewriting.
let empty_binding = binding_of(rows.tree(), raw_schema(), Vec::new());
let empty_tags = FusedSeries::try_new(
Arc::from(PROMQL),
rate_tree(),
empty_binding,
vec![input_plan(frame_schema())],
output_schema().to_dfschema_ref().unwrap(),
)
.unwrap();
let decoded = FusedSeries::deserialize(&empty_tags.serialize().unwrap()).unwrap();
assert_eq!(decoded.schema(), empty_tags.schema());
assert_eq!(decoded.binding().tags, empty_tags.binding().tags);
assert_eq!(decoded.schema(), empty_tags.schema());
let projected = fused(binary_tree(raw_schema(), raw_schema()));
let projected_bytes = projected.serialize().unwrap();
let projected_decoded = FusedSeries::deserialize(&projected_bytes).unwrap();
assert_eq!(projected_decoded.schema(), projected.schema());
let partial = partial_node(AggregateKind::Sum);
let partial_decoded = FusedSeries::deserialize(&partial.serialize().unwrap()).unwrap();
assert_eq!(partial_decoded.partial_fold(), partial.partial_fold());
assert_eq!(partial_decoded.binding().tags, partial.binding().tags);
assert_eq!(partial_decoded.schema(), partial.schema());
let mut bad_version = bytes.clone();
bad_version[..4].copy_from_slice(&99u32.to_le_bytes());
assert!(
@@ -979,6 +1009,297 @@ fn envelope_node_with_tags(tags: Vec<String>) -> FusedSeries {
.expect("the frame layout matches the evaluated root")
}
fn mutate_envelope_schema(bytes: &[u8], mutate: impl FnOnce(&mut Vec<Field>)) -> Vec<u8> {
let mut pos = 4;
let tree_len = u32::from_le_bytes(bytes[pos..pos + 4].try_into().unwrap()) as usize;
pos += 4 + tree_len + 24;
let metric_tag = bytes[pos];
pos += 1;
if metric_tag == 1 {
let n = u32::from_le_bytes(bytes[pos..pos + 4].try_into().unwrap()) as usize;
pos += 4 + n;
}
let skip_text = |bytes: &[u8], pos: &mut usize| {
let n = u32::from_le_bytes(bytes[*pos..*pos + 4].try_into().unwrap()) as usize;
*pos += 4 + n;
};
skip_text(bytes, &mut pos); // time index
let tag_count = u32::from_le_bytes(bytes[pos..pos + 4].try_into().unwrap()) as usize;
pos += 4;
for _ in 0..tag_count {
skip_text(bytes, &mut pos);
}
pos += 1; // format
skip_text(bytes, &mut pos); // PromQL text
let partial_tag = bytes[pos];
pos += 1;
if partial_tag == 1 {
let n = u32::from_le_bytes(bytes[pos..pos + 4].try_into().unwrap()) as usize;
pos += 4 + n;
}
let schema_len_at = pos;
let schema_len =
u32::from_le_bytes(bytes[schema_len_at..schema_len_at + 4].try_into().unwrap()) as usize;
let schema: Schema =
postcard::from_bytes(&bytes[schema_len_at + 4..schema_len_at + 4 + schema_len]).unwrap();
let mut fields = schema
.fields()
.iter()
.map(|f| f.as_ref().clone())
.collect::<Vec<_>>();
mutate(&mut fields);
let changed = postcard::to_stdvec(&Schema::new(fields)).unwrap();
let mut result = bytes[..schema_len_at].to_vec();
result.extend((changed.len() as u32).to_le_bytes());
result.extend(changed);
result
}
fn plain_envelope(tree: NativeTree, tags: Vec<String>, schema: Schema) -> Vec<u8> {
let binding = binding_of(&tree, raw_schema(), tags);
FusedSeries::try_new(
Arc::from(PROMQL),
tree,
binding,
vec![input_plan(frame_schema())],
Arc::new(datafusion::common::DFSchema::try_from(schema).unwrap()),
)
.unwrap()
.serialize()
.unwrap()
}
fn assert_decode_error(bytes: &[u8], expected: &str) {
let error = FusedSeries::deserialize(bytes).unwrap_err().to_string();
assert!(
error.contains(expected),
"expected {expected:?}, got {error:?}"
);
}
#[test]
fn plain_rows_decode_rejects_invalid_schemas() {
let tree = rate_tree();
let base = output_schema();
let base_envelope = plain_envelope(
tree.clone(),
vec![HOST.into(), NAME.into()],
base.as_ref().clone(),
);
let check = |schema: Schema, expected: &str| {
assert_decode_error(
&mutate_envelope_schema(&base_envelope, |fields| {
*fields = schema.fields().iter().map(|f| f.as_ref().clone()).collect()
}),
expected,
)
};
check(
Schema::new(vec![
Field::new(VALUE, DataType::Float64, true),
Field::new(HOST, DataType::Utf8, true),
Field::new(HOST, DataType::Utf8, true),
Field::new(NAME, DataType::Utf8, true),
Field::new(
TIME,
DataType::Timestamp(TimeUnit::Millisecond, None),
false,
),
]),
"duplicate tag column",
);
check(
Schema::new(vec![
Field::new(VALUE, DataType::Float64, true),
Field::new(
TIME,
DataType::Timestamp(TimeUnit::Millisecond, None),
false,
),
Field::new(NAME, DataType::Utf8, true),
Field::new(
TIME,
DataType::Timestamp(TimeUnit::Millisecond, None),
false,
),
]),
"duplicate time column",
);
// Binding labels do not override the labels in the serialized output schema.
let collision_schema = Schema::new(vec![
Field::new(VALUE, DataType::Float64, true),
Field::new(
TIME,
DataType::Timestamp(TimeUnit::Millisecond, None),
false,
),
]);
let differing_binding = plain_envelope(tree.clone(), vec![HOST.into()], collision_schema);
assert!(FusedSeries::deserialize(&differing_binding).is_ok());
let time_tag = plain_envelope(
tree.clone(),
vec![TIME.into()],
Schema::new(vec![
Field::new(VALUE, DataType::Float64, true),
Field::new(
TIME,
DataType::Timestamp(TimeUnit::Millisecond, None),
false,
),
]),
);
assert_decode_error(&time_tag, "cannot be the time column");
for datatype in [
DataType::Int64,
DataType::Timestamp(TimeUnit::Microsecond, None),
] {
check(
Schema::new(vec![
Field::new(VALUE, DataType::Float64, true),
Field::new(HOST, DataType::Utf8, true),
Field::new(NAME, DataType::Utf8, true),
Field::new(TIME, datatype, false),
]),
"must be Timestamp(Millisecond, None)",
);
}
check(
Schema::new(vec![
Field::new(VALUE, DataType::Float64, true),
Field::new(HOST, DataType::Utf8, true),
Field::new(NAME, DataType::Utf8, true),
Field::new(TIME, DataType::Timestamp(TimeUnit::Millisecond, None), true),
]),
"must not be nullable",
);
check(
Schema::new(vec![
Field::new(VALUE, DataType::Float64, true),
Field::new(HOST, DataType::Int64, true),
Field::new(NAME, DataType::Utf8, true),
Field::new(
TIME,
DataType::Timestamp(TimeUnit::Millisecond, None),
false,
),
]),
"must be Utf8",
);
check(
Schema::new(vec![
Field::new(HOST, DataType::Utf8, true),
Field::new(NAME, DataType::Utf8, true),
Field::new(
TIME,
DataType::Timestamp(TimeUnit::Millisecond, None),
false,
),
]),
"missing field column",
);
check(
Schema::new(vec![
Field::new(VALUE, DataType::Float64, true),
Field::new(OTHER, DataType::Float64, true),
Field::new(HOST, DataType::Utf8, true),
Field::new(NAME, DataType::Utf8, true),
Field::new(
TIME,
DataType::Timestamp(TimeUnit::Millisecond, None),
false,
),
]),
"must be Utf8",
);
let scalar = scalar_tree(grid(), 1.0);
let scalar_schema = Schema::new(vec![
Field::new(VALUE, DataType::Float64, true),
Field::new(HOST, DataType::Utf8, true),
Field::new(
TIME,
DataType::Timestamp(TimeUnit::Millisecond, None),
false,
),
]);
let scalar_binding = FusedSeriesBinding {
grid: grid(),
metric_name: Some(Arc::from(METRIC)),
time_index: TIME.to_string(),
tags: vec![HOST.into()],
};
let scalar_bytes = FusedSeries::try_new(
Arc::from(PROMQL),
scalar,
scalar_binding,
vec![],
Arc::new(datafusion::common::DFSchema::try_from(scalar_schema).unwrap()),
)
.unwrap()
.serialize()
.unwrap();
assert_decode_error(&scalar_bytes, "scalar root cannot produce 1 tag columns");
}
#[test]
fn series_frame_decode_rejects_schema_mismatch() {
let node = envelope_node_with_tags(vec![HOST.into(), NAME.into()]);
let bytes = node.serialize().unwrap();
let wrong = mutate_envelope_schema(&bytes, |fields| {
fields.push(Field::new("extra", DataType::Utf8, true))
});
assert_decode_error(&wrong, "FusedSeries SeriesFrame schema mismatch");
}
#[test]
fn partial_decode_rejects_format_schema_and_tag_order_mismatch() {
let tree = rate_tree();
let node = FusedSeries::try_new_partial(
Arc::from(PROMQL),
tree.clone(),
partial_binding_of(&tree, &[HOST, JOB]),
vec![input_plan(frame_schema())],
partial_op(AggregateKind::Sum, &[HOST, JOB]),
)
.unwrap();
let bytes = node.serialize().unwrap();
let wrong_schema = mutate_envelope_schema(&bytes, |fields| {
fields.push(Field::new("extra", DataType::Utf8, true))
});
assert_decode_error(
&wrong_schema,
"partial FusedSeries output schema does not match state_schema(binding.tags)",
);
let reversed = mutate_envelope_schema(&bytes, |fields| fields.swap(1, 2));
assert_decode_error(
&reversed,
"partial FusedSeries output schema does not match state_schema(binding.tags)",
);
// The format tag follows fixed-width grid data and length-prefixed metric/time/tag strings.
let mut wrong_format = bytes.clone();
let mut pos = 8 + crate::native::plan::encode_tree(node.tree()).unwrap().len() + 8 + 8 + 8;
let text_len = |b: &[u8], at: &mut usize| {
let n = u32::from_le_bytes(b[*at..*at + 4].try_into().unwrap()) as usize;
*at += 4 + n;
};
if wrong_format[pos] == 1 {
pos += 1;
text_len(&wrong_format, &mut pos); // metric name
} else {
pos += 1;
}
text_len(&wrong_format, &mut pos); // time index
let count = u32::from_le_bytes(wrong_format[pos..pos + 4].try_into().unwrap()) as usize;
pos += 4;
for _ in 0..count {
text_len(&wrong_format, &mut pos);
}
wrong_format[pos] = 1;
assert_decode_error(&wrong_format, "partial FusedSeries must use PlainRows");
}
/// The largest tag count the envelope supports is written whole.
#[test]
fn fused_series_envelope_serializes_the_maximum_tag_count() {
src/promql/src/histogram_eval.rs
+30 -2
View File
@@ -50,7 +50,7 @@ use crate::map::validate_grid_series;
const BUCKET_LABEL: &str = "le";
/// `histogram_quantile` or `histogram_fraction` with its constant arguments.
#[derive(Debug, Clone, Copy, PartialEq)]
#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)]
pub enum HistogramFunction {
/// `histogram_quantile(q, v)`.
Quantile(f64),
@@ -273,7 +273,7 @@ pub(crate) fn histogram_string(h: &HistogramView<'_>) -> Result<String> {
}
/// A function of one native histogram.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, Hash)]
pub enum NativeHistogramFunction {
Count,
Sum,
@@ -346,3 +346,31 @@ impl NativeHistogramFunction {
})
}
}
impl PartialEq for HistogramFunction {
fn eq(&self, o: &Self) -> bool {
match (self, o) {
(Self::Quantile(a), Self::Quantile(b)) => a.to_bits() == b.to_bits(),
(Self::Fraction { lower: a, upper: b }, Self::Fraction { lower: c, upper: d }) => {
a.to_bits() == c.to_bits() && b.to_bits() == d.to_bits()
}
_ => false,
}
}
}
impl Eq for HistogramFunction {}
impl std::hash::Hash for HistogramFunction {
fn hash<H: std::hash::Hasher>(&self, h: &mut H) {
match self {
Self::Quantile(x) => {
0u8.hash(h);
x.to_bits().hash(h)
}
Self::Fraction { lower, upper } => {
1u8.hash(h);
lower.to_bits().hash(h);
upper.to_bits().hash(h)
}
}
}
}
src/promql/src/info.rs
+1 -1
View File
@@ -43,7 +43,7 @@ use crate::vector_match::{Budget, HISTOGRAM_ROW_BYTES, owned_histogram_bytes};
const IDENTIFYING_LABELS: [&str; 2] = ["instance", "job"];
/// The second argument of `info`, split the way Prometheus uses it.
#[derive(Clone, Debug)]
#[derive(Clone, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
pub struct InfoSelector {
/// The `__name__` matchers, which choose the info metrics; none means `target_info`.
pub name_matchers: Vec<(MatchOp, Arc<str>)>,
src/promql/src/map.rs
+33 -1
View File
@@ -46,7 +46,7 @@ use crate::grid_series::GridSeriesBuilder;
use crate::map_time::DateFunction;
/// Float math supported by the evaluated-series kernel.
#[derive(Clone, Copy, Debug)]
#[derive(Clone, Copy, Debug, serde::Serialize, serde::Deserialize)]
pub enum MathOp {
/// Absolute value.
Abs,
@@ -452,3 +452,35 @@ fn float_sample(sample: SampleRef<'_>) -> Option<f64> {
_ => unreachable!("validated as float"),
}
}
impl PartialEq for MathOp {
fn eq(&self, o: &Self) -> bool {
match (self, o) {
(Self::Round { nearest: a }, Self::Round { nearest: b }) => a.to_bits() == b.to_bits(),
(Self::Clamp { min: a, max: b }, Self::Clamp { min: c, max: d }) => {
a.to_bits() == c.to_bits() && b.to_bits() == d.to_bits()
}
(Self::ClampMin { min: a }, Self::ClampMin { min: b })
| (Self::ClampMax { max: a }, Self::ClampMax { max: b }) => a.to_bits() == b.to_bits(),
(Self::Date(a), Self::Date(b)) => a == b,
_ => std::mem::discriminant(self) == std::mem::discriminant(o),
}
}
}
impl Eq for MathOp {}
impl std::hash::Hash for MathOp {
fn hash<H: std::hash::Hasher>(&self, h: &mut H) {
std::mem::discriminant(self).hash(h);
match self {
Self::Round { nearest: x } | Self::ClampMin { min: x } | Self::ClampMax { max: x } => {
x.to_bits().hash(h)
}
Self::Clamp { min, max } => {
min.to_bits().hash(h);
max.to_bits().hash(h)
}
Self::Date(d) => d.hash(h),
_ => (),
}
}
}
src/promql/src/map_label.rs
+149
View File
@@ -80,6 +80,89 @@ fn rewrite_all(
/// label names are checked.
#[derive(Debug)]
pub struct LabelRewrite(Change);
impl serde::Serialize for LabelRewrite {
fn serialize<S: serde::Serializer>(
&self,
serializer: S,
) -> std::result::Result<S::Ok, S::Error> {
#[derive(serde::Serialize)]
enum Wire<'a> {
Replace {
dst: &'a str,
replacement: &'a str,
src: &'a str,
pattern: &'a str,
},
Join {
dst: &'a str,
separator: &'a str,
src: Vec<&'a str>,
},
}
let wire = match &self.0 {
Change::Replace {
dst,
replacement,
src,
pattern,
..
} => Wire::Replace {
dst,
replacement,
src,
pattern,
},
Change::Join {
dst,
separator,
src,
} => Wire::Join {
dst,
separator,
src: src.iter().map(AsRef::as_ref).collect(),
},
};
serde::Serialize::serialize(&wire, serializer)
}
}
impl<'de> serde::Deserialize<'de> for LabelRewrite {
fn deserialize<D: serde::Deserializer<'de>>(
deserializer: D,
) -> std::result::Result<Self, D::Error> {
#[derive(serde::Deserialize)]
enum Wire {
Replace {
dst: String,
replacement: String,
src: String,
pattern: String,
},
Join {
dst: String,
separator: String,
src: Vec<String>,
},
}
match <Wire as serde::Deserialize>::deserialize(deserializer)? {
Wire::Replace {
dst,
replacement,
src,
pattern,
} => {
Self::replace(&dst, &replacement, &src, &pattern).map_err(serde::de::Error::custom)
}
Wire::Join {
dst,
separator,
src,
} => {
let refs: Vec<_> = src.iter().map(String::as_str).collect();
Self::join(&dst, &separator, &refs).map_err(serde::de::Error::custom)
}
}
}
}
#[derive(Debug)]
enum Change {
@@ -98,6 +181,72 @@ enum Change {
},
}
impl PartialEq for LabelRewrite {
fn eq(&self, other: &Self) -> bool {
match (&self.0, &other.0) {
(
Change::Replace {
dst: a,
replacement: b,
src: c,
pattern: d,
..
},
Change::Replace {
dst: w,
replacement: x,
src: y,
pattern: z,
..
},
) => (a, b, c, d) == (w, x, y, z),
(
Change::Join {
dst: a,
separator: b,
src: c,
},
Change::Join {
dst: w,
separator: x,
src: y,
},
) => (a, b, c) == (w, x, y),
_ => false,
}
}
}
impl Eq for LabelRewrite {}
impl std::hash::Hash for LabelRewrite {
fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
match &self.0 {
Change::Replace {
dst,
replacement,
src,
pattern,
..
} => {
0u8.hash(state);
dst.hash(state);
replacement.hash(state);
src.hash(state);
pattern.hash(state);
}
Change::Join {
dst,
separator,
src,
} => {
1u8.hash(state);
dst.hash(state);
separator.hash(state);
src.hash(state);
}
}
}
}
impl LabelRewrite {
/// `label_replace`'s change, failing on an invalid expression or an empty destination.
pub fn replace(dst: &str, replacement: &str, src: &str, regex: &str) -> Result<Self> {
src/promql/src/map_time.rs
+1 -1
View File
@@ -27,7 +27,7 @@ use crate::grid_series::external;
mod tests;
/// A PromQL date function: `minute(v)`, `year()` and so on.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, Hash)]
pub enum DateFunction {
DayOfMonth,
DayOfWeek,
src/promql/src/native/mod.rs
-1
View File
@@ -34,4 +34,3 @@ pub mod execute;
pub mod metrics;
pub mod plan;
pub(crate) mod runs;
pub(crate) mod wire;
src/promql/src/native/plan.rs
+243 -25
View File
@@ -26,8 +26,36 @@
//! What the tree cannot express yet is deliberate for this stage: `scalar(v)` has no node. The
//! legality below is therefore about the operations that exist, not about PromQL's whole surface.
use std::cell::Cell;
use std::hash::Hash;
use std::sync::Arc;
const MAX_NATIVE_NODE_DEPTH: usize = 128;
thread_local! {
static NATIVE_NODE_DEPTH: Cell<usize> = const { Cell::new(0) };
}
struct NativeNodeDepthGuard;
impl NativeNodeDepthGuard {
fn enter<E: serde::de::Error>() -> std::result::Result<Self, E> {
NATIVE_NODE_DEPTH.with(|depth| {
let current = depth.get();
if current >= MAX_NATIVE_NODE_DEPTH {
return Err(E::custom(format!(
"native tree nesting exceeds {MAX_NATIVE_NODE_DEPTH}"
)));
}
depth.set(current + 1);
Ok(Self)
})
}
}
impl Drop for NativeNodeDepthGuard {
fn drop(&mut self) {
NATIVE_NODE_DEPTH.with(|depth| depth.set(depth.get() - 1));
}
}
use datafusion::error::{DataFusionError, Result};
use datatypes::series::samples::{FieldSpec, SeriesSchema};
use datatypes::series::time::EvalGrid;
@@ -48,7 +76,7 @@ use crate::vector_match::{BinaryMatching, MatchingLabels, SetOp};
/// a grid it is raw samples carrying their own timestamps. A scalar is one value per grid point
/// and no labels. Intermediate aggregate state is a different type altogether, so an operation
/// that cannot consume it is rejected by type rather than by convention.
#[derive(Clone, Debug)]
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, PartialEq, Hash, Eq)]
pub enum NativeType {
/// Samples, evaluated when the schema carries a grid and raw otherwise.
Samples(Arc<SeriesSchema>),
@@ -95,14 +123,12 @@ impl NativeType {
/// Whether `self` binds the same thing as `other`: the same samples or the same state.
///
/// Bindings are compared by allocation identity, because a compiler that meant one schema or
/// one state has no reason to allocate it twice. A scalar is produced by a node and never
/// handed to one by a source, so it never matches. This is about the binding itself: grids are
/// aligned by [`EvalGrid::spec`] everywhere else, here and in the kernels.
/// Bindings compare their semantic content, not process-local allocation identity. Scalars are
/// produced by a node and never handed to one by a source, so they never match.
fn same_binding(&self, other: &NativeType) -> bool {
match (self, other) {
(Self::Samples(mine), Self::Samples(theirs)) => Arc::ptr_eq(mine, theirs),
(Self::AggregateState(mine), Self::AggregateState(theirs)) => Arc::ptr_eq(mine, theirs),
(Self::Samples(mine), Self::Samples(theirs)) => **mine == **theirs,
(Self::AggregateState(mine), Self::AggregateState(theirs)) => **mine == **theirs,
_ => false,
}
}
@@ -112,7 +138,7 @@ impl NativeType {
///
/// State is indexed by `(group labels, absolute grid index)`, so it carries the grid it was
/// accumulated on. It is not a query value and does not pretend to be one.
#[derive(Clone, Debug)]
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, PartialEq, Hash, Eq)]
pub struct AggregateStateSchema {
/// How the groups its rows are indexed by are identified.
pub group: Grouping,
@@ -127,7 +153,7 @@ pub struct AggregateStateSchema {
/// The binding records what the input was created for: the output a consumer type-checks against,
/// and the ordering it guarantees. An instant selection and a window both read the ordering,
/// rejecting an interleaved input; no other consumer reads it yet.
#[derive(Clone, Debug)]
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, PartialEq, Hash, Eq)]
pub struct BoundInput {
/// Child index in the DataFusion plan. The compiler initially assigns selector-occurrence
/// indices; the caller maps them to child indices when constructing the plan.
@@ -146,14 +172,14 @@ pub struct BoundInput {
/// A scalar that varies per grid point is the same [`NativeType::Scalar`] as this constant: what
/// keeps a binary operation from label-matching it is its type, not its constness. The varying
/// ones are [`NativeExpr::EvalTime`] and a binary operation between scalars.
#[derive(Clone, Debug)]
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
pub struct BoundScalar {
/// The constant value.
pub value: f64,
}
/// An instant selection: the last sample at or before each grid point, within a lookback.
#[derive(Clone, Debug)]
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, PartialEq, Hash, Eq)]
pub struct BoundInstant {
/// How far back a sample may lie before it counts as stale.
pub lookback_ms: i64,
@@ -168,7 +194,7 @@ pub struct BoundInstant {
}
/// What an instant selection reports at a grid point for the sample it selected there.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize, Hash)]
pub enum InstantValue {
/// The sample itself.
Sample,
@@ -183,7 +209,7 @@ pub enum InstantValue {
///
/// The input and output grids are bound separately: a subquery evaluates its inner expression on
/// its own grid and consumes the result directly, without resampling it onto the outer grid.
#[derive(Clone, Debug)]
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, PartialEq, Hash, Eq)]
pub struct BoundWindow {
/// The function this node runs.
pub function: RangeFunction,
@@ -196,7 +222,7 @@ pub struct BoundWindow {
}
/// A chain of point-wise operations applied to a whole evaluated value.
#[derive(Clone, Debug)]
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, PartialEq, Hash, Eq)]
pub struct BoundMap {
/// The operations, applied in order.
pub ops: Arc<[MapOp]>,
@@ -205,7 +231,7 @@ pub struct BoundMap {
/// A math function whose scalar parameters are bound as nodes, because at least one of them varies
/// per grid point; Prometheus reads each parameter at every step. With constant parameters the same
/// function is a [`MathOp`] in a [`BoundMap`].
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize, Hash)]
pub enum ParameterizedOp {
/// `round(v, to_nearest)`.
Round,
@@ -230,7 +256,7 @@ impl ParameterizedOp {
/// One operation of a map, applied to each series on its own: math over its float values, a
/// function of its native histograms, or a change of its labels. Each can give two series one label
/// set, which is why all live in a map.
#[derive(Clone, Debug)]
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, PartialEq, Hash, Eq)]
pub enum MapOp {
Math(MathOp),
Histogram(NativeHistogramFunction),
@@ -251,7 +277,7 @@ pub enum MapOp {
/// [`RangeFunction::QuantileOverTime`] does.
///
/// [`RangeFunction::QuantileOverTime`]: crate::range_eval::RangeFunction::QuantileOverTime
#[derive(Clone, Copy, Debug, PartialEq)]
#[derive(Clone, Copy, Debug, serde::Serialize, serde::Deserialize)]
pub enum AggregateKind {
/// `sum`: the total of each group's values at each grid point.
Sum,
@@ -278,7 +304,7 @@ pub enum AggregateKind {
}
/// Where an aggregate sits in the execution, which decides whether its output is a query value.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize, Hash)]
pub enum AggregateStage {
/// Every contribution to each group lies within this execution, so the output is complete.
Complete,
@@ -296,7 +322,7 @@ pub enum AggregateStage {
/// `without` written that way would group a query by labels it meant to remove.
///
/// [`aggregate::Grouping`]: crate::aggregate::Grouping
#[derive(Clone, Debug, PartialEq, Eq)]
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize, Hash)]
pub enum Grouping {
/// The labels a group is identified by; every other label is dropped.
By(Arc<[Arc<str>]>),
@@ -305,7 +331,7 @@ pub enum Grouping {
}
/// An aggregate bound to its kind, grouping and stage.
#[derive(Clone, Debug)]
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, PartialEq, Hash, Eq)]
pub struct BoundAggregate {
/// Which aggregate to compute.
pub kind: AggregateKind,
@@ -321,7 +347,7 @@ pub struct BoundAggregate {
///
/// Its constants are bound here, as a window's are: a `k` or a ratio only known at run time has no
/// field.
#[derive(Clone, Debug, PartialEq)]
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
pub enum BoundVectorOp {
/// `topk` (`top`) or `bottomk`: the `k` largest or smallest values of each group.
TopK { k: f64, top: bool, group: Grouping },
@@ -349,7 +375,7 @@ pub enum BoundVectorOp {
///
/// It has no cardinality, `bool` or fill: PromQL matches a set operator many-to-many and refuses
/// the other modifiers on it.
#[derive(Clone, Debug, PartialEq, Eq)]
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize, Hash)]
pub struct BoundSetOp {
/// `and`, `or` or `unless`.
pub op: SetOp,
@@ -358,7 +384,7 @@ pub struct BoundSetOp {
}
/// What a node computes. Every variant holds a bound operation, never a name to resolve.
#[derive(Clone, Debug)]
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, PartialEq, Eq, Hash)]
pub enum NativeExpr {
/// A scan or an exchange boundary.
Input(BoundInput),
@@ -434,7 +460,7 @@ pub enum NativeExpr {
}
/// One node of a bound tree.
#[derive(Clone, Debug)]
#[derive(Clone, Debug, serde::Serialize, PartialEq, Eq, Hash)]
pub struct NativeNode {
/// What this node computes, including its bound operands.
pub expr: NativeExpr,
@@ -504,7 +530,7 @@ fn merge_shape_supported(node: &NativeNode) -> bool {
///
/// The tree itself carries no run-time state: every execution builds its own, so one compiled plan
/// can serve every partition.
#[derive(Clone, Debug)]
#[derive(Clone, Debug, serde::Serialize, PartialEq, Eq, Hash)]
pub struct NativeTree {
root: NativeNode,
}
@@ -526,6 +552,37 @@ impl NativeTree {
}
}
impl<'de> serde::Deserialize<'de> for NativeNode {
fn deserialize<D: serde::Deserializer<'de>>(
deserializer: D,
) -> std::result::Result<Self, D::Error> {
let _guard = NativeNodeDepthGuard::enter::<D::Error>()?;
#[derive(serde::Deserialize)]
struct Wire {
expr: NativeExpr,
output: NativeType,
}
let wire = Wire::deserialize(deserializer)?;
Ok(Self {
expr: wire.expr,
output: wire.output,
})
}
}
impl<'de> serde::Deserialize<'de> for NativeTree {
fn deserialize<D: serde::Deserializer<'de>>(
deserializer: D,
) -> std::result::Result<Self, D::Error> {
#[derive(serde::Deserialize)]
struct Wire {
root: NativeNode,
}
let wire = Wire::deserialize(deserializer)?;
Self::try_new(wire.root).map_err(serde::de::Error::custom)
}
}
impl NativeNode {
/// Checks this node and its inputs against the contract of its operation.
pub fn validate(&self) -> Result<()> {
@@ -1025,3 +1082,164 @@ fn invalid(reason: &str) -> DataFusionError {
#[cfg(test)]
mod tests;
/// Encode a validated native tree in the versioned Postcard envelope.
pub(crate) fn encode_tree(tree: &NativeTree) -> Result<Vec<u8>, String> {
const MAGIC: &[u8; 4] = b"NPRQ";
const VERSION: u32 = 2;
const MAX_BYTES: usize = 1_048_576;
let payload = postcard::to_stdvec(tree).map_err(|e| e.to_string())?;
if payload.len() > MAX_BYTES {
return Err("oversized native tree payload".into());
}
let mut bytes = Vec::with_capacity(12 + payload.len());
bytes.extend_from_slice(MAGIC);
bytes.extend_from_slice(&VERSION.to_le_bytes());
bytes.extend_from_slice(&(payload.len() as u32).to_le_bytes());
bytes.extend_from_slice(&payload);
Ok(bytes)
}
/// Decode and validate a versioned native tree envelope.
pub(crate) fn decode_tree(bytes: &[u8]) -> Result<NativeTree, String> {
const MAGIC: &[u8; 4] = b"NPRQ";
const VERSION: u32 = 2;
const MAX_BYTES: usize = 1_048_576;
if bytes.len() < 12 {
return Err("truncated native tree envelope".into());
}
if &bytes[..4] != MAGIC {
return Err("invalid native tree magic".into());
}
let version = u32::from_le_bytes(bytes[4..8].try_into().unwrap());
if version != VERSION {
return Err(format!("unknown native tree version {version}"));
}
let len = u32::from_le_bytes(bytes[8..12].try_into().unwrap()) as usize;
if len > MAX_BYTES {
return Err("oversized native tree payload".into());
}
if bytes.len() < 12 + len {
return Err("truncated native tree payload".into());
}
if bytes.len() != 12 + len {
return Err("trailing bytes after native tree payload".into());
}
let (tree, remainder) = postcard::take_from_bytes::<NativeTree>(&bytes[12..])
.map_err(|e| format!("invalid native tree payload: {e}"))?;
if !remainder.is_empty() {
return Err("trailing bytes after native tree payload".into());
}
Ok(tree)
}
impl PartialEq for BoundScalar {
fn eq(&self, o: &Self) -> bool {
self.value.to_bits() == o.value.to_bits()
}
}
impl Eq for BoundScalar {}
impl std::hash::Hash for BoundScalar {
fn hash<H: std::hash::Hasher>(&self, h: &mut H) {
self.value.to_bits().hash(h);
}
}
impl PartialEq for AggregateKind {
fn eq(&self, o: &Self) -> bool {
match (self, o) {
(Self::Quantile { quantile: a }, Self::Quantile { quantile: b }) => {
a.to_bits() == b.to_bits()
}
(Self::Sum, Self::Sum)
| (Self::Avg, Self::Avg)
| (Self::Count, Self::Count)
| (Self::Min, Self::Min)
| (Self::Max, Self::Max)
| (Self::Group, Self::Group)
| (Self::Stddev, Self::Stddev)
| (Self::Stdvar, Self::Stdvar) => true,
_ => false,
}
}
}
impl Eq for AggregateKind {}
impl std::hash::Hash for AggregateKind {
fn hash<H: std::hash::Hasher>(&self, h: &mut H) {
std::mem::discriminant(self).hash(h);
if let Self::Quantile { quantile } = self {
quantile.to_bits().hash(h);
}
}
}
impl PartialEq for BoundVectorOp {
fn eq(&self, o: &Self) -> bool {
match (self, o) {
(
Self::TopK {
k: a,
top: b,
group: c,
},
Self::TopK {
k: x,
top: y,
group: z,
},
) => a.to_bits() == x.to_bits() && b == y && c == z,
(Self::LimitK { k: a, group: b }, Self::LimitK { k: x, group: y }) => {
a.to_bits() == x.to_bits() && b == y
}
(Self::LimitRatio { ratio: a }, Self::LimitRatio { ratio: b }) => {
a.to_bits() == b.to_bits()
}
(Self::Sort(a), Self::Sort(b)) => a == b,
(Self::Absent { labels: a }, Self::Absent { labels: b }) => a == b,
(Self::Histogram(a), Self::Histogram(b)) => a == b,
(
Self::CountValues { label: a, group: b },
Self::CountValues { label: x, group: y },
) => a == x && b == y,
_ => false,
}
}
}
impl std::hash::Hash for BoundVectorOp {
fn hash<H: std::hash::Hasher>(&self, h: &mut H) {
match self {
Self::TopK { k, top, group } => {
0u8.hash(h);
k.to_bits().hash(h);
top.hash(h);
group.hash(h)
}
Self::LimitK { k, group } => {
1u8.hash(h);
k.to_bits().hash(h);
group.hash(h)
}
Self::LimitRatio { ratio } => {
2u8.hash(h);
ratio.to_bits().hash(h)
}
Self::Sort(x) => {
3u8.hash(h);
x.hash(h)
}
Self::Absent { labels } => {
4u8.hash(h);
labels.hash(h)
}
Self::Histogram(x) => {
5u8.hash(h);
x.hash(h)
}
Self::CountValues { label, group } => {
6u8.hash(h);
label.hash(h);
group.hash(h)
}
}
}
}
impl Eq for BoundVectorOp {}
src/promql/src/native/plan/tests.rs
+250 -8
View File
@@ -28,7 +28,9 @@ use crate::extension_plan::{FusedSeries, FusedSeriesBinding};
use crate::info::InfoSelector;
use crate::map::MathOp;
use crate::range_eval::{RangeFunction, RangeWindow};
use crate::vector_match::{BinaryMatching, BinaryOp, Cardinality, FillValues, MatchingLabels};
use crate::vector_match::{
BinaryMatching, BinaryOp, Cardinality, FillValues, MatchingLabels, SetOp,
};
fn grid(start: i64, step: i64, len: u64) -> Arc<EvalGrid> {
Arc::new(EvalGrid::try_new(GridSpec::try_new(start, step, len).unwrap()).unwrap())
@@ -1345,7 +1347,10 @@ fn a_scalar_node_must_produce_a_scalar() {
fn an_input_must_report_the_binding_it_was_bound_to() {
let node = NativeNode {
expr: NativeExpr::Input(BoundInput {
output: raw(),
output: NativeType::Samples(Arc::new(SeriesSchema {
fields: fields(1),
eval_grid: None,
})),
source_index: 0,
ordering: SeriesOrdering::KeyThenTime,
}),
@@ -1372,17 +1377,105 @@ fn an_input_can_bind_aggregate_state() {
// A `Final` aggregate can then consume it, which is the shape of a distributed aggregate.
assert!(final_aggregate(bound).validate().is_ok());
// A fresh state allocation is a different binding, so it cannot be reported as the bound one.
let other = NativeNode {
// Equivalent independent allocations are accepted; field, order, grid, and group changes remain invalid.
let evaluated_a = evaluated(fields(2), grid(0, 60_000, 4));
let equivalent = NativeNode {
expr: NativeExpr::Input(BoundInput {
output: state.output.clone(),
output: evaluated_a.clone(),
source_index: 0,
ordering: SeriesOrdering::Contiguous,
}),
output: partial_aggregate(window_over(input(), None)).output,
output: evaluated(fields(2), grid(0, 60_000, 4)),
};
let error = other.validate().unwrap_err();
assert!(error.to_string().contains("not the binding"), "{error}");
assert!(equivalent.validate().is_ok());
let mismatches = [
evaluated(fields(1), grid(0, 60_000, 4)),
evaluated(
Arc::from([
FieldSpec {
name: Arc::from("f1"),
},
FieldSpec {
name: Arc::from("f0"),
},
]),
grid(0, 60_000, 4),
),
evaluated(fields(2), grid(60_000, 60_000, 4)),
];
for output in mismatches {
let mismatch = NativeNode {
expr: NativeExpr::Input(BoundInput {
output: evaluated_a.clone(),
source_index: 0,
ordering: SeriesOrdering::Contiguous,
}),
output,
};
assert!(mismatch.validate().is_err());
}
let state_a = partial_aggregate(window_over(input(), None)).output;
let NativeType::AggregateState(schema_a) = &state_a else {
unreachable!()
};
let state_group_mismatch = NativeType::AggregateState(Arc::new(super::AggregateStateSchema {
group: Grouping::Without(Arc::new([Arc::from("job")])),
accumulators: schema_a.accumulators.clone(),
grid: schema_a.grid.clone(),
}));
let mismatch = NativeNode {
expr: NativeExpr::Input(BoundInput {
output: state_a,
source_index: 0,
ordering: SeriesOrdering::Contiguous,
}),
output: state_group_mismatch,
};
assert!(mismatch.validate().is_err());
let state_grid = grid(0, 60_000, 4);
let state = NativeType::AggregateState(Arc::new(super::AggregateStateSchema {
group: Grouping::By(Arc::new([Arc::from("job")])),
accumulators: fields(2),
grid: Arc::clone(&state_grid),
}));
let equivalent_state = NativeType::AggregateState(Arc::new(super::AggregateStateSchema {
group: Grouping::By(Arc::new([Arc::from("job")])),
accumulators: fields(2),
grid: Arc::clone(&state_grid),
}));
let state_input = |bound: NativeType, output: NativeType| NativeNode {
expr: NativeExpr::Input(BoundInput {
output: bound,
source_index: 0,
ordering: SeriesOrdering::Contiguous,
}),
output,
};
assert!(
state_input(state.clone(), equivalent_state)
.validate()
.is_ok()
);
let reversed_accumulators = NativeType::AggregateState(Arc::new(super::AggregateStateSchema {
group: Grouping::By(Arc::new([Arc::from("job")])),
accumulators: Arc::from([
FieldSpec {
name: Arc::from("f1"),
},
FieldSpec {
name: Arc::from("f0"),
},
]),
grid: state_grid,
}));
assert!(
state_input(state.clone(), reversed_accumulators)
.validate()
.is_err()
);
}
/// A vector operation reads the values at each grid point, so raw samples, which have no shared points,
@@ -1476,3 +1569,152 @@ fn info_reads_the_sample_times_of_one_field() {
assert!(error.to_string().contains(expected), "{error}");
}
}
#[test]
fn native_tree_deserialization_enforces_node_depth_limit() {
use super::MAX_NATIVE_NODE_DEPTH;
assert!(std::env::var_os("RUST_MIN_STACK").is_none());
fn nested_maps(map_count: usize) -> NativeTree {
let mut node = input_on(grid(0, 60_000, 1));
let output = node.output.clone();
for _ in 0..map_count {
node = NativeNode {
expr: NativeExpr::Map {
input: Box::new(node),
op: BoundMap {
ops: Arc::new([MapOp::Math(MathOp::Abs)]),
},
},
output: output.clone(),
};
}
NativeTree { root: node }
}
// A leaf counts as a node too: 127 map wrappers plus the input gives exactly 128.
let at_limit = postcard::to_stdvec(&nested_maps(MAX_NATIVE_NODE_DEPTH - 1)).unwrap();
let decoded: NativeTree = postcard::from_bytes(&at_limit).unwrap();
assert!(decoded.root().validate().is_ok());
// Serialize the hostile value directly; it deliberately bypasses try_new so the 129-node
// shape reaches the shipping Postcard decoder without recursively validating on the way in.
// Postcard reduces custom serde errors to its unit error variant, so assert rejection here;
// the guard's message text is not available through this codec.
let over_limit = postcard::to_stdvec(&nested_maps(MAX_NATIVE_NODE_DEPTH)).unwrap();
let error = postcard::from_bytes::<NativeTree>(&over_limit).unwrap_err();
assert_eq!(error, postcard::Error::SerdeDeCustom);
}
#[test]
fn native_tree_postcard_float_fidelity_round_trip() {
for bits in [
0x7ff8_0000_0000_0001,
0,
0x8000_0000_0000_0000,
f64::INFINITY.to_bits(),
f64::NEG_INFINITY.to_bits(),
1,
] {
let value = f64::from_bits(bits);
let tree = NativeTree::try_new(scalar_on(value, grid(0, 1, 1))).unwrap();
let decoded = NativeTree::try_new(
super::decode_tree(&super::encode_tree(&tree).unwrap())
.unwrap()
.root()
.clone(),
)
.unwrap();
let NativeExpr::Scalar(value) = &decoded.root().expr else {
panic!("scalar shape changed")
};
assert_eq!(value.value.to_bits(), bits);
}
}
#[test]
fn native_tree_envelope_rejects_version_size_trailing_and_truncation() {
let tree = NativeTree::try_new(input()).unwrap();
let bytes = super::encode_tree(&tree).unwrap();
let mut bad_version = bytes.clone();
bad_version[4..8].copy_from_slice(&99u32.to_le_bytes());
bad_version[12..].fill(0xff);
assert!(
super::decode_tree(&bad_version)
.unwrap_err()
.contains("unknown native tree version")
);
let mut oversized = bytes.clone();
oversized[8..12].copy_from_slice(&(1_048_577u32).to_le_bytes());
assert!(
super::decode_tree(&oversized)
.unwrap_err()
.contains("oversized")
);
let mut trailing = bytes.clone();
trailing.push(0);
assert!(
super::decode_tree(&trailing)
.unwrap_err()
.contains("trailing")
);
assert!(
super::decode_tree(&bytes[..5])
.unwrap_err()
.contains("truncated")
);
}
#[test]
fn native_tree_non_prefix_shapes_round_trip() {
let grid = grid(0, 60_000, 4);
let lhs = window_on(input(), None, Arc::clone(&grid));
let rhs = window_on(input_with_fields(0), None, Arc::clone(&grid));
let binary = NativeNode {
expr: NativeExpr::Binary {
left: Box::new(lhs.clone()),
right: Box::new(rhs.clone()),
matching: add(),
},
output: evaluated(Arc::new([]), Arc::clone(&grid)),
};
let aggregate = partial_aggregate(window_on(input(), None, Arc::clone(&grid)));
let set = NativeNode {
expr: NativeExpr::Set {
left: Box::new(lhs),
right: Box::new(rhs),
op: super::BoundSetOp {
op: SetOp::Or,
labels: MatchingLabels::Ignoring(Vec::new()),
},
},
output: evaluated(Arc::new([]), Arc::clone(&grid)),
};
let trees = vec![
NativeTree::try_new(input()).unwrap(),
NativeTree::try_new(input_on(Arc::clone(&grid))).unwrap(),
NativeTree::try_new(scalar_on(3.0, Arc::clone(&grid))).unwrap(),
NativeTree::try_new(binary).unwrap(),
NativeTree::try_new(aggregate).unwrap(),
NativeTree::try_new(set).unwrap(),
];
for tree in trees {
let decoded = super::decode_tree(&super::encode_tree(&tree).unwrap()).unwrap();
assert_eq!(decoded, tree);
}
}
#[test]
fn independently_allocated_binding_mismatch_is_rejected() {
let node = input_with_fields(1);
let mut mismatch = node.clone();
let NativeExpr::Input(binding) = &mut mismatch.expr else {
unreachable!()
};
binding.output = NativeType::Samples(Arc::new(SeriesSchema {
fields: fields(2),
eval_grid: None,
}));
assert!(mismatch.validate().is_err());
}
src/promql/src/native/wire.rs
-1347
View File
File diff suppressed because it is too large Load Diff
src/promql/src/range_eval.rs
+2 -2
View File
@@ -45,7 +45,7 @@ pub(crate) use function::quantile;
/// `(E - range_ms, E]`, where `E` is `T - offset_ms`, or `anchor_ms - offset_ms` with `@`. A
/// modifier widens that: `anchored` also reads the lookback before `E - range_ms`, `smoothed` the
/// lookback on both sides, and a scan feeding the window has to read that much more.
#[derive(Clone, Copy, Debug)]
#[derive(Clone, Copy, Debug, serde::Serialize, serde::Deserialize, PartialEq, Hash, Eq)]
pub struct RangeWindow {
/// Positive window length in milliseconds.
pub range_ms: i64,
@@ -62,7 +62,7 @@ pub struct RangeWindow {
/// The experimental range selector modifiers of Prometheus, which read samples beyond the
/// window to pin its boundary values. Not to be confused with an `@` anchor, which fixes the
/// window's end.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize, Hash)]
pub enum RangeModifier {
/// `anchored`: the window also reads the last sample at or before its start, up to the
/// statement's lookback before it, and takes it as the value at the start.
src/promql/src/range_eval/function.rs
+35 -1
View File
@@ -23,7 +23,7 @@ use crate::go_sort;
use crate::range_eval::{BatchKernel, FloatWindow, FloatWindows, RangeModifier, WindowKernel};
/// A PromQL range-vector function over float and native histogram samples.
#[derive(Clone, Copy, Debug, PartialEq)]
#[derive(Clone, Copy, Debug, serde::Serialize, serde::Deserialize)]
pub enum RangeFunction {
Rate,
Increase,
@@ -611,3 +611,37 @@ fn median_absolute_deviation(
deviations.extend(values.iter().map(|value| (value - median).abs()));
quantile(0.5, deviations, sorted)
}
impl PartialEq for RangeFunction {
fn eq(&self, o: &Self) -> bool {
match (self, o) {
(Self::QuantileOverTime(a), Self::QuantileOverTime(b))
| (Self::PredictLinear(a), Self::PredictLinear(b)) => a.to_bits() == b.to_bits(),
(
Self::DoubleExponentialSmoothing {
smoothing: a,
trend: b,
},
Self::DoubleExponentialSmoothing {
smoothing: c,
trend: d,
},
) => a.to_bits() == c.to_bits() && b.to_bits() == d.to_bits(),
_ => std::mem::discriminant(self) == std::mem::discriminant(o),
}
}
}
impl Eq for RangeFunction {}
impl std::hash::Hash for RangeFunction {
fn hash<H: std::hash::Hasher>(&self, h: &mut H) {
std::mem::discriminant(self).hash(h);
match self {
Self::QuantileOverTime(x) | Self::PredictLinear(x) => x.to_bits().hash(h),
Self::DoubleExponentialSmoothing { smoothing, trend } => {
smoothing.to_bits().hash(h);
trend.to_bits().hash(h)
}
_ => (),
}
}
}
src/promql/src/select.rs
+2 -2
View File
@@ -44,7 +44,7 @@ pub enum Grouping {
}
/// The order `sort_instant` arranges an instant result in.
#[derive(Clone, Debug, PartialEq, Eq)]
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize, Hash)]
pub enum SortOrder {
/// `sort`: ascending values, NaN last.
Asc,
@@ -371,7 +371,7 @@ pub struct LabelMatcher {
}
/// The operator of a [`LabelMatcher`].
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize, Hash)]
pub enum MatchOp {
Equal,
NotEqual,
src/promql/src/vector_match.rs
+12 -6
View File
@@ -67,7 +67,7 @@ pub(crate) fn visited_pairs() -> usize {
}
/// A PromQL binary operator between two instant vectors.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize, Hash)]
pub enum BinaryOp {
Add,
Sub,
@@ -113,7 +113,7 @@ impl BinaryOp {
}
/// The labels two series are matched by.
#[derive(Clone, Debug, PartialEq, Eq)]
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize, Hash)]
pub enum MatchingLabels {
/// `on(...)`: only these labels, `__name__` included when listed.
On(Vec<Arc<str>>),
@@ -123,7 +123,7 @@ pub enum MatchingLabels {
}
/// How many series of each side may share a match group.
#[derive(Clone, Debug, PartialEq, Eq)]
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize, Hash)]
pub enum Cardinality {
OneToOne,
/// `group_left(...)`: the right side is the "one" side; the listed labels are copied from it.
@@ -133,7 +133,7 @@ pub enum Cardinality {
}
/// A vector–vector binary operation with its modifiers.
#[derive(Clone, Debug, PartialEq, Eq)]
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize, Hash)]
pub struct BinaryMatching {
pub op: BinaryOp,
/// `bool`: comparisons return 1 or 0 instead of filtering.
@@ -146,7 +146,7 @@ pub struct BinaryMatching {
/// The values of `fill`, `fill_left` and `fill_right`, which stand in for a sample that has no
/// match on the other side: `lhs` for a missing left-hand sample and `rhs` for a missing
/// right-hand one, as the query writes them, `group_right` included (prometheus/prometheus#19939).
#[derive(Clone, Copy, Debug, Default)]
#[derive(Clone, Copy, Debug, Default, serde::Serialize, serde::Deserialize)]
pub struct FillValues {
pub lhs: Option<f64>,
pub rhs: Option<f64>,
@@ -161,9 +161,15 @@ impl PartialEq for FillValues {
}
impl Eq for FillValues {}
impl std::hash::Hash for FillValues {
fn hash<H: std::hash::Hasher>(&self, h: &mut H) {
std::hash::Hash::hash(&self.lhs.map(f64::to_bits), h);
std::hash::Hash::hash(&self.rhs.map(f64::to_bits), h);
}
}
/// A PromQL set operator between two instant vectors.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize, Hash)]
pub enum SetOp {
And,
Or,