feat(datatypes): describe label decoding in the raw series frame schema (#9786)

* feat(datatypes): describe label decoding in the raw series frame schema

A raw series frame written with key-encoded labels carries the primary key
as opaque bytes; only the binding held by the scanner says how to turn
those bytes back into a metric name and labels. A consumer that receives
the frame without that binding cannot decode it, so the schema now
describes its own label layout.

The description is compact JSON under `greptime.series.label_decode`,
carrying a version, the metric name, the primary-key encoding, the ordered
key columns with their physical types, and the logical tags to expose
-- exactly what decoding reads. It is derived from the binding, so there
is one writer, and a reader that does not implement the version it finds
rejects the frame instead of guessing.

`unpack_raw_series` keeps its binding argument and now requires it to
agree with the description the schema declares, so the two cannot drift.
`PrimaryKeyLabelDecoder` reduces both of its inputs to one internal key
layout, so the metadata-backed and description-backed paths share their
validation instead of growing a second decoder.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* refactor(datatypes): compare label descriptions by what their order means

The frame's declared label-decode description was compared to the binding's with
plain equality, but the description has one part whose order carries no meaning:
the decoder sorts logical tags by name when it binds them, so the same tags listed
in another order denote the same labels and were being refused.

`same_label_layout` canonicalises exactly that part and leaves `primary_key` in
place, because for a dense key the array position is the encoding position: the same
columns in another order are a different key, not the same key reordered. Both
fields now say which order they mean.

Three checks go with it. The binding stays for a stated reason -- it establishes
identity, the physical region key layout a decoded identity came from -- rather than
because a count of call sites kept it. The post-pack schema comparison in the series
scan could not fire: the leaf already re-derives that schema and compares it, and
every `SeriesSchema` field is covered by the checks between them. And two test
groups re-asserted what the tracked v1_metadata fixture pins case by case.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* feat(promql): decode a raw frame from the frame alone

`unpack_raw_series` took the label mode, the ordering and the caller's binding
along with the stream, so a consumer that received a frame from a peer still
needed a local copy of the binding to read it — the thing the frame's own
description exists to avoid, and the reason the description was freezing a wire
format for nobody.

The stream is now the only input: the field schema, the label mode, the label
decode description and the producer ordering all come from the frame's schema,
and every frame is validated whether or not a caller asked for it. `expected`
stays as an optional check on the schema that is declared — field names and their
order, and no evaluation grid — and supplies nothing to the decoder, so a caller
cannot smuggle in a layout the frame does not declare.

Three call sites compared the caller's binding against the frame and all three
were load-bearing: a peer's response is not trusted to be the region that was
planned, and a native plan's child is an arbitrary execution plan. Those became
`ensure_frame_matches_label_source`, which states what it compares: the label
layout, not the region. Two regions with the same tag ids and names compare equal,
and the doc now says so instead of claiming to establish identity.

A stream with no scan-bound identity reports `NO_BOUND_LABEL_SOURCE` and answers
`label_decode_spec()` instead; every consumer that decides whether to decode reads
the description, so the placeholder never stands in for "these labels are
complete".

The merge contract is the description, not the binding: inputs are compared by
the layout they declare, and a binding that could not be expressed on the wire — a
`RegionMetadata` snapshot — no longer makes the merge stricter than the frames it
merges. `logical_tags` gets one rule rather than two: the writer canonicalizes a
copy, because the caller's binding keeps its own order and the table scan compares
producer tag lists positionally, and readers compare by equality.

The text embeds `ConcreteDataType`'s serde, which `LABEL_DECODE_VERSION` has to
cover. Rather than mirror all 25 datatype variants into a type this module owns, a
frozen fixture records the text of every variant and an exhaustive match makes a
new variant fail to compile until it is recorded. The version policy is written
next to the constant: exact-text comparison across nodes, unknown versions
rejected, any same-version change breaking — with the planning-time fallback
stated as intended and marked not implemented.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* refactor(store-api): make every key-encoded raw stream declare its layout

A raw series stream whose chunks carry key-encoded labels had to be decoded
against a scan-bound `RawLabelSource`, and two consumers compensated for
producers that stayed silent by deriving a description from that binding:
`pack_raw_series` and the merge's `declared_layout`. `materialize_raw_labels`
did not compensate, so an unframed producer stream reached it with no
description and a chunk carrying no labels, and failed.

The stream itself is the authority for its chunks. A key-encoded producer
knows its own region metadata and can state the layout it writes, exactly as a
frame states the layout it carries, so the split contract — "a producer
declares its identity, a frame declares its layout" — is replaced by one rule:
a stream whose chunks carry key-encoded labels declares the description, and a
stream whose chunks carry complete labels answers `None`.

mito2's `RawStream` caches the derivation at construction, so the declaration
exists before the first poll and for an empty stream. The trait documentation
states the rule and no longer exempts unframed producers. Both fallbacks are
gone: `pack_raw_series` reads the stream's declaration only, and rejects a
key-encoded stream that declares none rather than packing it as an
explicit-label frame, which would describe chunks carrying no labels as if they
carried all of them. The merge reads the declaration only.

Test fixtures that model key-encoded producers now declare their layout too;
fixtures that carry complete labels keep answering `None`. The new mito2 test
decodes a real produced key through the declared description for both dense and
sparse encodings, and checks the declaration is present before polling even for
an empty stream.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* refactor(common-query): let the declaration decide the frame mode alone

`pack_raw_series` rejected a key-encoded stream that declared no description
outright, so a stream that stays silent could not be packed at all. That check
was redundant: the declaration already decides the frame's label mode, and the
first chunk whose labels disagree with it is refused by `RawSeriesChunkCursor`
with "raw chunk label mode does not match frame mode", so no such stream can
pack data either way.

What the check added was an early error for a producer that already breaks the
contract, in exchange for reading the stream's scan binding here — in the one
function whose rule is that the declaration is the only decode input. Drop it,
so pack keeps deciding the mode from the declaration and nothing else.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* test(store-api): assert the fixture's declared label decode description

The mock prepared source derives a `LabelDecodeSpec` from its label source and
reports it through `RawSeriesStream::label_decode_spec`, but no assertion read
it, so the derivation could drift from the chunks the fixture actually builds.
Check it against the description the fixture's own key columns imply: an
explicit source declares nothing, a primary-key source declares the metric, the
dense encoding, the one key column and its tag.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* refactor(promql): let the frame's declaration be the only label authority

Two things still treated a caller-supplied binding as authoritative after the
declaration became the decode input:

- A merged stream forwarded its first input's scan-bound identity, though a
  merge is not a region scan and decodes from the declaration alone. It now
  reports no bound identity.
- `KeyColumn.tag` repeated an invariant every key column is a tag. The check
  moves to the one place a decoder is bound, so a snapshot built without
  validation is still rejected instead of silently decoded.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

---------

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
2026-10-21 19:11:24 +08:00
committed by GitHub
parent f235434b7f
commit 7309800aef
25 changed files with 3265 additions and 602 deletions
+218 -31
View File
@@ -15,10 +15,18 @@
//! Streaming boundary between finite raw-series chunks and DataFusion record batches.
//!
//! Schema fields identify the ordered logical payload fields and codec layout; additional
//! application metadata is permitted when stable across the stream. The caller-provided
//! [`RawLabelSource`] is the label binding, not authenticated by the frame schema or field
//! provenance. This adapter preserves scan candidates only; it is not a residual filter,
//! authorization check, remote transport, or compaction boundary. Only
//! application metadata is permitted when stable across the stream. A key-encoded frame also
//! declares its own label decode description, and the frame is the sole source of decode
//! information: [`unpack_raw_series`] reads the fields, the label mode, the description, and the
//! producer ordering from the received schema, keeps the description on the returned stream, and
//! therefore needs no [`RawLabelSource`]. A caller that holds a scan-bound identity — the physical
//! region key layout a decoded identity came from — compares that identity's declared label
//! decode layout with [`ensure_frame_matches_label_source`] instead of passing it as a decode
//! input; that comparison is equality of the description alone and proves nothing about region
//! identity or provenance, and neither side is authenticated as field provenance. This adapter
//! preserves
//! scan candidates only; it is not a residual filter, authorization check, remote transport, or
//! compaction boundary. Only
//! [`SeriesOrdering::Contiguous`] and [`SeriesOrdering::KeyThenTime`] are accepted;
//! `Interleaved` is rejected before any poll.
@@ -32,14 +40,19 @@ use datafusion_common::DataFusionError;
use datatypes::arrow::datatypes::{Schema as ArrowSchema, SchemaRef as ArrowSchemaRef};
use datatypes::error::InvalidSeriesSnafu;
use datatypes::series::frame::{
LABEL_DECODE_VERSION, LabelDecodeColumn, LabelDecodeEncoding, LabelDecodeSpec, LabelDecodeTag,
RawLabelMode, RawSeriesChunkCursor, RawSeriesFrameBuilder, RawSeriesFrameDecoder,
raw_series_frame_schema,
raw_series_frame_label_decode, raw_series_frame_schema_with_label_decode,
raw_series_frame_series_schema,
};
use datatypes::series::raw::RawSeriesChunk;
use datatypes::series::samples::SeriesSchema;
use datatypes::series::{MetricName, SeriesOrdering};
use futures_util::Stream;
use store_api::region_engine::{RawLabelSource, RawSeriesStream, SendableRawSeriesStream};
use store_api::codec::PrimaryKeyEncoding;
use store_api::region_engine::{
NO_BOUND_LABEL_SOURCE, RawLabelSource, RawSeriesStream, SendableRawSeriesStream,
};
mod record_batches;
@@ -50,13 +63,35 @@ mod tests;
const ORDERING_KEY: &str = "greptime.series.ordering";
/// Builds the canonical raw frame schema with the producer's guaranteed ordering.
/// Builds the canonical raw frame schema with the producer's label binding and ordering.
///
/// Ordering records a producer guarantee; it is not inferred from frame contents. `Interleaved`
/// has no v1 wire representation and is rejected before the schema is assembled.
/// has no v1 wire representation and is rejected before the schema is assembled. The label binding
/// determines both the label mode and, for a primary-key binding, the self-describing label decode
/// description written into the schema, so a frame can never declare a label layout it does not
/// describe.
pub fn raw_series_frame_schema_with_ordering(
schema: &SeriesSchema,
label_source: &RawLabelSource,
ordering: SeriesOrdering,
) -> datafusion_common::Result<ArrowSchemaRef> {
frame_schema_with(
schema,
label_mode(label_source),
label_decode_spec(label_source).as_ref(),
ordering,
)
}
/// Assembles a declared frame schema from an explicit label mode and decode description.
///
/// The mode and description pair is checked by
/// [`raw_series_frame_schema_with_label_decode`], so a key mode without a description and an
/// explicit mode with one are both rejected here instead of reaching the wire.
fn frame_schema_with(
schema: &SeriesSchema,
mode: RawLabelMode,
decode: Option<&LabelDecodeSpec>,
ordering: SeriesOrdering,
) -> datafusion_common::Result<ArrowSchemaRef> {
let ordering = match ordering {
@@ -68,7 +103,8 @@ pub fn raw_series_frame_schema_with_ordering(
));
}
};
let schema = raw_series_frame_schema(schema, mode).map_err(codec_error)?;
let schema =
raw_series_frame_schema_with_label_decode(schema, mode, decode).map_err(codec_error)?;
let mut metadata = schema.metadata().clone();
metadata.insert(ORDERING_KEY.to_string(), ordering.to_string());
Ok(Arc::new(ArrowSchema::new_with_metadata(
@@ -77,6 +113,56 @@ pub fn raw_series_frame_schema_with_ordering(
)))
}
/// Derives the label decode description a frame must carry for this binding.
///
/// A primary-key binding reduces to exactly what decoding reads: the metric name, the primary-key
/// encoding, the ordered key columns with their physical types, and the logical tags. Region
/// metadata guarantees the key list holds every tag column and nothing else, so the description
/// needs no per-column role. Explicit bindings already carry complete labels and declare no
/// description.
///
/// The result is canonical: its logical tags are in [`LabelDecodeSpec::canonicalize`] order, which
/// is what the writer serializes, so a derived description and a received one for the same labels
/// are the same value and the same text. Derivation cannot fail, because a physical type is copied
/// rather than translated; a description a reader cannot honour is rejected when it is read, and a
/// decoder rejects the tags it cannot bind.
pub fn label_decode_spec(source: &RawLabelSource) -> Option<LabelDecodeSpec> {
let RawLabelSource::PrimaryKey {
metric_name,
physical_metadata,
logical_tags,
} = source
else {
return None;
};
let primary_key = physical_metadata
.primary_key_columns()
.map(|column| LabelDecodeColumn {
column_id: column.column_id,
name: column.column_schema.name.clone(),
physical_type: column.column_schema.data_type.clone(),
})
.collect();
let mut description = LabelDecodeSpec {
version: LABEL_DECODE_VERSION,
metric_name: metric_name.to_string(),
encoding: match physical_metadata.primary_key_encoding {
PrimaryKeyEncoding::Dense => LabelDecodeEncoding::Dense,
PrimaryKeyEncoding::Sparse => LabelDecodeEncoding::Sparse,
},
primary_key,
logical_tags: logical_tags
.iter()
.map(|tag| LabelDecodeTag {
column_id: tag.column_id,
name: tag.name.to_string(),
})
.collect(),
};
description.canonicalize();
Some(description)
}
/// Reads the declared producer ordering, conservatively treating legacy frames as contiguous.
///
/// Only the two orderings representable on the wire can be returned.
@@ -99,17 +185,23 @@ pub fn raw_series_frame_ordering(
/// work bound, not a resident-memory cap. A ready batch is yielded without polling the source
/// again. Producers must keep each series' chunks contiguous. Validation checks adjacent chunk
/// conditions but does not certify nonadjacent recurrence for `Contiguous` ordering.
///
/// The stream's own declaration is the only decode input: a frame unpacked without a scan-bound
/// [`RawLabelSource`] round trips unchanged instead of being re-declared from a binding it does not
/// have, and nothing is derived from a binding here. The declaration also decides the frame's label
/// mode, so a chunk whose labels disagree with it is rejected when that chunk is coded.
pub fn pack_raw_series(
source: SendableRawSeriesStream,
) -> datafusion_common::Result<DfSendableRecordBatchStream> {
let ordering = source.ordering();
ensure_supported_ordering(ordering)?;
let schema = source.schema().clone();
let label_source = source.label_source().clone();
let mode = label_mode(&label_source);
// The declared frame schema carries the producer ordering; the shared builder's schema
// intentionally omits it, so every ready batch is declared before it is yielded.
let frame_schema = raw_series_frame_schema_with_ordering(&schema, mode, ordering)?;
let decode = source.label_decode_spec().cloned();
let mode = declared_label_mode(&decode);
// The declared frame schema carries the producer ordering and the label decode description;
// the shared builder's schema intentionally omits both, so every ready batch is declared under
// this schema before it is yielded, including when no row is ever produced.
let frame_schema = frame_schema_with(&schema, mode, decode.as_ref(), ordering)?;
let builder = RawSeriesFrameBuilder::try_new(
&schema,
mode,
@@ -129,33 +221,48 @@ pub fn pack_raw_series(
}))
}
/// Unpacks DataFusion batches using the caller's explicit scan binding and ordering.
/// Unpacks DataFusion batches declared by [`pack_raw_series`].
///
/// Only [`SeriesOrdering::Contiguous`] and [`SeriesOrdering::KeyThenTime`] streams are
/// accepted; an `Interleaved` declaration is rejected before the stream is returned, including
/// for an empty stream. Producers must keep each series' chunks contiguous. Validation checks
/// adjacent chunk conditions but does not certify nonadjacent recurrence for `Contiguous`
/// ordering.
/// The frame is the sole source of decode information: the payload fields, the label mode, the
/// label decode description, and the producer ordering are all read from the stream's own schema,
/// so unpacking a frame a producer wrote needs no local copy of its binding. Only
/// [`SeriesOrdering::Contiguous`] and [`SeriesOrdering::KeyThenTime`] streams are accepted; an
/// `Interleaved` declaration is rejected before the stream is returned, including for an empty
/// stream. Producers must keep each series' chunks contiguous. Validation checks adjacent chunk
/// conditions but does not certify nonadjacent recurrence for `Contiguous` ordering.
///
/// The supplied binding is retained unchanged and is not validated as field provenance.
/// Every frame is validated here, unconditionally. The declared layout must be this revision's, the
/// label decode description must parse, name a known version, and agree with the declared label
/// mode — a key frame must declare one and an explicit frame must not, since explicit labels are
/// already complete. The declared fields and the empty batch are then decoded once, so a stream that
/// could not be decoded is rejected before it is returned.
///
/// `expected` is a check, never an input. When it is given it must match the schema the frame
/// declares — the same logical fields in the same order, and no evaluation grid — and a mismatch is
/// an error. The decoder and the returned stream's schema are built from the frame either way, so a
/// caller cannot supply a field layout the frame does not declare.
pub fn unpack_raw_series(
source: DfSendableRecordBatchStream,
expected: Arc<SeriesSchema>,
label_source: RawLabelSource,
ordering: SeriesOrdering,
expected: Option<Arc<SeriesSchema>>,
) -> datafusion_common::Result<SendableRawSeriesStream> {
ensure_supported_ordering(ordering)?;
let mode = label_mode(&label_source);
let frame_schema = source.schema();
let decoder = RawSeriesFrameDecoder::try_new(&expected, mode).map_err(codec_error)?;
let declared = raw_series_frame_label_decode(frame_schema.as_ref()).map_err(codec_error)?;
let mode = declared_label_mode(&declared);
let schema =
Arc::new(raw_series_frame_series_schema(frame_schema.as_ref()).map_err(codec_error)?);
if let Some(expected) = &expected {
ensure_declared_schema(&schema, expected)?;
}
let ordering = raw_series_frame_ordering(frame_schema.as_ref())?;
let decoder = RawSeriesFrameDecoder::try_new(&schema, mode).map_err(codec_error)?;
// Validate even empty streams' declared schema and layout before returning the stream.
let empty = DfRecordBatch::new_empty(frame_schema.clone());
decoder.decode(&empty).map_err(codec_error)?;
Ok(Box::pin(UnpackStream {
source: Some(source),
schema: expected,
schema,
frame_schema,
label_source,
label_decode_spec: declared,
ordering,
validator: SequenceValidator::new(ordering),
pending: Vec::new().into_iter(),
@@ -163,6 +270,66 @@ pub fn unpack_raw_series(
}))
}
/// Rejects an `expected` series schema the frame does not declare.
///
/// Only the two things a caller can get wrong are checked, because those are the only two things
/// [`SeriesSchema`] holds: the logical fields and their order, and an evaluation grid a raw frame
/// cannot carry.
fn ensure_declared_schema(
declared: &SeriesSchema,
expected: &SeriesSchema,
) -> datafusion_common::Result<()> {
if expected.eval_grid.is_some() {
return Err(invalid(
"expected raw series schema declares an evaluation grid the frame cannot carry",
));
}
if expected.fields.len() != declared.fields.len()
|| expected
.fields
.iter()
.zip(declared.fields.iter())
.any(|(expected, declared)| expected.name != declared.name)
{
return Err(invalid(
"raw frame fields do not match the expected series schema",
));
}
Ok(())
}
/// Checks that a frame declares the label decode layout a scan-bound binding describes.
///
/// This compares one thing: the description parsed from the frame's
/// `greptime.series.label_decode` metadata against the description derived from `label_source`, by
/// plain equality. A difference in the version, the metric name, the primary-key encoding, the
/// ordered key columns and their physical types, or the logical tags is a mismatch. Nothing else is
/// compared, so agreeing descriptions do **not** establish that these are the region's chunks: no
/// region id, table id, or provenance is part of either side, and two regions whose tags have the
/// same ids, names, and types produce equal descriptions. The check answers "does this frame describe
/// the label layout this binding expects", never "is this the region I asked for".
///
/// It belongs to the caller that holds both sides: a frame describes its own labels, so
/// [`unpack_raw_series`] decodes it without a binding, and a caller that received the frame from a
/// peer compares the declaration here instead of decoding under whichever side it trusts.
///
/// The comparison is plain equality. It used to canonicalize logical tag order on both sides, which
/// gave local comparisons one rule and the byte-exact remote comparison another; the writer now
/// canonicalizes instead (see [`LabelDecodeSpec::canonicalize`]), so a description that reaches a
/// reader already has one spelling and equality says everything there is to say about it.
pub fn ensure_frame_matches_label_source(
frame: &ArrowSchema,
label_source: &RawLabelSource,
) -> datafusion_common::Result<()> {
let declared = raw_series_frame_label_decode(frame).map_err(codec_error)?;
if declared != label_decode_spec(label_source) {
return Err(invalid(
"raw frame label decode description does not match the supplied binding",
));
}
Ok(())
}
fn label_mode(source: &RawLabelSource) -> RawLabelMode {
match source {
RawLabelSource::PrimaryKey { .. } => RawLabelMode::Key,
@@ -170,6 +337,15 @@ fn label_mode(source: &RawLabelSource) -> RawLabelMode {
}
}
/// Returns the label mode a frame declares: a key frame describes its key, an explicit frame
/// carries complete labels and declares no description.
fn declared_label_mode(declared: &Option<LabelDecodeSpec>) -> RawLabelMode {
match declared {
Some(_) => RawLabelMode::Key,
None => RawLabelMode::Explicit,
}
}
/// Rejects orderings this adapter cannot validate; a bad declaration is an error, not a fallback.
fn ensure_supported_ordering(ordering: SeriesOrdering) -> datafusion_common::Result<()> {
match ordering {
@@ -408,7 +584,8 @@ struct UnpackStream {
source: Option<DfSendableRecordBatchStream>,
schema: Arc<SeriesSchema>,
frame_schema: datatypes::arrow::datatypes::SchemaRef,
label_source: RawLabelSource,
/// The description parsed from the frame schema; the only input decoding needs.
label_decode_spec: Option<LabelDecodeSpec>,
ordering: SeriesOrdering,
validator: SequenceValidator,
pending: std::vec::IntoIter<RawSeriesChunk>,
@@ -420,8 +597,18 @@ impl RawSeriesStream for UnpackStream {
&self.schema
}
/// Returns [`NO_BOUND_LABEL_SOURCE`]: a frame establishes its own label layout and carries no
/// scan-bound identity, so decoding reads [`Self::label_decode_spec`] and never this. There is
/// no physical region key layout to report and none is invented from the description, which
/// deliberately carries no region metadata.
fn label_source(&self) -> &RawLabelSource {
&self.label_source
&NO_BOUND_LABEL_SOURCE
}
/// Returns the label decode description the frame declared, which is what decoding reads; an
/// explicit-label frame declares none.
fn label_decode_spec(&self) -> Option<&LabelDecodeSpec> {
self.label_decode_spec.as_ref()
}
fn ordering(&self) -> SeriesOrdering {
@@ -1130,13 +1130,7 @@ async fn exact_int64_filter_sort_adapter_frame_pipeline() {
.into(),
eval_grid: None,
});
let mut decoded = super::super::unpack_raw_series(
frames,
expected,
store_api::region_engine::RawLabelSource::Explicit,
SeriesOrdering::Contiguous,
)
.unwrap();
let mut decoded = super::super::unpack_raw_series(frames, Some(expected)).unwrap();
let chunk = decoded.next().await.unwrap().unwrap();
assert!(chunk.is_last);
assert_eq!(chunk.time.as_buffer().as_ref(), &[1, 2]);
File diff suppressed because it is too large Load Diff
@@ -193,13 +193,7 @@ async fn streaming_round_trip_preserves_sliced_multifield_payload_and_owners() {
chunk: Some(chunk),
});
let frames = pack_raw_series(input).unwrap();
let mut decoded = unpack_raw_series(
frames,
schema,
RawLabelSource::Explicit,
SeriesOrdering::Contiguous,
)
.unwrap();
let mut decoded = unpack_raw_series(frames, Some(schema)).unwrap();
let mut output = Vec::new();
while let Some(item) = decoded.next().await {
output.push(item.unwrap());
@@ -425,13 +419,7 @@ async fn single_row_sliced_payload_keeps_zero_copy_buffers_and_releases_owners()
chunk: Some(chunk),
});
let frames = pack_raw_series(input).unwrap();
let mut decoded = unpack_raw_series(
frames,
schema,
RawLabelSource::Explicit,
SeriesOrdering::Contiguous,
)
.unwrap();
let mut decoded = unpack_raw_series(frames, Some(schema)).unwrap();
let mut output = Vec::new();
while let Some(item) = decoded.next().await {
output.push(item.unwrap());
+327 -3
View File
@@ -18,11 +18,45 @@
//! chunk and [`RawSeriesFrameBuilder`] emits rows directly as Arrow arrays once a row or estimated
//! byte bound is reached. Decoding reuses one prepared schema per stream through
//! [`RawSeriesFrameDecoder`]. Column layout, metadata, and wire strings stay on the frozen `v1`
//! contract.
//! contract; a key-encoded frame additionally declares a versioned [`LabelDecodeSpec`] so its
//! labels can be decoded from the received schema alone.
//!
//! # Label decode description versioning
//!
//! The description is one `greptime.series.label_decode` metadata entry, and remote plans compare
//! `greptime.series.*` metadata **as exact text, byte for byte**: a frontend and a datanode that
//! write different text for the same labels disagree about the schema, and the query fails instead
//! of falling back. The rules that follow from that are:
//!
//! - A reader rejects a version it does not implement, rather than guessing. This module writes and
//! reads [`LABEL_DECODE_VERSION`] only.
//! - [`LABEL_DECODE_VERSION`] covers the whole text, including each key column's
//! [`LabelDecodeColumn::physical_type`]. That field is a [`ConcreteDataType`], so the text embeds
//! [`ConcreteDataType`]'s serde shape: it is not spelled here, and a change in that type's
//! serialization does alter this text. What keeps such a change from going unnoticed is the
//! frozen fixture `tests/compatibility/wire/raw_series_label_decode/v1_physical_types.json`,
//! which records the text of every datatype variant and fails the crate's tests when any of them
//! is serialized differently.
//! - Any change to the shape of this text — a renamed or added field, a different
//! [`ConcreteDataType`] spelling, a different `logical_tags` order — is a **breaking** change at
//! the same version, because it makes two nodes of different builds disagree. Such a change must
//! bump [`LABEL_DECODE_VERSION`] and update that fixture with it.
//! - The text is canonical: [`LabelDecodeSpec::canonicalize`] fixes the only part whose order
//! carries no meaning, so equal labels always write equal bytes.
//!
//! Deployment rule: during a rolling upgrade a frame may carry a version the receiving node does not
//! implement, and a byte-for-byte schema comparison may fail for that reason alone. The intended
//! behaviour is that such a mismatch disables the native series path at planning time and falls back
//! to the generic path, instead of failing the query at execution.
//!
//! TODO(narthex): implement that planning-time fallback. No planner reads
//! `raw_series_frame_label_decode` to decide support today, so a version or text mismatch currently
//! surfaces as an execution-time error from the raw-frame decode or the remote schema comparison.
#[cfg(test)]
mod tests;
use std::collections::HashSet;
use std::sync::Arc;
use arrow::array::{
@@ -33,12 +67,14 @@ use arrow::buffer::{Buffer, NullBuffer, OffsetBuffer, ScalarBuffer};
use arrow::compute::concat;
use arrow::datatypes::{DataType, Field, Fields, Schema, SchemaRef};
use arrow::record_batch::{RecordBatch, RecordBatchOptions};
use serde::{Deserialize, Serialize};
use crate::data_type::ConcreteDataType;
use crate::error::{InvalidSeriesSnafu, Result};
use crate::series::histogram::{HistogramBatch, fields as histogram_fields};
use crate::series::raw::{RawSeriesChunk, RawTimestamps};
use crate::series::samples::{
DenseSamplesRef, HistogramSamples, Samples, SeriesFields, SeriesSchema,
DenseSamplesRef, FieldSpec, HistogramSamples, Samples, SeriesFields, SeriesSchema,
};
use crate::series::{Label, MetricName, invalid_series};
@@ -46,6 +82,12 @@ const LAYOUT: &str = "greptime.series.layout";
const CONTENT: &str = "greptime.series.content";
const LABELS: &str = "greptime.series.labels";
const TIME: &str = "greptime.series.time";
const GRID: &str = "greptime.series.grid";
/// Metadata key of the versioned label decode description carried by key-encoded frames.
const LABEL_DECODE: &str = "greptime.series.label_decode";
/// Version of the label decode description this revision writes and reads.
pub const LABEL_DECODE_VERSION: u32 = 1;
/// Whether a raw frame carries only the complete key or explicit metric labels as well.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -56,7 +98,166 @@ pub enum RawLabelMode {
Explicit,
}
/// Primary-key encoding of the labels a key-encoded frame carries.
///
/// This mirrors the storage contract's `PrimaryKeyEncoding` without making this crate depend on
/// it; Dense and Sparse keys use different decode layouts.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum LabelDecodeEncoding {
/// Dense key: every primary-key field is encoded in order, with no ids on the wire.
Dense,
/// Sparse key: `(column id, value)` pairs, so the layout only frames the field stream.
Sparse,
}
/// One ordered key column needed to decode key-encoded labels.
///
/// Only tag columns reach this list: region metadata rejects a primary key that holds the time
/// index or a non-`Tag` column, and it requires the primary key to name every tag column. The
/// array's position is therefore the encoding position, with nothing to disambiguate.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct LabelDecodeColumn {
/// Physical column id of this key column.
pub column_id: u32,
/// Physical column name; a logical tag of the same id must repeat it.
pub name: String,
/// Physical data type of the encoded value, in [`ConcreteDataType`]'s own serde form. Dense
/// keys carry no types on the wire, so the codec is reconstructed from this exact ordered type
/// list.
pub physical_type: ConcreteDataType,
}
/// One logical tag name bound to a primary-key column.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct LabelDecodeTag {
/// Physical column id this logical tag reads.
pub column_id: u32,
/// Logical tag name exposed to queries.
pub name: String,
}
/// Versioned, self-contained description of how to decode key-encoded labels.
///
/// It carries exactly what decoding a primary-key label source reads and nothing else: no region
/// identity, no field columns, no time index. A consumer that receives the frame schema can
/// therefore reconstruct an equivalent decoder without a scan-side binding, while a description
/// that is missing, of an unknown version, or structurally inconsistent is a hard error instead of
/// a silent fallback.
///
/// It is JSON-encoded in one schema metadata entry, always through
/// [`raw_series_frame_schema_with_label_decode`], so both the producing and the expecting side of a
/// `greptime.series.*` exact metadata comparison see byte-identical text. The text is
/// self-describing but not a general interchange format: readers reject unknown fields here only
/// at this structure's own levels, while [`LabelDecodeColumn::physical_type`] is decoded by
/// [`ConcreteDataType`]'s own representation and is therefore exactly as stable as that type's
/// serde. The frozen text this revision writes is pinned by
/// `tests/compatibility/wire/raw_series_label_decode/v1_physical_types.json`; see the module
/// documentation for what a change to it means.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct LabelDecodeSpec {
/// Description version; see [`LABEL_DECODE_VERSION`].
pub version: u32,
/// Metric name attached to every decoded label set; may be empty.
pub metric_name: String,
/// Primary-key encoding that determines the decode layout.
pub encoding: LabelDecodeEncoding,
/// Every tag column of the primary key, in primary-key encoding order. This is exactly the
/// set of encoded key fields: region metadata forbids a primary key from naming the time
/// index or a field column, and requires it to name every tag column, so no time-index or
/// field entry can appear here. The order is significant and is compared as written: a dense
/// decode reads field *n* at position *n*, so the same columns in another order describe a
/// different key, not the same key in another order.
pub primary_key: Vec<LabelDecodeColumn>,
/// Logical tags, as bound by the producer, in [`LabelDecodeSpec::canonicalize`] order.
///
/// The order carries no meaning: the decoder sorts the tags by name when it binds them
/// (`src/mito-codec/src/labels.rs:235`), so a decoded label set does not depend on it. It is
/// nevertheless fixed rather than left to the caller, because this description is compared as
/// exact text across nodes: two producers with the same tags in another order must write
/// byte-identical text, or one side rejects the other as a schema mismatch.
///
/// The semantic checks on these tags (reserved and internal names, name equality with the bound
/// column, membership in `primary_key`) are the decoder's, matching what a typed binding gets.
pub logical_tags: Vec<LabelDecodeTag>,
}
impl LabelDecodeSpec {
/// Puts the order-insensitive part of the description in its canonical order.
///
/// [`raw_series_frame_schema_with_label_decode`] applies this to a copy before serializing, and
/// so does the description a binding derives (in `common_query::series_frame`), so both sides of
/// a text comparison produce the same bytes and a reader can compare descriptions structurally.
/// `primary_key` is deliberately untouched: its order is the encoding order.
///
/// The writer canonicalizes a **copy** on purpose. A caller's binding keeps its own
/// `logical_tags` order, and that order is meaningful elsewhere: `src/table/src/table/scan.rs`
/// compares two bindings' tag lists positionally (`lt.iter().zip(rt.iter())`), so reordering the
/// caller's list here would make two bindings for the same labels compare unequal. Only the
/// serialized text is canonical; nothing outside it is rewritten.
///
/// Sorting by `(column_id, name)` is a total order for a validated description, which names
/// each tag column and each tag name at most once, so the canonical form is unique.
pub fn canonicalize(&mut self) {
self.logical_tags
.sort_unstable_by(|a, b| (a.column_id, &a.name).cmp(&(b.column_id, &b.name)));
}
/// Checks the version and the structural invariants a decoder relies on.
///
/// Semantic binding checks (reserved names, internal columns, name equality against the bound
/// column) stay in the decoder so both a typed binding and a decoded description fail with the
/// same errors.
pub fn validate(&self) -> Result<()> {
if self.version != LABEL_DECODE_VERSION {
return Err(err(format!(
"unsupported label decode description version {}",
self.version
)));
}
let mut column_ids = HashSet::with_capacity(self.primary_key.len());
for column in &self.primary_key {
if !column_ids.insert(column.column_id) {
return Err(err(format!(
"duplicate primary-key column id {} in label decode description",
column.column_id
)));
}
}
let mut tag_ids = HashSet::with_capacity(self.logical_tags.len());
let mut tag_names = HashSet::with_capacity(self.logical_tags.len());
for tag in &self.logical_tags {
if !tag_ids.insert(tag.column_id) {
return Err(err(format!(
"duplicate logical tag column id {} in label decode description",
tag.column_id
)));
}
if !tag_names.insert(tag.name.as_str()) {
return Err(err(format!(
"duplicate logical tag {} in label decode description",
tag.name
)));
}
if !column_ids.contains(&tag.column_id) {
return Err(err(format!(
"logical tag {} is not a primary-key column of the label decode description",
tag.name
)));
}
}
Ok(())
}
}
/// Builds the typed schema for a raw frame, preserving logical field names and order.
///
/// This is the column layout shared with the frame builder and decoder; it deliberately carries no
/// label decode description. A key-encoded stream must be declared through
/// [`raw_series_frame_schema_with_label_decode`], which refuses such a layout without one.
pub fn raw_series_frame_schema(schema: &SeriesSchema, mode: RawLabelMode) -> Result<SchemaRef> {
if schema.eval_grid.is_some() {
return invalid_series("raw series schema cannot have an evaluation grid");
@@ -137,6 +338,129 @@ pub fn raw_series_frame_schema(schema: &SeriesSchema, mode: RawLabelMode) -> Res
Ok(Arc::new(Schema::new_with_metadata(columns, metadata)))
}
/// Builds the typed schema for a raw frame together with its label decode description.
///
/// The description is written only for [`RawLabelMode::Key`], where labels stay encoded in
/// `read_key`; explicit-label frames carry complete labels and must not declare one. Both
/// directions are enforced here so a frame cannot claim a label layout it does not describe.
pub fn raw_series_frame_schema_with_label_decode(
schema: &SeriesSchema,
mode: RawLabelMode,
decode: Option<&LabelDecodeSpec>,
) -> Result<SchemaRef> {
match (mode, decode) {
(RawLabelMode::Key, None) => {
return Err(err(
"key-encoded raw frames must carry a label decode description",
));
}
(RawLabelMode::Explicit, Some(_)) => {
return Err(err("explicit raw frames carry no label decode description"));
}
(RawLabelMode::Key, Some(_)) | (RawLabelMode::Explicit, None) => {}
}
let layout = raw_series_frame_schema(schema, mode)?;
let Some(decode) = decode else {
return Ok(layout);
};
// Writing is faithful: the description records what the producer declares, and nothing is
// rewritten or dropped. The one exception is the order of `logical_tags`, which carries no
// meaning to a decoder and which this text nevertheless has to fix so two producers of the same
// labels write the same bytes; a copy is canonicalized and the caller's value is untouched. A
// description this revision cannot honour is rejected when it is read, which is the only
// direction a consumer can rely on.
let mut canonical = decode.clone();
canonical.canonicalize();
let encoded = serde_json::to_string(&canonical)
.map_err(|e| err(format!("cannot encode label decode description: {e}")))?;
let mut metadata = layout.metadata().clone();
metadata.insert(LABEL_DECODE.to_string(), encoded);
Ok(Arc::new(Schema::new_with_metadata(
layout.fields().clone(),
metadata,
)))
}
/// Recovers the label decode description a raw frame schema declares.
///
/// Returns `None` for an explicit-label frame, which must not declare one. A key-encoded frame must
/// declare one, so a missing description is an error rather than a reason to reuse a caller-side
/// binding. A present description is parsed strictly: an unknown version, a missing or unknown
/// field at this structure's own levels, a type mismatch, or a description whose logical tags do
/// not name listed key columns is an error, never a silent default. Only the description's own
/// levels reject unknown fields; [`LabelDecodeColumn::physical_type`] is a
/// [`ConcreteDataType`] and accepts whatever that type's own serde accepts.
pub fn raw_series_frame_label_decode(schema: &Schema) -> Result<Option<LabelDecodeSpec>> {
let mode = match schema.metadata().get(LABELS).map(String::as_str) {
Some("key") => RawLabelMode::Key,
Some("explicit") => RawLabelMode::Explicit,
_ => return Err(err("raw frame schema declares no known label mode")),
};
let decode = match schema.metadata().get(LABEL_DECODE) {
None => None,
Some(encoded) => {
let decode: LabelDecodeSpec = serde_json::from_str(encoded)
.map_err(|e| err(format!("malformed label decode description: {e}")))?;
decode.validate()?;
Some(decode)
}
};
match (mode, &decode) {
(RawLabelMode::Key, None) => Err(err(
"key-encoded raw frame declares no label decode description",
)),
(RawLabelMode::Explicit, Some(_)) => Err(err(
"explicit raw frame declares a label decode description",
)),
(RawLabelMode::Key, Some(_)) | (RawLabelMode::Explicit, None) => Ok(decode),
}
}
/// Reconstructs the series schema a raw frame declares.
///
/// A raw frame names its payload fields by name only: every member of the `fields` struct carries
/// the same payload type ([`raw_series_frame_schema`]), and [`SeriesSchema`] holds nothing else per
/// field. The reconstruction is therefore exact rather than a projection — the declared names in
/// declared order, and no evaluation grid, which a raw frame cannot carry.
///
/// The reserved layout, content, and time metadata are checked here together with the absence of an
/// evaluation grid, so a frame this revision cannot read is rejected instead of being silently
/// interpreted as the current layout.
pub fn raw_series_frame_series_schema(schema: &Schema) -> Result<SeriesSchema> {
for (key, expected) in [(LAYOUT, "v1"), (CONTENT, "raw"), (TIME, "timestamps")] {
if schema.metadata().get(key).map(String::as_str) != Some(expected) {
return Err(err(format!("raw frame declares no known {key} value")));
}
}
if schema
.metadata()
.keys()
.any(|key| key == GRID || key.starts_with("greptime.series.grid."))
{
return Err(err("raw series schema cannot have an evaluation grid"));
}
let fields = schema
.fields()
.iter()
.find(|field| field.name() == "fields")
.ok_or_else(|| err("raw frame declares no fields column"))?;
let DataType::Struct(members) = fields.data_type() else {
return Err(err("raw frame fields column is not a struct"));
};
if members.is_empty() {
return Err(err("raw series schema must have fields"));
}
Ok(SeriesSchema {
fields: members
.iter()
.map(|member| FieldSpec {
name: Arc::from(member.name().as_str()),
})
.collect(),
eval_grid: None,
})
}
/// Bounded multi-row builder for finite raw frame batches.
///
/// Rows accumulate until the row limit or the estimated byte limit is reached; the next
@@ -871,7 +1195,7 @@ fn decode_raw_series_frame(
if actual
.metadata()
.keys()
.any(|key| key == "greptime.series.grid" || key.starts_with("greptime.series.grid."))
.any(|key| key == GRID || key.starts_with("greptime.series.grid."))
|| actual.fields() != expected.fields()
{
return invalid_series("raw frame schema mismatch");
+527
View File
@@ -3058,3 +3058,530 @@ fn decoder_reuses_cached_schema_and_rejects_schema_evolution() {
assert!(explicit_decoder.decode(&first).is_err());
assert!(key_decoder.decode(&nameless).is_err());
}
/// A dense description carrying the ordered key columns and the logical tags the decoder reads.
fn dense_description() -> LabelDecodeSpec {
LabelDecodeSpec {
version: LABEL_DECODE_VERSION,
metric_name: "cpu_usage".to_string(),
encoding: LabelDecodeEncoding::Dense,
primary_key: vec![
LabelDecodeColumn {
column_id: 1,
name: "host".to_string(),
physical_type: crate::data_type::ConcreteDataType::string_datatype(),
},
LabelDecodeColumn {
column_id: 2,
name: "region".to_string(),
physical_type: crate::data_type::ConcreteDataType::string_datatype(),
},
],
logical_tags: vec![LabelDecodeTag {
column_id: 1,
name: "host".to_string(),
}],
}
}
#[test]
fn label_decode_description_roundtrips_through_schema_metadata() {
let description = dense_description();
let key = raw_series_frame_schema_with_label_decode(
&test_schema(&["value"]),
RawLabelMode::Key,
Some(&description),
)
.unwrap();
assert_eq!(
raw_series_frame_label_decode(&key).unwrap(),
Some(description.clone())
);
// The description is metadata-only: the column contract is untouched.
assert_eq!(
key.fields(),
raw_series_frame_schema(&test_schema(&["value"]), RawLabelMode::Key)
.unwrap()
.fields()
);
assert_eq!(key.metadata().get(LABELS).map(String::as_str), Some("key"));
}
#[test]
fn explicit_frames_declare_no_label_decode_description() {
let schema = test_schema(&["value"]);
let explicit =
raw_series_frame_schema_with_label_decode(&schema, RawLabelMode::Explicit, None).unwrap();
assert!(explicit.metadata().get(LABEL_DECODE).is_none());
assert_eq!(raw_series_frame_label_decode(&explicit).unwrap(), None);
// A key frame without a description, or an explicit frame with one, is rejected instead of
// being written as a frame the consumer cannot decode.
assert!(
raw_series_frame_schema_with_label_decode(&schema, RawLabelMode::Key, None).is_err(),
"key-encoded frames must describe their labels"
);
assert!(
raw_series_frame_schema_with_label_decode(
&schema,
RawLabelMode::Explicit,
Some(&dense_description())
)
.is_err(),
"explicit frames carry complete labels"
);
}
/// The raw frame's own columns with exactly the supplied metadata map.
fn schema_with_metadata(metadata: Vec<(String, String)>) -> Schema {
let layout = raw_series_frame_schema(&test_schema(&["value"]), RawLabelMode::Key).unwrap();
Schema::new_with_metadata(layout.fields().clone(), metadata.into_iter().collect())
}
/// The four metadata entries every raw frame declares, with the given label mode.
fn base_metadata(labels: &str) -> Vec<(String, String)> {
[
("greptime.series.layout", "v1"),
("greptime.series.content", "raw"),
("greptime.series.labels", labels),
("greptime.series.time", "timestamps"),
]
.into_iter()
.map(|(key, value)| (key.to_string(), value.to_string()))
.collect()
}
#[test]
fn label_decode_description_rejects_inconsistent_bindings() {
let schema = test_schema(&["value"]);
let mut cases = Vec::new();
let mut duplicate_column = dense_description();
duplicate_column.primary_key[1].column_id = 1;
cases.push(("duplicate primary-key column", duplicate_column));
let mut unbound_tag = dense_description();
unbound_tag.logical_tags[0].column_id = 3;
cases.push(("logical tag outside the primary key", unbound_tag));
let mut duplicate_tag = dense_description();
let tag = duplicate_tag.logical_tags[0].clone();
duplicate_tag.logical_tags.push(tag);
cases.push(("duplicate logical tag", duplicate_tag));
for (label, description) in cases {
// Writing is faithful: the description records what the producer declares. Reading is
// where a description this revision cannot honour is rejected, so a producer cannot turn a
// bad binding into a frame a consumer would decode under silently.
let frame = raw_series_frame_schema_with_label_decode(
&schema,
RawLabelMode::Key,
Some(&description),
)
.unwrap_or_else(|error| panic!("{label}: {error}"));
// The writer records what the producer declares, so the rejection has to come from the
// reader; one assertion covers it, since the frame it reads is the text just written.
assert!(
raw_series_frame_label_decode(&frame).is_err(),
"{label}: an inconsistent description is rejected when read"
);
}
}
/// A frame declaring no label mode has no label layout to describe. The rest of the parse and
/// validation rejections (unknown version, missing/unknown/wrongly typed fields, malformed JSON,
/// and mode and description disagreeing in either direction) are pinned case for case by
/// [`raw_series_label_decode_metadata_fixture`], so they are not re-asserted here.
#[test]
fn label_decode_rejects_a_frame_without_a_label_mode() {
let without_mode = base_metadata("key")
.into_iter()
.filter(|(key, _)| key != LABELS)
.collect();
assert!(
raw_series_frame_label_decode(&schema_with_metadata(without_mode)).is_err(),
"a frame without a label mode has no label layout to describe"
);
}
#[test]
fn raw_series_label_decode_metadata_fixture() {
use std::collections::HashMap;
#[derive(serde::Deserialize)]
struct Case {
name: String,
metadata: HashMap<String, String>,
/// The encoding a decoding consumer expects, or `null` when the fixture must be rejected.
encoding: Option<String>,
}
let cases: Vec<Case> = serde_json::from_str(include_str!(
"../../../../../tests/compatibility/wire/raw_series_label_decode/v1_metadata.json"
))
.unwrap();
for case in cases {
let layout = raw_series_frame_schema(&test_schema(&["value"]), RawLabelMode::Key).unwrap();
let metadata = case
.metadata
.iter()
.map(|(key, value)| (key.clone(), value.clone()))
.collect();
let frame_schema = Schema::new_with_metadata(layout.fields().clone(), metadata);
match case.encoding.as_deref() {
None => assert!(
raw_series_frame_label_decode(&frame_schema).is_err(),
"{}: must be rejected",
case.name
),
Some(expected) => {
let decode = raw_series_frame_label_decode(&frame_schema)
.unwrap_or_else(|error| panic!("{}: {error}", case.name))
.unwrap_or_else(|| panic!("{}: no description", case.name));
let encoding = match decode.encoding {
LabelDecodeEncoding::Dense => "dense",
LabelDecodeEncoding::Sparse => "sparse",
};
assert_eq!(encoding, expected, "{}", case.name);
assert_eq!(decode.version, LABEL_DECODE_VERSION, "{}", case.name);
assert_eq!(decode.metric_name, "cpu_usage", "{}", case.name);
assert_eq!(
decode.primary_key,
vec![
LabelDecodeColumn {
column_id: 1,
name: "host".to_string(),
physical_type: crate::data_type::ConcreteDataType::string_datatype(),
},
LabelDecodeColumn {
column_id: 2,
name: "shard".to_string(),
physical_type: crate::data_type::ConcreteDataType::uint32_datatype(),
},
],
"{}: the ordered key columns are decoded in place",
case.name
);
assert_eq!(
decode.logical_tags,
vec![
LabelDecodeTag {
column_id: 1,
name: "host".to_string(),
},
LabelDecodeTag {
column_id: 2,
name: "shard".to_string(),
},
],
"{}",
case.name
);
if expected == "dense" {
// The fixture text is the encoder's own output, byte for byte: re-encoding the
// description it decodes to must reproduce it exactly.
let roundtrip = raw_series_frame_schema_with_label_decode(
&test_schema(&["value"]),
RawLabelMode::Key,
Some(&decode),
)
.unwrap();
assert_eq!(
roundtrip.metadata().get(LABEL_DECODE),
case.metadata.get(LABEL_DECODE),
"{}: re-encoding reproduces the fixture text",
case.name
);
}
}
}
}
}
/// A tag column typed as a timestamp is described like any other key column: the description
/// carries its physical type, and the text is exactly what the encoder wrote.
#[test]
fn label_decode_description_carries_a_timestamp_typed_tag_column() {
let mut description = dense_description();
description.primary_key[1].name = "observed".to_string();
description.primary_key[1].physical_type =
crate::data_type::ConcreteDataType::timestamp_millisecond_datatype();
description.logical_tags.push(LabelDecodeTag {
column_id: 2,
name: "observed".to_string(),
});
let frame = raw_series_frame_schema_with_label_decode(
&test_schema(&["value"]),
RawLabelMode::Key,
Some(&description),
)
.unwrap();
let text = frame.metadata().get(LABEL_DECODE).unwrap();
// The expected text is the encoder's real output for a timestamp-typed tag column.
assert!(
text.contains("\"physical_type\":{\"Timestamp\":{\"Millisecond\":null}}"),
"the encoder wrote a timestamp type: {text}"
);
let recovered = raw_series_frame_label_decode(&frame).unwrap().unwrap();
assert_eq!(recovered, description);
assert_eq!(
recovered.primary_key[1].physical_type,
crate::data_type::ConcreteDataType::timestamp_millisecond_datatype()
);
}
/// Two producers that list the same logical tags in a different order write the same bytes.
///
/// The tag order carries no meaning to a decoder, so the writer sorts its own copy before
/// serializing and leaves the caller's value alone. The encoding order of `primary_key` is a
/// different thing entirely: it is the order of the key columns, so it is written as given.
#[test]
fn label_decode_text_is_independent_of_the_callers_tag_order() {
let mut ascending = dense_description();
ascending.logical_tags = vec![
LabelDecodeTag {
column_id: 1,
name: "host".to_string(),
},
LabelDecodeTag {
column_id: 2,
name: "region".to_string(),
},
];
let mut descending = ascending.clone();
descending.logical_tags.reverse();
// The encoding order of the key columns is not the tag order: reverse it in the second
// description so a writer that sorted `primary_key` too would be caught below.
descending.primary_key.reverse();
let write = |description: &LabelDecodeSpec| {
let frame = raw_series_frame_schema_with_label_decode(
&test_schema(&["value"]),
RawLabelMode::Key,
Some(description),
)
.unwrap();
frame.metadata().get(LABEL_DECODE).unwrap().clone()
};
let same_tags = {
let mut reordered = ascending.clone();
reordered.logical_tags.reverse();
write(&reordered)
};
assert_eq!(
write(&ascending),
same_tags,
"two tag orders wrote different bytes"
);
assert_eq!(
ascending.logical_tags,
vec![
LabelDecodeTag {
column_id: 1,
name: "host".to_string(),
},
LabelDecodeTag {
column_id: 2,
name: "region".to_string(),
},
],
"the writer reordered the caller's description"
);
// The key columns keep the order they were given, so the two descriptions are different
// frames rather than the same bytes: that is the part canonicalization must not touch.
assert_ne!(
write(&ascending),
write(&descending),
"the writer reordered the encoding columns"
);
let recovered = raw_series_frame_label_decode(
&raw_series_frame_schema_with_label_decode(
&test_schema(&["value"]),
RawLabelMode::Key,
Some(&descending),
)
.unwrap(),
)
.unwrap()
.unwrap();
assert_eq!(
recovered.primary_key, descending.primary_key,
"the recovered key order is not the encoding order"
);
}
/// One entry of `v1_physical_types.json`: a [`ConcreteDataType`] variant and the exact text this
/// revision's serde writes for a representative value of it.
#[derive(serde::Deserialize)]
struct FrozenPhysicalType {
/// The [`ConcreteDataType`] variant, as [`concrete_datatype_variants`] spells it.
variant: String,
/// The `serde_json` text this revision writes for that variant's representative value.
text: String,
}
/// Spells every [`ConcreteDataType`] variant exactly once, together with one representative value of
/// it.
///
/// Everything that has to notice a new variant is derived from this one list:
///
/// - [`variant_name`] matches on it with **no wildcard arm**, so adding a variant to
/// [`ConcreteDataType`] is a compile error until the variant is added here.
/// - [`datatype_representatives`] yields one value per listed variant, so the same edit is what
/// supplies the value the tests read the serde text from.
/// - [`VARIANT_NAMES`] is generated from it too, which is what
/// [`the_frozen_text_covers_every_datatype_variant`] requires a fixture entry for.
macro_rules! concrete_datatype_variants {
($($variant:ident => $value:expr),+ $(,)?) => {
/// The [`ConcreteDataType`] variant a value belongs to, spelled as the fixture names it.
///
/// Exhaustive on purpose: a new variant of [`ConcreteDataType`] makes this match fail to
/// compile, which is what makes a missing fixture entry impossible to land.
fn variant_name(data_type: &ConcreteDataType) -> &'static str {
match data_type {
$(ConcreteDataType::$variant(_) => stringify!($variant),)+
}
}
/// One representative value per variant, in the order listed above.
fn datatype_representatives() -> Vec<ConcreteDataType> {
vec![$($value),+]
}
/// Every [`ConcreteDataType`] variant name, in the order listed above.
const VARIANT_NAMES: &[&str] = &[$(stringify!($variant)),+];
};
}
concrete_datatype_variants!(
Null => ConcreteDataType::null_datatype(),
Boolean => ConcreteDataType::boolean_datatype(),
Int8 => ConcreteDataType::int8_datatype(),
Int16 => ConcreteDataType::int16_datatype(),
Int32 => ConcreteDataType::int32_datatype(),
Int64 => ConcreteDataType::int64_datatype(),
UInt8 => ConcreteDataType::uint8_datatype(),
UInt16 => ConcreteDataType::uint16_datatype(),
UInt32 => ConcreteDataType::uint32_datatype(),
UInt64 => ConcreteDataType::uint64_datatype(),
Float32 => ConcreteDataType::float32_datatype(),
Float64 => ConcreteDataType::float64_datatype(),
Decimal128 => ConcreteDataType::decimal128_datatype(10, 2),
Binary => ConcreteDataType::binary_datatype(),
String => ConcreteDataType::string_datatype(),
Date => ConcreteDataType::date_datatype(),
Timestamp => ConcreteDataType::timestamp_millisecond_datatype(),
Time => ConcreteDataType::time_microsecond_datatype(),
Duration => ConcreteDataType::duration_nanosecond_datatype(),
Interval => ConcreteDataType::interval_month_day_nano_datatype(),
List => ConcreteDataType::list_datatype(Arc::new(ConcreteDataType::string_datatype())),
Dictionary => ConcreteDataType::dictionary_datatype(
ConcreteDataType::int32_datatype(),
ConcreteDataType::string_datatype(),
),
Struct => ConcreteDataType::struct_datatype(crate::types::StructType::new(Arc::new(vec![
crate::types::StructField::new("host", ConcreteDataType::string_datatype(), false),
crate::types::StructField::new("port", ConcreteDataType::uint16_datatype(), true),
]))),
Json => ConcreteDataType::json_datatype(),
Vector => ConcreteDataType::vector_datatype(3),
);
/// The frozen text of every [`ConcreteDataType`] variant, as a label decode description embeds it.
const FROZEN_PHYSICAL_TYPES: &str = include_str!(
"../../../../../tests/compatibility/wire/raw_series_label_decode/v1_physical_types.json"
);
fn frozen_physical_types() -> Vec<FrozenPhysicalType> {
serde_json::from_str(FROZEN_PHYSICAL_TYPES).expect("the frozen physical-type fixture parses")
}
/// The representative value of one variant, named the way the fixture names it.
fn representative(variant: &str) -> ConcreteDataType {
VARIANT_NAMES
.iter()
.zip(datatype_representatives())
.find(|(name, _)| **name == variant)
.unwrap_or_else(|| panic!("{variant}: no representative value is listed for this variant"))
.1
}
/// The text a description embeds for a key column's type is not this module's to spell: it is
/// [`ConcreteDataType`]'s own serde, so it changes whenever that type's serde does.
///
/// `v1_physical_types.json` freezes that text for every variant. This test fails when a variant no
/// longer parses back from its frozen text, when it no longer re-serializes to it, or when the
/// parsed value is no longer the type that variant names — which is what a silent serde change
/// looks like. Since the frame text is compared byte for byte across nodes, such a change is a
/// [`LABEL_DECODE_VERSION`] bump, and this test is the alarm that raises it.
#[test]
fn every_datatype_round_trips_through_its_frozen_text() {
use crate::data_type::DataType as _;
for entry in frozen_physical_types() {
let parsed: ConcreteDataType = serde_json::from_str(&entry.text).unwrap_or_else(|error| {
panic!("{}: {} cannot be read: {error}", entry.variant, entry.text)
});
assert_eq!(
serde_json::to_string(&parsed).unwrap(),
entry.text,
"{}: re-serializes to the frozen text",
entry.variant
);
assert_eq!(
parsed.name(),
representative(&entry.variant).name(),
"{}: the frozen text still names the same type",
entry.variant
);
assert_eq!(
parsed,
representative(&entry.variant),
"{}: the frozen text still describes the same value",
entry.variant
);
}
}
/// The fixture above only freezes a variant if it has an entry, so every variant must have one.
///
/// This test is what makes that hold: [`variant_name`] matches every [`ConcreteDataType`] variant
/// with no wildcard arm, so a new variant does not compile until it is listed in
/// [`concrete_datatype_variants`]; that listing also extends [`VARIANT_NAMES`] and
/// [`datatype_representatives`], and this test then requires a fixture entry for it and rejects an
/// entry naming a variant that does not exist.
#[test]
fn the_frozen_text_covers_every_datatype_variant() {
let frozen = frozen_physical_types();
for data_type in datatype_representatives() {
let variant = variant_name(&data_type);
let entry = frozen
.iter()
.find(|entry| entry.variant == variant)
.unwrap_or_else(|| {
panic!("{variant}: v1_physical_types.json has no entry for this datatype variant")
});
// The entry's text is this variant's text: read it back and name the variant it describes.
let parsed: ConcreteDataType = serde_json::from_str(&entry.text)
.unwrap_or_else(|error| panic!("{variant}: {} is unreadable: {error}", entry.text));
assert_eq!(
variant_name(&parsed),
variant,
"{variant}: the entry's text describes this variant"
);
}
// Nothing in the fixture names a variant that does not exist, and no variant is listed twice.
for entry in &frozen {
assert!(
VARIANT_NAMES.contains(&entry.variant.as_str()),
"{}: the fixture names a variant ConcreteDataType does not have",
entry.variant
);
}
assert_eq!(
frozen.len(),
VARIANT_NAMES.len(),
"one fixture entry per datatype variant, no duplicates"
);
}
+10
View File
@@ -81,6 +81,15 @@ pub enum Error {
location: Location,
},
#[snafu(display("Invalid label decode description"))]
InvalidLabelDecodeDescription {
// Box the source to reduce the size of the error.
#[snafu(source(from(datatypes::error::Error, Box::new)))]
source: Box<datatypes::error::Error>,
#[snafu(implicit)]
location: Location,
},
#[snafu(display("Encode null value"))]
IndexEncodeNull {
#[snafu(implicit)]
@@ -108,6 +117,7 @@ impl ErrorExt for Error {
StatusCode::InvalidArguments
}
NotSupportedField { .. } | UnsupportedOperation { .. } => StatusCode::Unsupported,
InvalidLabelDecodeDescription { source, .. } => source.status_code(),
InvalidSparsePrimaryKey { .. } | InvalidDensePrimaryKey { .. } => {
StatusCode::InvalidArguments
}
+156 -43
View File
@@ -17,11 +17,12 @@ use std::sync::Arc;
use api::v1::SemanticType;
use common_telemetry::debug;
use datatypes::data_type::ConcreteDataType;
use datatypes::series::frame::{LabelDecodeEncoding, LabelDecodeSpec};
use datatypes::series::{Label, MetricName};
use datatypes::value::Value;
use snafu::OptionExt;
use snafu::{OptionExt, ResultExt};
use store_api::codec::PrimaryKeyEncoding;
use store_api::metadata::RegionMetadataRef;
use store_api::metric_engine_consts::is_metric_engine_internal_column;
use store_api::region_engine::RawLabelSource;
use store_api::storage::{
@@ -30,8 +31,10 @@ use store_api::storage::{
};
use crate::error::{
InvalidDensePrimaryKeySnafu, InvalidSparsePrimaryKeySnafu, Result, UnsupportedOperationSnafu,
InvalidDensePrimaryKeySnafu, InvalidLabelDecodeDescriptionSnafu, InvalidSparsePrimaryKeySnafu,
Result, UnsupportedOperationSnafu,
};
use crate::row_converter::SortField;
use crate::row_converter::dense::DensePrimaryKeyCodec;
use crate::row_converter::sparse::{SparseOffsetsCache, SparsePrimaryKeyView};
@@ -68,10 +71,41 @@ struct BoundLabel {
position: usize,
}
/// Decodes labels on demand using the metadata snapshot bound to a raw-label source.
pub struct PrimaryKeyLabelDecoder {
/// One ordered key column of the minimal decode layout.
struct KeyColumn {
id: ColumnId,
name: String,
data_type: ConcreteDataType,
}
/// Minimal primary-key layout labels can be decoded from.
///
/// It holds only what decoding reads: the metric name, the primary-key encoding, and the ordered
/// primary-key columns with their physical types. A scan-bound [`RawLabelSource::PrimaryKey`] and a
/// frame schema's [`LabelDecodeSpec`] both reduce to this view, so both bind the same decoder.
/// Every key column is a `Tag` column, which is checked when a decoder is bound, so the layout
/// carries no per-column role.
struct KeyLayout {
metric_name: Arc<str>,
physical_metadata: RegionMetadataRef,
encoding: PrimaryKeyEncoding,
columns: Vec<KeyColumn>,
}
impl KeyLayout {
/// Finds a listed primary-key column.
fn column(&self, id: ColumnId) -> Option<&KeyColumn> {
self.columns.iter().find(|column| column.id == id)
}
/// Returns the position of a listed primary-key column.
fn position(&self, id: ColumnId) -> Option<usize> {
self.columns.iter().position(|column| column.id == id)
}
}
/// Decodes labels on demand using the minimal primary-key layout bound to a raw-label source.
pub struct PrimaryKeyLabelDecoder {
key: KeyLayout,
complete: bool,
include_name: bool,
labels: Vec<BoundLabel>,
@@ -85,6 +119,9 @@ impl PrimaryKeyLabelDecoder {
/// Explicit sources already contain complete labels and are rejected. Dense decoding
/// requires the full key layout described by the bound snapshot; compatibility with
/// older key layouts must be normalized upstream.
///
/// Every primary-key column must be a `Tag` column; [`RegionMetadata::validate`] enforces it, so
/// this only rejects a snapshot built without validation.
pub fn new(source: &RawLabelSource, projection: LabelProjection) -> Result<Self> {
let RawLabelSource::PrimaryKey {
metric_name,
@@ -98,36 +135,113 @@ impl PrimaryKeyLabelDecoder {
.fail();
};
let metadata = physical_metadata.as_ref();
let mut labels = Vec::with_capacity(logical_tags.len());
if let Some(column) = metadata
.primary_key_columns()
.find(|column| column.semantic_type != SemanticType::Tag)
{
return InvalidSparsePrimaryKeySnafu {
reason: format!(
"primary-key column {} is not a tag column",
column.column_schema.name
),
}
.fail();
}
let key = KeyLayout {
metric_name: metric_name.clone(),
encoding: metadata.primary_key_encoding,
columns: metadata
.primary_key_columns()
.map(|column| KeyColumn {
id: column.column_id,
name: column.column_schema.name.clone(),
data_type: column.column_schema.data_type.clone(),
})
.collect(),
};
Self::bind(
key,
logical_tags
.iter()
.map(|tag| (tag.column_id, tag.name.as_ref())),
projection,
)
}
/// Binds a decoder to a frame schema's label decode description and projection.
///
/// This is the decoding path for a consumer that received only the raw frame schema: the
/// description carries the metric name, the primary-key encoding, the ordered key columns with
/// their physical types, and the logical tags. The description is validated first, so a version
/// or structure this revision cannot honour is an error instead of a best-effort decode.
///
/// The description lists exactly the encoded key fields, and region metadata guarantees each
/// of them is a tag column, so no per-column role is carried or re-checked; an array position
/// is an encoding position. The logical tags are still checked here, one by one, exactly as
/// for a typed binding, which is the only place the decoder inspects them (a listed column no
/// tag selects is not examined).
pub fn from_decode_spec(spec: &LabelDecodeSpec, projection: LabelProjection) -> Result<Self> {
spec.validate()
.context(InvalidLabelDecodeDescriptionSnafu)?;
let key = KeyLayout {
metric_name: Arc::from(spec.metric_name.as_str()),
encoding: match spec.encoding {
LabelDecodeEncoding::Dense => PrimaryKeyEncoding::Dense,
LabelDecodeEncoding::Sparse => PrimaryKeyEncoding::Sparse,
},
columns: spec
.primary_key
.iter()
.map(|column| KeyColumn {
id: column.column_id,
name: column.name.clone(),
data_type: column.physical_type.clone(),
})
.collect(),
};
Self::bind(
key,
spec.logical_tags
.iter()
.map(|tag| (tag.column_id, tag.name.as_str())),
projection,
)
}
/// Binds the ordered logical tags of `key` to one projection and one decode layout.
fn bind<'a, I>(key: KeyLayout, tags: I, projection: LabelProjection) -> Result<Self>
where
I: IntoIterator<Item = (ColumnId, &'a str)>,
{
let mut labels = Vec::new();
let mut names = HashSet::new();
let mut ids = HashSet::new();
for tag in logical_tags.iter() {
let column = metadata.column_by_id(tag.column_id);
if !names.insert(tag.name.clone())
|| !ids.insert(tag.column_id)
|| tag.name.as_ref() == "__name__"
|| is_internal_column(&tag.name)
|| is_metric_engine_internal_column(&tag.name)
|| ReservedColumnId::is_reserved(tag.column_id)
|| !column.is_some_and(|column| {
column.semantic_type == SemanticType::Tag
&& column.column_schema.name.as_str() == tag.name.as_ref()
&& metadata.primary_key.contains(&tag.column_id)
})
let mut logical_tag_count = 0;
for (column_id, name) in tags {
logical_tag_count += 1;
if !names.insert(name)
|| !ids.insert(column_id)
|| name == "__name__"
|| is_internal_column(name)
|| is_metric_engine_internal_column(name)
|| ReservedColumnId::is_reserved(column_id)
|| !key
.column(column_id)
.is_some_and(|column| column.name == name)
{
return InvalidSparsePrimaryKeySnafu {
reason: format!("invalid logical tag binding {}", tag.name),
reason: format!("invalid logical tag binding {name}"),
}
.fail();
}
let position = metadata.primary_key_index(tag.column_id).context(
InvalidSparsePrimaryKeySnafu {
reason: format!("logical tag {} is not a primary-key column", tag.name),
},
)?;
let position = key
.position(column_id)
.context(InvalidSparsePrimaryKeySnafu {
reason: format!("logical tag {name} is not a primary-key column"),
})?;
labels.push(BoundLabel {
id: tag.column_id,
name: tag.name.clone(),
id: column_id,
name: Arc::from(name),
position,
});
}
@@ -143,14 +257,19 @@ impl PrimaryKeyLabelDecoder {
let requested: HashSet<_> = requested.iter().map(AsRef::as_ref).collect();
labels.retain(|label| requested.contains(label.name.as_ref()));
}
let layout = match metadata.primary_key_encoding {
let layout = match key.encoding {
PrimaryKeyEncoding::Sparse => Layout::Sparse(SparseOffsetsCache::new()),
PrimaryKeyEncoding::Dense => Layout::Dense {
codec: DensePrimaryKeyCodec::new(metadata),
codec: DensePrimaryKeyCodec::with_fields(
key.columns
.iter()
.map(|column| (column.id, SortField::new(column.data_type.clone())))
.collect(),
),
offsets: Vec::new(),
},
};
let encoding = match metadata.primary_key_encoding {
let encoding = match key.encoding {
PrimaryKeyEncoding::Sparse => "sparse",
PrimaryKeyEncoding::Dense => "dense",
};
@@ -158,13 +277,12 @@ impl PrimaryKeyLabelDecoder {
debug!(
encoding,
projection_mode,
logical_tag_count = logical_tags.len(),
logical_tag_count,
selected_tag_count = labels.len(),
"Constructed primary key label decoder"
);
Ok(Self {
metric_name: metric_name.clone(),
physical_metadata: physical_metadata.clone(),
key,
complete,
include_name,
labels,
@@ -182,8 +300,8 @@ impl PrimaryKeyLabelDecoder {
let mut values = Vec::with_capacity(self.labels.len() + usize::from(self.include_name));
let Self {
layout,
key,
labels,
physical_metadata,
complete,
value_scratch,
..
@@ -192,12 +310,7 @@ impl PrimaryKeyLabelDecoder {
Layout::Sparse(cache) => {
let mut view = SparsePrimaryKeyView::new(read_key, cache)?;
view.visit_label_ids(|id| {
if *complete
&& !physical_metadata.column_by_id(id).is_some_and(|column| {
column.semantic_type == SemanticType::Tag
&& physical_metadata.primary_key.contains(&id)
})
{
if *complete && key.column(id).is_none() {
return InvalidSparsePrimaryKeySnafu {
reason: format!("unrecognized sparse tag id {id}"),
}
@@ -240,7 +353,7 @@ impl PrimaryKeyLabelDecoder {
}
if *complete {
Ok(DecodedLabels::Complete(
MetricName::try_new(Some(self.metric_name.clone()), values).map_err(|error| {
MetricName::try_new(Some(key.metric_name.clone()), values).map_err(|error| {
InvalidSparsePrimaryKeySnafu {
reason: error.to_string(),
}
@@ -251,7 +364,7 @@ impl PrimaryKeyLabelDecoder {
if self.include_name {
values.push(Label {
name: Arc::from("__name__"),
value: self.metric_name.clone(),
value: self.key.metric_name.clone(),
});
values.sort_unstable_by(|a, b| a.name.cmp(&b.name));
}
+111 -2
View File
@@ -209,7 +209,7 @@ fn dense_and_sparse_decode_complete_or_projected_labels() {
}
#[test]
fn projection_is_bound_to_snapshot_and_distinguishes_missing_and_empty() {
fn projection_snapshots_its_layout_and_distinguishes_missing_and_empty() {
let original = metadata(
&[
(21, "old", ConcreteDataType::string_datatype()),
@@ -250,7 +250,10 @@ fn projection_is_bound_to_snapshot_and_distinguishes_missing_and_empty() {
let mut all = PrimaryKeyLabelDecoder::new(&old_source, LabelProjection::All).unwrap();
drop(old_source);
drop(original);
assert!(weak_metadata.upgrade().is_some());
assert!(
weak_metadata.upgrade().is_none(),
"the decoder keeps its own key layout (ids, positions, types), not the region metadata"
);
assert_eq!(
decoder.decode(&key).unwrap(),
DecodedLabels::Projected(vec![Label {
@@ -601,3 +604,109 @@ fn dense_typed_values_keep_exact_integer_string_representation() {
"9007199254740993"
);
}
/// One field is enough: the frame schema only carries ordered payload fields and metadata.
fn frame_series_schema() -> Arc<datatypes::series::samples::SeriesSchema> {
Arc::new(datatypes::series::samples::SeriesSchema {
fields: vec![datatypes::series::samples::FieldSpec {
name: Arc::from("value"),
}]
.into(),
eval_grid: None,
})
}
/// A key-encoded frame's schema describes its own labels: recovering the description from the
/// schema and decoding with it matches decoding with the original bound source, for both primary
/// key encodings and both projections.
#[test]
fn frame_label_decode_description_decodes_like_the_bound_source() {
for encoding in [PrimaryKeyEncoding::Dense, PrimaryKeyEncoding::Sparse] {
let bound = source(
string_metadata(encoding),
"requests_total",
&[(8, "zeta"), (41, "alpha")],
);
let description = common_query::series_frame::label_decode_spec(&bound)
.expect("a primary-key binding describes its key layout");
let frame_schema = datatypes::series::frame::raw_series_frame_schema_with_label_decode(
&frame_series_schema(),
datatypes::series::frame::RawLabelMode::Key,
Some(&description),
)
.unwrap();
let recovered = datatypes::series::frame::raw_series_frame_label_decode(&frame_schema)
.unwrap()
.expect("the frame declares its decode description");
assert_eq!(recovered, description);
let key = encoded(encoding, &values());
for projection in [
LabelProjection::All,
LabelProjection::Only(Arc::from([Arc::<str>::from("alpha")])),
LabelProjection::Only(Arc::from([])),
] {
let mut direct = PrimaryKeyLabelDecoder::new(&bound, projection.clone()).unwrap();
let mut from_frame =
PrimaryKeyLabelDecoder::from_decode_spec(&recovered, projection).unwrap();
assert_eq!(
from_frame.decode(&key).unwrap(),
direct.decode(&key).unwrap()
);
}
}
}
/// A description this revision cannot honour is rejected when a decoder is bound, not decoded
/// best-effort.
#[test]
fn frame_label_decode_description_rejects_unsupported_version() {
let bound = source(
string_metadata(PrimaryKeyEncoding::Dense),
"requests_total",
&[(8, "zeta")],
);
let description = common_query::series_frame::label_decode_spec(&bound).unwrap();
let mut future = description.clone();
future.version += 1;
assert!(PrimaryKeyLabelDecoder::from_decode_spec(&future, LabelProjection::All).is_err());
// The description must list the column a logical tag reads.
let mut unbound = description;
unbound.primary_key.retain(|column| column.column_id != 8);
assert!(PrimaryKeyLabelDecoder::from_decode_spec(&unbound, LabelProjection::All).is_err());
}
/// A primary-key column that is not a `Tag` column is rejected when a decoder is bound, because the
/// layout records no per-column role and decodes every listed key column as a tag.
///
/// `RegionMetadata::validate` rejects such metadata, so this snapshot is built without validation to
/// exercise the check the decoder performs itself.
#[test]
fn binding_rejects_a_primary_key_column_that_is_not_a_tag() {
let mut builder = RegionMetadataBuilder::new(RegionId::new(7, 1));
builder.push_column_metadata(ColumnMetadata {
column_schema: ColumnSchema::new(
"ts",
ConcreteDataType::timestamp_millisecond_datatype(),
false,
),
semantic_type: SemanticType::Timestamp,
column_id: 0,
});
builder.push_column_metadata(ColumnMetadata {
column_schema: ColumnSchema::new("not_a_tag", ConcreteDataType::string_datatype(), true),
semantic_type: SemanticType::Field,
column_id: 8,
});
builder.primary_key(vec![8]);
builder.primary_key_encoding(PrimaryKeyEncoding::Sparse);
let invalid = Arc::new(builder.build_without_validation().unwrap());
let bound = source(invalid, "m", &[(8, "not_a_tag")]);
let error = PrimaryKeyLabelDecoder::new(&bound, LabelProjection::All)
.err()
.expect("a non-tag primary-key column must be rejected")
.to_string();
assert!(error.contains("not a tag column"), "{error}");
}
+226 -25
View File
@@ -5118,7 +5118,7 @@ async fn test_mito_handle_query_raw_series_candidate_projection_and_mvcc() {
};
let native_plan = Arc::new(RegionScanExec::new(native_scanner, request.clone(), None).unwrap());
let native_plan = native_plan
.try_to_series_scan(native_source.clone())
.try_to_series_scan(native_source)
.unwrap()
.expect("Mito raw-series capability");
let native_schema = native_plan.series_schema().clone();
@@ -5133,8 +5133,7 @@ async fn test_mito_handle_query_raw_series_candidate_projection_and_mvcc() {
frame_ordering,
"the legacy producer must declare key-then-time frames"
);
let unpacked =
unpack_raw_series(native_frames, native_schema, native_source, frame_ordering).unwrap();
let unpacked = unpack_raw_series(native_frames, Some(native_schema)).unwrap();
let native_schema = unpacked.schema().clone();
let mut native = unpacked.try_collect::<Vec<_>>().await.unwrap();
assert_eq!(
@@ -5370,14 +5369,14 @@ async fn test_two_phase_native_raw_stream_declares_contiguous_frames() {
let schema = schema.expect("the scan must emit at least one partition stream");
let packed = pack_raw_series(Box::pin(ChainedRawSeriesStream::new(
Arc::clone(&schema),
source.clone(),
source,
SeriesOrdering::Contiguous,
chunks,
)))
.unwrap();
let ordering = raw_series_frame_ordering(packed.schema().as_ref()).unwrap();
assert_eq!(SeriesOrdering::Contiguous, ordering);
let validated = unpack_raw_series(packed, schema, source, ordering)
let validated = unpack_raw_series(packed, Some(schema))
.unwrap()
.try_collect::<Vec<_>>()
.await
@@ -5720,18 +5719,12 @@ async fn test_two_phase_raw_stream_stays_contiguous_across_disjoint_partition_as
// not final.
let packed = pack_raw_series(Box::pin(ChainedRawSeriesStream::new(
Arc::clone(&schema),
source.clone(),
source,
SeriesOrdering::Contiguous,
complete.clone(),
)))
.unwrap();
let unpacked = unpack_raw_series(
packed,
Arc::clone(&schema),
source,
SeriesOrdering::Contiguous,
)
.unwrap();
let unpacked = unpack_raw_series(packed, Some(schema)).unwrap();
let validated = unpacked.try_collect::<Vec<_>>().await.unwrap();
assert_eq!(
complete_rows,
@@ -6064,18 +6057,12 @@ async fn test_two_phase_raw_same_partition_multi_round_assignment_stays_contiguo
// the chunks this reader produced, including the chunks the second assignment added.
let packed = pack_raw_series(Box::pin(ChainedRawSeriesStream::new(
Arc::clone(&schema),
source.clone(),
source,
SeriesOrdering::Contiguous,
chunks.clone(),
)))
.unwrap();
let unpacked = unpack_raw_series(
packed,
Arc::clone(&schema),
source,
SeriesOrdering::Contiguous,
)
.unwrap();
let unpacked = unpack_raw_series(packed, Some(Arc::clone(&schema))).unwrap();
let validated = unpacked.try_collect::<Vec<_>>().await.unwrap();
assert_eq!(
stream_rows,
@@ -6200,6 +6187,8 @@ fn primary_key_label_source(
struct ChainedRawSeriesStream {
schema: Arc<datatypes::series::samples::SeriesSchema>,
label_source: store_api::region_engine::RawLabelSource,
/// The description this stream declares, derived from `label_source` once, like a producer.
label_decode_spec: Option<datatypes::series::frame::LabelDecodeSpec>,
ordering: datatypes::series::SeriesOrdering,
chunks: std::vec::IntoIter<datatypes::series::raw::RawSeriesChunk>,
}
@@ -6211,9 +6200,11 @@ impl ChainedRawSeriesStream {
ordering: datatypes::series::SeriesOrdering,
chunks: Vec<datatypes::series::raw::RawSeriesChunk>,
) -> Self {
let label_decode_spec = common_query::series_frame::label_decode_spec(&label_source);
Self {
schema,
label_source,
label_decode_spec,
ordering,
chunks: chunks.into_iter(),
}
@@ -6240,11 +6231,221 @@ impl store_api::region_engine::RawSeriesStream for ChainedRawSeriesStream {
&self.label_source
}
fn label_decode_spec(&self) -> Option<&datatypes::series::frame::LabelDecodeSpec> {
self.label_decode_spec.as_ref()
}
fn ordering(&self) -> datatypes::series::SeriesOrdering {
self.ordering
}
}
/// Decodes each chunk's primary key through `description`, as a consumer holding only the stream
/// would, and asserts none of the chunks carries inline labels.
fn decode_declared_labels(
description: &datatypes::series::frame::LabelDecodeSpec,
chunks: &[datatypes::series::raw::RawSeriesChunk],
) -> Vec<datatypes::series::MetricName> {
use mito_codec::labels::{DecodedLabels, LabelProjection, PrimaryKeyLabelDecoder};
let mut decoder =
PrimaryKeyLabelDecoder::from_decode_spec(description, LabelProjection::All).unwrap();
chunks
.iter()
.map(|chunk| {
assert!(
chunk.metric.is_none(),
"a key-encoded chunk leaves its labels undecoded"
);
match decoder.decode(chunk.primary_key.as_slice()).unwrap() {
DecodedLabels::Complete(metric) => metric,
DecodedLabels::Projected(_) => {
panic!("decoding every label must yield complete labels")
}
}
})
.collect()
}
/// Every key-encoded raw series stream declares its own label decode description before the first
/// poll, and that declaration is the one derived from the binding the scan read under. Dense and
/// sparse producers are both covered, plus a stream that ends without producing a chunk, so the
/// producer and the consumer are exercised against the same declaration.
#[tokio::test]
async fn test_mito_raw_series_stream_declares_its_label_decode_description() {
use store_api::region_engine::QueryScanContext;
// Dense encoding: one flushed series under the default region layout.
let mut env = TestEnv::with_prefix("test_mito_raw_series_self_describing_dense").await;
let engine = env.create_engine(MitoConfig::default()).await;
let region_id = RegionId::new(1, 1);
let request = CreateRequestBuilder::new().build();
let schema = test_util::rows_schema(&request);
engine
.handle_request(region_id, RegionRequest::Create(request))
.await
.unwrap();
test_util::put_rows(
&engine,
region_id,
Rows {
schema,
rows: test_util::build_rows_for_key("host-a", 0, 2, 0),
},
)
.await;
test_util::flush_region(&engine, region_id, None).await;
let request = ScanRequest {
distribution: Some(TimeSeriesDistribution::PerSeries),
preserve_pk_dictionary_encoding: true,
..Default::default()
};
let mut scanner = engine.handle_query(region_id, request).await.unwrap();
let source = primary_key_label_source(&scanner.metadata());
let expected = common_query::series_frame::label_decode_spec(&source)
.expect("a primary-key binding describes its key");
let ranges = scanner.properties().partitions.clone();
scanner
.prepare(PrepareRequest {
ranges: Some(ranges.clone()),
target_partitions: Some(ranges.len()),
..Default::default()
})
.unwrap();
let metrics = ExecutionPlanMetricsSet::default();
let mut chunks = Vec::new();
for partition in 0..ranges.len() {
let stream = scanner
.scan_series_partition(&QueryScanContext::default(), &metrics, partition, &source)
.unwrap()
.expect("a dense PerSeries scan produces raw series");
// The producer states its layout up front, before anything is polled.
assert_eq!(
Some(&expected),
stream.label_decode_spec(),
"dense partition {partition} must declare its label decode description before polling"
);
chunks.extend(stream.try_collect::<Vec<_>>().await.unwrap());
}
assert_eq!(
2,
chunks.iter().map(|chunk| chunk.time.len()).sum::<usize>()
);
let dense_metrics = decode_declared_labels(&expected, &chunks);
assert_eq!(dense_metrics.len(), chunks.len());
for metric in &dense_metrics {
assert_eq!(metric.name(), Some("fixture_metric"));
assert_eq!(metric.get("tag_0"), Some("host-a"));
}
// A statically empty stream declares the same layout while producing no chunk at all.
let empty_request = ScanRequest {
filters: vec![col("field_0").gt(lit(1.0e9_f64))],
distribution: Some(TimeSeriesDistribution::PerSeries),
preserve_pk_dictionary_encoding: true,
..Default::default()
};
let mut empty_scanner = engine.handle_query(region_id, empty_request).await.unwrap();
let empty_source = primary_key_label_source(&empty_scanner.metadata());
let empty_expected = common_query::series_frame::label_decode_spec(&empty_source)
.expect("a primary-key binding describes its key");
let empty_ranges = empty_scanner.properties().partitions.clone();
assert!(
!empty_ranges.is_empty(),
"a filtered PerSeries scan still assigns its pinned input to a partition"
);
empty_scanner
.prepare(PrepareRequest {
ranges: Some(empty_ranges.clone()),
target_partitions: Some(empty_ranges.len()),
..Default::default()
})
.unwrap();
let mut empty_chunks = Vec::new();
for partition in 0..empty_ranges.len() {
let stream = empty_scanner
.scan_series_partition(
&QueryScanContext::default(),
&metrics,
partition,
&empty_source,
)
.unwrap()
.expect("a statically empty PerSeries scan still produces a raw series stream");
assert_eq!(
Some(&empty_expected),
stream.label_decode_spec(),
"an empty partition {partition} must still declare its layout before polling"
);
empty_chunks.extend(stream.try_collect::<Vec<_>>().await.unwrap());
}
assert!(
empty_chunks.is_empty(),
"the filtered-out stream produced no chunk"
);
// Sparse encoding: the two-phase producer encodes table id and TSID into the key too.
let mut env = TestEnv::with_prefix("test_mito_raw_series_self_describing_sparse").await;
let engine = env
.create_engine(MitoConfig {
experimental_series_scan_v2: true,
..Default::default()
})
.await;
let region_id = RegionId::new(1, 1);
let (metadata, row_schema) = create_two_phase_sparse_region(&engine, region_id).await;
let key = new_sparse_primary_key(&["host-a", "zone-1"], &metadata, 10, 0);
put_sparse_rows(
&engine,
region_id,
&row_schema,
vec![(key.clone(), 10, 1), (key, 20, 2)],
)
.await;
let request = ScanRequest {
distribution: Some(TimeSeriesDistribution::PerSeries),
preserve_pk_dictionary_encoding: true,
..Default::default()
};
let mut scanner = engine.handle_query(region_id, request).await.unwrap();
let source = primary_key_label_source(&scanner.metadata());
let expected = common_query::series_frame::label_decode_spec(&source)
.expect("a primary-key binding describes its key");
let ranges = scanner.properties().partitions.clone();
scanner
.prepare(PrepareRequest {
ranges: Some(ranges.clone()),
target_partitions: Some(ranges.len()),
..Default::default()
})
.unwrap();
let mut chunks = Vec::new();
for partition in 0..ranges.len() {
let stream = scanner
.scan_series_partition(&QueryScanContext::default(), &metrics, partition, &source)
.unwrap()
.expect("a sparse PerSeries scan produces raw series");
assert_eq!(
Some(&expected),
stream.label_decode_spec(),
"sparse partition {partition} must declare its label decode description before polling"
);
chunks.extend(stream.try_collect::<Vec<_>>().await.unwrap());
}
assert_eq!(
2,
chunks.iter().map(|chunk| chunk.time.len()).sum::<usize>()
);
let metrics_seen = decode_declared_labels(&expected, &chunks);
assert_eq!(metrics_seen.len(), chunks.len());
for metric in &metrics_seen {
assert_eq!(metric.name(), Some("fixture_metric"));
assert_eq!(metric.get("tag_0"), Some("host-a"));
assert_eq!(metric.get("tag_1"), Some("zone-1"));
}
}
#[tokio::test]
async fn test_mito_raw_series_int64_predicate_uses_original_precision() {
use datatypes::series::samples::SampleRef;
@@ -6388,7 +6589,7 @@ async fn test_mito_raw_series_int64_predicate_uses_original_precision() {
}]),
};
let native_plan = Arc::new(RegionScanExec::new(native_scanner, native_request, None).unwrap())
.try_to_series_scan(native_source.clone())
.try_to_series_scan(native_source)
.unwrap()
.expect("typed int64 gate is supported");
let native_schema = native_plan.series_schema().clone();
@@ -6402,7 +6603,7 @@ async fn test_mito_raw_series_int64_predicate_uses_original_precision() {
ordering,
"the legacy producer must declare key-then-time frames"
);
let chunks = unpack_raw_series(frames, native_schema, native_source, ordering)
let chunks = unpack_raw_series(frames, Some(native_schema))
.unwrap()
.try_collect::<Vec<_>>()
.await
@@ -6447,7 +6648,7 @@ async fn test_mito_raw_series_int64_predicate_uses_original_precision() {
}]),
};
let empty_plan = Arc::new(RegionScanExec::new(empty_scanner, empty_request, None).unwrap())
.try_to_series_scan(empty_source.clone())
.try_to_series_scan(empty_source)
.unwrap()
.expect("statically empty scan still has native capability");
let empty_schema = empty_plan.series_schema().clone();
@@ -6460,7 +6661,7 @@ async fn test_mito_raw_series_int64_predicate_uses_original_precision() {
empty_ordering,
"a statically empty legacy scan still declares key-then-time frames"
);
let empty_chunks = unpack_raw_series(empty_frames, empty_schema, empty_source, empty_ordering)
let empty_chunks = unpack_raw_series(empty_frames, Some(empty_schema))
.unwrap()
.try_collect::<Vec<_>>()
.await
+16
View File
@@ -25,6 +25,7 @@ use common_query::error::Error as QueryError;
use common_query::native_histogram::{
histogram_samples_from_arrow, is_native_histogram_value_type,
};
use common_query::series_frame::label_decode_spec;
use common_recordbatch::filter;
use datafusion::physical_plan::PhysicalExpr;
use datatypes::arrow::array::{
@@ -37,6 +38,7 @@ use datatypes::arrow::datatypes::DataType;
use datatypes::arrow::record_batch::RecordBatch;
use datatypes::prelude::ConcreteDataType;
use datatypes::series::SeriesOrdering;
use datatypes::series::frame::LabelDecodeSpec;
use datatypes::series::raw::{RawSeriesChunk, RawTimestamps};
use datatypes::series::samples::{Samples, SeriesFields, SeriesSchema};
use futures::stream::BoxStream;
@@ -128,6 +130,13 @@ fn chunk(
struct RawStream {
schema: Arc<SeriesSchema>,
label_source: RawLabelSource,
/// The label decode description this stream declares, derived from `label_source` once.
///
/// Key-encoded chunks leave `metric` as `None`, so a consumer that holds only the stream needs
/// the description to decode them; deriving it here makes the producer self-describing instead
/// of leaving the derivation to whoever packs the chunks. An invalid binding is cached as
/// derived and rejected by the decoder downstream, never silently turned into `None`.
label_decode_spec: Option<LabelDecodeSpec>,
ordering: SeriesOrdering,
inner: BoxStream<'static, Result<RawSeriesChunk, BoxedError>>,
}
@@ -146,6 +155,11 @@ impl RawSeriesStream for RawStream {
fn label_source(&self) -> &RawLabelSource {
&self.label_source
}
/// Returns the description derived from this stream's binding before the first poll; chunks
/// carry key-encoded labels, so the description is what decoding needs.
fn label_decode_spec(&self) -> Option<&LabelDecodeSpec> {
self.label_decode_spec.as_ref()
}
fn ordering(&self) -> SeriesOrdering {
self.ordering
}
@@ -161,6 +175,7 @@ pub(crate) fn make_stream(
ordering: SeriesOrdering,
) -> SendableRawSeriesStream {
let output_schema = schema.clone();
let label_decode_spec = label_decode_spec(&label_source);
let stream = async_stream::stream! {
let mut input = inner;
let mut pending: Option<RawSeriesChunk> = None;
@@ -267,6 +282,7 @@ pub(crate) fn make_stream(
Box::pin(RawStream {
schema,
label_source,
label_decode_spec,
ordering,
inner: Box::pin(stream),
})
@@ -16,7 +16,7 @@ use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use common_query::series_frame::{raw_series_frame_ordering, unpack_raw_series};
use common_query::series_frame::{ensure_frame_matches_label_source, unpack_raw_series};
use datafusion::arrow::array::Array;
use datafusion::arrow::datatypes::SchemaRef;
use datafusion::arrow::record_batch::RecordBatch;
@@ -31,7 +31,6 @@ use datafusion::physical_plan::{
DisplayAs, DisplayFormatType, ExecutionPlan, PhysicalExpr, PlanProperties, RecordBatchStream,
SendableRecordBatchStream,
};
use datatypes::series::SeriesOrdering;
use datatypes::series::raw::{RawSeriesChunk, RawTimestamps};
use datatypes::series::samples::{DenseSamplesRef, Samples, SeriesFields, SeriesSchema};
use futures::{Stream, ready};
@@ -86,16 +85,16 @@ impl InstantManipulate {
return Ok(None);
}
let declared_schema = input.schema();
let ordering = raw_series_frame_ordering(declared_schema.as_ref())?;
// Validate the declared raw-frame schema without executing the child.
// Validate the declared raw-frame schema without executing the child. The frame states its
// own label layout, so the plan's binding is checked against that declaration rather than
// supplied to the decoder.
ensure_frame_matches_label_source(declared_schema.as_ref(), &label_source)?;
unpack_raw_series(
Box::pin(RecordBatchStreamAdapter::new(
declared_schema.clone(),
futures::stream::empty(),
)),
series_schema.clone(),
label_source.clone(),
ordering,
Some(series_schema.clone()),
)?;
Ok(Some(Arc::new(NativeInstantManipulateExec::new(
input,
@@ -103,7 +102,6 @@ impl InstantManipulate {
declared_schema,
series_schema,
label_source,
ordering,
))))
}
}
@@ -114,7 +112,6 @@ struct NativeInstantManipulateExec {
raw_schema: SchemaRef,
series_schema: Arc<SeriesSchema>,
label_source: RawLabelSource,
ordering: SeriesOrdering,
properties: Arc<PlanProperties>,
}
impl NativeInstantManipulateExec {
@@ -124,7 +121,6 @@ impl NativeInstantManipulateExec {
raw_schema: SchemaRef,
series_schema: Arc<SeriesSchema>,
label_source: RawLabelSource,
ordering: SeriesOrdering,
) -> Self {
let properties = Arc::new(PlanProperties::new(
EquivalenceProperties::new(mapping.schema()),
@@ -138,7 +134,6 @@ impl NativeInstantManipulateExec {
raw_schema,
series_schema,
label_source,
ordering,
properties,
}
}
@@ -190,7 +185,6 @@ impl ExecutionPlan for NativeInstantManipulateExec {
self.raw_schema.clone(),
self.series_schema.clone(),
self.label_source.clone(),
self.ordering,
)))
}
fn execute(
@@ -204,12 +198,10 @@ impl ExecutionPlan for NativeInstantManipulateExec {
));
}
let frame = self.input.execute(partition, context.clone())?;
let source = unpack_raw_series(
frame,
self.series_schema.clone(),
self.label_source.clone(),
self.ordering,
)?;
// The child's frames state their own label layout; the plan's binding is an expectation
// checked against that declaration, not a decode input.
ensure_frame_matches_label_source(frame.schema().as_ref(), &self.label_source)?;
let source = unpack_raw_series(frame, Some(self.series_schema.clone()))?;
let source = materialize_raw_labels(source)
.map_err(|e| DataFusionError::Execution(e.to_string()))?;
let reservation =
+30 -12
View File
@@ -23,6 +23,7 @@ use std::task::{Context, Poll};
use common_error::ext::BoxedError;
use datatypes::error::InvalidSeriesSnafu;
use datatypes::series::SeriesOrdering;
use datatypes::series::frame::LabelDecodeSpec;
use datatypes::series::raw::RawSeriesChunk;
use datatypes::series::samples::SeriesSchema;
use futures::Stream;
@@ -31,17 +32,20 @@ use store_api::region_engine::{RawLabelSource, RawSeriesStream, SendableRawSerie
pub use merge::merge_ordered_raw_series;
/// Materializes complete metric labels from a scan-bound raw-series source.
/// Materializes complete metric labels from a raw-series frame stream.
///
/// The caller supplies an already-bound source. Adjacent-chunk pack/unpack validation remains
/// the caller's responsibility, and nonadjacent key recurrence is governed by the producer
/// contract. Explicit labels are retained as supplied; their completeness cannot be inferred
/// from the schema. No authorization or cross-stream completion is performed here.
/// The labels are decoded from the stream's own label decode description, so a consumer that holds
/// only the frame needs no scan-bound binding; a stream that carries complete labels already, and
/// so declares no description, needs no decoder and is forwarded as supplied. Adjacent-chunk
/// pack/unpack validation remains the caller's responsibility, and nonadjacent key recurrence is
/// governed by the producer contract. Explicit labels are retained as supplied; their completeness
/// cannot be inferred from the schema. No authorization or cross-stream completion is performed
/// here.
pub fn materialize_raw_labels(
source: SendableRawSeriesStream,
) -> Result<SendableRawSeriesStream, BoxedError> {
let schema = Arc::clone(source.schema());
let label_source = source.label_source().clone();
let decode = source.label_decode_spec().cloned();
let ordering = source.ordering();
if schema.eval_grid.is_some() {
return Err(BoxedError::new(
@@ -59,12 +63,12 @@ pub fn materialize_raw_labels(
.build(),
));
}
let decoder = match &label_source {
RawLabelSource::PrimaryKey { .. } => Some(
PrimaryKeyLabelDecoder::new(&label_source, LabelProjection::All)
let decoder = match &decode {
Some(description) => Some(
PrimaryKeyLabelDecoder::from_decode_spec(description, LabelProjection::All)
.map_err(BoxedError::new)?,
),
RawLabelSource::Explicit => None,
None => None,
};
Ok(Box::pin(MaterializedRawSeriesStream {
@@ -125,6 +129,12 @@ impl RawSeriesStream for MaterializedRawSeriesStream {
&self.label_source
}
/// Returns `None`: materialized labels are already complete, so this stream declares no label
/// decode description and a consumer needs no decoder.
fn label_decode_spec(&self) -> Option<&LabelDecodeSpec> {
None
}
fn ordering(&self) -> SeriesOrdering {
self.ordering
}
@@ -150,10 +160,18 @@ impl MaterializedRawSeriesStream {
}
}
(None, Some(metric)) => Some(metric),
_ => {
(Some(_), Some(_)) => {
return Err(BoxedError::new(
InvalidSeriesSnafu {
reason: "raw chunk label mode does not match source",
reason: "raw chunk already carries labels but the source describes a key",
}
.build(),
));
}
(None, None) => {
return Err(BoxedError::new(
InvalidSeriesSnafu {
reason: "raw chunk carries no labels and the source describes no key",
}
.build(),
));
+68 -59
View File
@@ -23,17 +23,25 @@ use std::task::{Context, Poll};
use common_error::ext::BoxedError;
use datatypes::error::InvalidSeriesSnafu;
use datatypes::series::SeriesOrdering;
use datatypes::series::frame::LabelDecodeSpec;
use datatypes::series::raw::RawSeriesChunk;
use datatypes::series::samples::{SeriesFields, SeriesSchema};
use futures::Stream;
use mito_codec::labels::{LabelProjection, PrimaryKeyLabelDecoder};
use store_api::region_engine::{RawLabelSource, RawSeriesStream, SendableRawSeriesStream};
use store_api::region_engine::{
NO_BOUND_LABEL_SOURCE, RawLabelSource, RawSeriesStream, SendableRawSeriesStream,
};
/// Merge compatible, explicitly supplied raw streams by complete read key and time.
///
/// Callers provide authorized bindings and own snapshot/storage policy. This does not select an
/// MVCC view, resolve overlapping physical copies, or roll back emitted prefixes after errors.
/// `is_last` is global only over these inputs; there is no fallback, replay, or input sorting.
///
/// The merge contract is the label layout, not a binding: every input must declare the same one —
/// from its frame if it arrived as a frame, otherwise from its own binding — and the merged stream
/// advertises that one declaration. A consumer therefore decodes the merged stream from the frame;
/// the identity it reports is none, see [`RawSeriesStream::label_source`].
pub fn merge_ordered_raw_series(
inputs: Vec<SendableRawSeriesStream>,
) -> Result<SendableRawSeriesStream, BoxedError> {
@@ -42,7 +50,7 @@ pub fn merge_ordered_raw_series(
.next()
.ok_or_else(|| invalid("raw merge requires at least one input"))?;
let schema = Arc::clone(first.schema());
let binding = first.label_source().clone();
let decode = merged_decode_spec(first.as_ref().get_ref())?;
let mut streams = vec![Input::new(first)];
for stream in inputs {
streams.push(Input::new(stream));
@@ -52,13 +60,13 @@ pub fn merge_ordered_raw_series(
.stream
.as_ref()
.ok_or_else(|| invalid("raw merge input disappeared"))?;
validate_binding(stream.label_source())?;
validate_input(stream.as_ref().get_ref(), &schema, &binding)?;
validate_input(stream.as_ref().get_ref(), &schema)?;
validate_decode(stream.as_ref().get_ref(), &decode)?;
}
Ok(Box::pin(OrderedRawMerge {
streams,
schema,
binding,
decode,
candidate: None,
}))
}
@@ -67,17 +75,50 @@ fn invalid(reason: &'static str) -> BoxedError {
BoxedError::new(InvalidSeriesSnafu { reason }.build())
}
fn validate_binding(binding: &RawLabelSource) -> Result<(), BoxedError> {
PrimaryKeyLabelDecoder::new(binding, LabelProjection::All)
.map(|_| ())
.map_err(BoxedError::new)
/// Returns the label layout a stream declares, with no binding required to state it.
///
/// Every key-encoded stream declares its own description — a producer derives it from the binding it
/// read under and states it before the first poll, and a stream reassembled from a frame carries the
/// description its producer wrote — so the answer is exactly what decoding that stream's chunks
/// needs. A merge can therefore compare inputs that were framed and inputs that were not in the same
/// terms. `None` means the stream carries complete labels and declares no key layout at all.
fn declared_layout(stream: &dyn RawSeriesStream) -> Option<LabelDecodeSpec> {
stream.label_decode_spec().cloned()
}
fn validate_input(
/// Returns the one label decode description the merged stream advertises for all inputs.
///
/// The first input's declaration is forwarded unchanged, so a merge advertises exactly what its
/// inputs advertise, whether they are frame-decoded or straight from a producer. The declaration is
/// validated as a decode layout before it is advertised, so a downstream consumer holding only the
/// merged frame schema can decode it.
fn merged_decode_spec(stream: &dyn RawSeriesStream) -> Result<LabelDecodeSpec, BoxedError> {
let description = declared_layout(stream).ok_or_else(|| {
invalid("raw merge requires inputs that describe a primary-key label layout")
})?;
PrimaryKeyLabelDecoder::from_decode_spec(&description, LabelProjection::All)
.map_err(BoxedError::new)?;
Ok(description)
}
/// Requires every input to declare the advertised label layout.
///
/// The merged stream advertises one description, so an input declaring another would be decoded as
/// if it had the advertised encoding. Every input is compared through [`declared_layout`], which
/// reads the input's own declaration, so no input needs a binding to be present.
fn validate_decode(
stream: &dyn RawSeriesStream,
schema: &SeriesSchema,
binding: &RawLabelSource,
decode: &LabelDecodeSpec,
) -> Result<(), BoxedError> {
if declared_layout(stream).as_ref() != Some(decode) {
return Err(invalid(
"raw merge inputs describe different label decode layouts",
));
}
Ok(())
}
fn validate_input(stream: &dyn RawSeriesStream, schema: &SeriesSchema) -> Result<(), BoxedError> {
let other = stream.schema();
if stream.ordering() != SeriesOrdering::KeyThenTime
|| other.eval_grid.is_some()
@@ -88,58 +129,14 @@ fn validate_input(
.iter()
.zip(schema.fields.iter())
.any(|(a, b)| a.name != b.name)
|| !bindings_compatible(stream.label_source(), binding)
{
return Err(invalid(
"raw merge requires equal raw schemas, primary-key bindings, and ReadKeyThenTime ordering",
"raw merge requires equal raw schemas and ReadKeyThenTime ordering",
));
}
Ok(())
}
fn bindings_compatible(left: &RawLabelSource, right: &RawLabelSource) -> bool {
let (
RawLabelSource::PrimaryKey {
metric_name: lm,
physical_metadata: lmeta,
logical_tags: lt,
},
RawLabelSource::PrimaryKey {
metric_name: rm,
physical_metadata: rmeta,
logical_tags: rt,
},
) = (left, right)
else {
return false;
};
if lm != rm
|| lt.len() != rt.len()
|| lt.iter().any(|tag| {
!rt.iter()
.any(|other| tag.column_id == other.column_id && tag.name == other.name)
})
{
return false;
}
let lm = lmeta.as_ref();
let rm = rmeta.as_ref();
lm.primary_key == rm.primary_key
&& lm.primary_key_encoding == rm.primary_key_encoding
&& lm
.primary_key
.iter()
.all(|id| match (lm.column_by_id(*id), rm.column_by_id(*id)) {
(Some(a), Some(b)) => {
a.column_id == b.column_id
&& a.column_schema.name == b.column_schema.name
&& a.column_schema.data_type == b.column_schema.data_type
&& a.semantic_type == b.semantic_type
}
_ => false,
})
}
struct Input {
stream: Option<SendableRawSeriesStream>,
head: Option<RawSeriesChunk>,
@@ -223,7 +220,8 @@ impl Input {
struct OrderedRawMerge {
streams: Vec<Input>,
schema: Arc<SeriesSchema>,
binding: RawLabelSource,
/// The description the merged stream advertises for all inputs; see [`merged_decode_spec`].
decode: LabelDecodeSpec,
candidate: Option<RawSeriesChunk>,
}
@@ -441,8 +439,19 @@ impl RawSeriesStream for OrderedRawMerge {
fn schema(&self) -> &Arc<SeriesSchema> {
&self.schema
}
/// Returns [`NO_BOUND_LABEL_SOURCE`] unconditionally.
///
/// A merge is not a region scan, so it has no scan-bound identity to report, and it does not
/// invent one from its inputs. Its decode contract is [`Self::label_decode_spec`]; identity is
/// not something a merged stream can report.
fn label_source(&self) -> &RawLabelSource {
&self.binding
&NO_BOUND_LABEL_SOURCE
}
/// Returns the description this merge advertises for every input, so a consumer of the merged
/// stream decodes labels from the frame instead of from the binding.
fn label_decode_spec(&self) -> Option<&LabelDecodeSpec> {
Some(&self.decode)
}
fn ordering(&self) -> SeriesOrdering {
SeriesOrdering::KeyThenTime
+296 -160
View File
@@ -23,12 +23,15 @@ use std::task::{Context, Poll, Waker};
use api::v1::SemanticType;
use common_error::ext::{BoxedError, ErrorExt, PlainError};
use common_error::status_code::StatusCode;
use common_query::series_frame::{pack_raw_series, raw_series_frame_ordering, unpack_raw_series};
use common_query::series_frame::{
label_decode_spec, pack_raw_series, raw_series_frame_ordering, unpack_raw_series,
};
use datatypes::arrow::array::Float64Array;
use datatypes::arrow::buffer::{Buffer, ScalarBuffer};
use datatypes::data_type::ConcreteDataType;
use datatypes::prelude::Value;
use datatypes::schema::ColumnSchema;
use datatypes::series::frame::LabelDecodeSpec;
use datatypes::series::raw::{RawSeriesChunk, RawTimestamps};
use datatypes::series::samples::{FieldSpec, SampleRef, Samples, SeriesFields, SeriesSchema};
use datatypes::series::time::{EvalGrid, GridSpec};
@@ -162,6 +165,8 @@ fn chunk(key: Buffer, times: &[i64], values: &[f64], is_last: bool) -> RawSeries
struct TestStream {
schema: Arc<SeriesSchema>,
labels: RawLabelSource,
/// The description this input declares; `None` means its chunks carry complete labels.
decode: Option<LabelDecodeSpec>,
ordering: SeriesOrdering,
items: VecDeque<Poll<Option<Result<RawSeriesChunk, BoxedError>>>>,
// Delivered once the gate opens, so a blocked body can resume with later chunks.
@@ -204,6 +209,9 @@ impl RawSeriesStream for TestStream {
fn label_source(&self) -> &RawLabelSource {
&self.labels
}
fn label_decode_spec(&self) -> Option<&LabelDecodeSpec> {
self.decode.as_ref()
}
fn ordering(&self) -> SeriesOrdering {
self.ordering
}
@@ -228,6 +236,9 @@ fn stream_with(
(
Box::pin(TestStream {
schema,
// Derived once from the binding this fixture stores, as a producer does; an
// explicit-label fixture declares none.
decode: label_decode_spec(&labels),
labels,
ordering,
items: items.into_iter().collect(),
@@ -253,6 +264,7 @@ fn gated_stream(
(
Box::pin(TestStream {
schema: schema(),
decode: label_decode_spec(&labels),
labels,
ordering: SeriesOrdering::KeyThenTime,
items: [ready(first)].into(),
@@ -280,6 +292,7 @@ fn gated_resuming_stream(
(
Box::pin(TestStream {
schema: schema(),
decode: label_decode_spec(&labels),
labels,
ordering: SeriesOrdering::KeyThenTime,
items: [ready(first)].into(),
@@ -326,6 +339,7 @@ fn gated_pending_stream(
(
Box::pin(TestStream {
schema: schema(),
decode: label_decode_spec(&labels),
labels,
ordering: SeriesOrdering::KeyThenTime,
items: VecDeque::new(),
@@ -376,10 +390,7 @@ async fn merges_full_encoded_keys_and_interleaved_chunks_with_global_finality()
let mut merged = merge_ordered_raw_series(vec![left, right]).unwrap();
assert_eq!(merged.ordering(), SeriesOrdering::KeyThenTime);
assert!(merged.schema().eval_grid.is_none());
assert!(matches!(
merged.label_source(),
RawLabelSource::PrimaryKey { .. }
));
assert!(matches!(merged.label_source(), RawLabelSource::Explicit));
let mut observed = Vec::new();
while let Some(item) = merged.next().await {
let item = item.unwrap();
@@ -446,11 +457,10 @@ async fn merges_two_unpacked_frames_through_the_frame_codec() {
chunks.into_iter().map(ready),
);
let schema = Arc::clone(stream.schema());
let binding = stream.label_source().clone();
let packed = pack_raw_series(stream).unwrap();
let ordering = raw_series_frame_ordering(packed.schema().as_ref()).unwrap();
assert_eq!(SeriesOrdering::KeyThenTime, ordering);
unpack_raw_series(packed, schema, binding, ordering).unwrap()
unpack_raw_series(packed, Some(schema)).unwrap()
};
let left = framed(
label_source(1),
@@ -1219,193 +1229,97 @@ async fn greater_key_head_allows_finished_series_before_future_eof() {
assert_eq!(right_drops.load(Ordering::SeqCst), 1);
}
/// The merge contract is the label layout, which is exactly what a frame can carry: any difference a
/// frame could state — the metric name, the encoding, the key columns with their physical names,
/// types, and order, and the logical tag mapping — is a conflict and is rejected before any input is
/// polled.
#[tokio::test]
async fn preflight_rejects_incompatible_primary_key_bindings_without_polling() {
let full_baseline = label_source(1);
let physical_comparison_baseline = {
async fn preflight_rejects_incompatible_label_layouts_without_polling() {
let sensor = Arc::<[BoundTag]>::from([BoundTag {
column_id: 9,
name: Arc::from("sensor"),
}]);
let rack = Arc::<[BoundTag]>::from([BoundTag {
column_id: 8,
name: Arc::from("rack"),
}]);
let edited = |edit: &dyn Fn(&mut store_api::metadata::RegionMetadata),
logical_tags: Arc<[BoundTag]>| {
let RawLabelSource::PrimaryKey {
metric_name,
physical_metadata,
..
} = label_source(1)
} = label_source(2)
else {
unreachable!()
};
let mut changed = (*physical_metadata).clone();
edit(&mut changed);
RawLabelSource::PrimaryKey {
metric_name,
physical_metadata,
logical_tags: Arc::from([BoundTag {
column_id: 9,
name: Arc::from("sensor"),
}]),
physical_metadata: Arc::new(changed),
logical_tags,
}
};
let cases = [
(
"metric name",
label_source_with(2, PrimaryKeyEncoding::Dense, "other_metric"),
"binding",
),
(
"encoding",
label_source_with(2, PrimaryKeyEncoding::Sparse, "room_temperature"),
"binding",
),
(
"primary key column ids",
{
let RawLabelSource::PrimaryKey {
metric_name,
physical_metadata,
logical_tags,
} = label_source(2)
else {
unreachable!()
};
let mut changed = (*physical_metadata).clone();
changed.primary_key = vec![9, 8];
RawLabelSource::PrimaryKey {
metric_name,
physical_metadata: Arc::new(changed),
logical_tags,
}
},
"binding",
"primary key column order",
edited(
&|metadata| metadata.primary_key = vec![9, 8],
sensor.clone(),
),
),
(
"primary key physical name mismatch",
{
let RawLabelSource::PrimaryKey {
metric_name,
physical_metadata,
..
} = label_source(2)
else {
unreachable!()
};
let mut changed = (*physical_metadata).clone();
changed
.column_metadatas
.iter_mut()
.find(|column| column.column_id == 8)
.unwrap()
.column_schema
.name = "shelf".to_string();
RawLabelSource::PrimaryKey {
metric_name,
physical_metadata: Arc::new(changed),
logical_tags: Arc::from([BoundTag {
column_id: 9,
name: Arc::from("sensor"),
}]),
}
},
"merge compatibility",
),
(
"primary key semantic role",
{
let RawLabelSource::PrimaryKey {
metric_name,
physical_metadata,
..
} = label_source(2)
else {
unreachable!()
};
let mut changed = (*physical_metadata).clone();
changed
.column_metadatas
.iter_mut()
.find(|column| column.column_id == 8)
.unwrap()
.semantic_type = SemanticType::Field;
RawLabelSource::PrimaryKey {
metric_name,
physical_metadata: Arc::new(changed),
logical_tags: Arc::from([BoundTag {
column_id: 9,
name: Arc::from("sensor"),
}]),
}
},
"merge compatibility",
"primary key physical name",
edited(
&|metadata| {
metadata
.column_metadatas
.iter_mut()
.find(|column| column.column_id == 8)
.unwrap()
.column_schema
.name = "shelf".to_string();
},
sensor.clone(),
),
),
(
"primary key physical type",
{
let RawLabelSource::PrimaryKey {
metric_name,
physical_metadata,
logical_tags,
} = label_source(2)
else {
unreachable!()
};
let mut changed = (*physical_metadata).clone();
changed
.column_metadatas
.iter_mut()
.find(|column| column.column_id == 8)
.unwrap()
.column_schema
.data_type = ConcreteDataType::int64_datatype();
RawLabelSource::PrimaryKey {
metric_name,
physical_metadata: Arc::new(changed),
logical_tags,
}
},
"binding",
),
(
"logical tag mapping",
{
let RawLabelSource::PrimaryKey {
metric_name,
physical_metadata,
..
} = label_source(2)
else {
unreachable!()
};
RawLabelSource::PrimaryKey {
metric_name,
physical_metadata,
logical_tags: Arc::from([BoundTag {
column_id: 8,
name: Arc::from("rack"),
}]),
}
},
"binding",
edited(
&|metadata| {
metadata
.column_metadatas
.iter_mut()
.find(|column| column.column_id == 8)
.unwrap()
.column_schema
.data_type = ConcreteDataType::int64_datatype();
},
rack.clone(),
),
),
("logical tag mapping", edited(&|_| {}, rack)),
];
for (description, incompatible, reason) in cases {
let baseline = if reason == "merge compatibility" {
&physical_comparison_baseline
} else {
&full_baseline
};
let (first, polls_a, drops_a) =
stream((*baseline).clone(), SeriesOrdering::KeyThenTime, []);
for (description, incompatible) in cases {
let (first, polls_a, drops_a) = stream(label_source(1), SeriesOrdering::KeyThenTime, []);
let (second, polls_b, drops_b) = stream(incompatible, SeriesOrdering::KeyThenTime, []);
let error = match merge_ordered_raw_series(vec![first, second]) {
Ok(_) => panic!("{description} mismatch must be rejected"),
Err(error) => error.to_string(),
};
if reason == "merge compatibility" {
assert_eq!(
error,
"Invalid series: raw merge requires equal raw schemas, primary-key bindings, and ReadKeyThenTime ordering",
"{description}"
);
} else {
assert!(
error.to_lowercase().contains(reason),
"{description}: {error}"
);
}
assert!(
error.to_lowercase().contains("decode layouts"),
"{description}: {error}"
);
assert_eq!(polls_a.load(Ordering::SeqCst), 0);
assert_eq!(polls_b.load(Ordering::SeqCst), 0);
assert_eq!(drops_a.load(Ordering::SeqCst), 1);
@@ -1413,6 +1327,45 @@ async fn preflight_rejects_incompatible_primary_key_bindings_without_polling() {
}
}
/// A difference that lives only in local region metadata — here a primary-key column's semantic
/// role — is not a merge conflict: a frame cannot state it, and both sides still decode to the same
/// labels, so holding producer-side inputs to it would make them answerable to a stricter contract
/// than the wire form can express.
#[tokio::test]
async fn preflight_ignores_metadata_the_label_layout_cannot_express() {
let base = label_source(1);
let RawLabelSource::PrimaryKey {
metric_name,
physical_metadata,
logical_tags,
} = label_source(2)
else {
unreachable!()
};
let mut changed = (*physical_metadata).clone();
changed
.column_metadatas
.iter_mut()
.find(|column| column.column_id == 8)
.unwrap()
.semantic_type = SemanticType::Field;
let incompatible = RawLabelSource::PrimaryKey {
metric_name,
physical_metadata: Arc::new(changed),
logical_tags,
};
let (first, _, _) = stream(base, SeriesOrdering::KeyThenTime, []);
let (second, _, _) = stream(incompatible, SeriesOrdering::KeyThenTime, []);
let merged = merge_ordered_raw_series(vec![first, second])
.expect("the same label layout merges whatever a local metadata snapshot says");
assert_eq!(
merged.label_decode_spec(),
label_decode_spec(&label_source(1)).as_ref(),
"the merged stream advertises the layout both sides agree on"
);
}
#[tokio::test]
async fn metadata_snapshots_with_same_layout_are_accepted() {
let base = label_source(1);
@@ -1882,3 +1835,186 @@ async fn primary_key_codec_identity_survives_merge_then_label_materialization()
}
assert!(output.is_last);
}
/// The merged stream advertises one decode description, so an input that describes its own labels
/// differently is rejected instead of being decoded under the advertised layout. An input carrying
/// the same layout is accepted, which is what a merge over unpacked frames looks like.
#[tokio::test]
async fn merge_rejects_inputs_that_describe_different_decode_layouts() {
let carried =
label_decode_spec(&label_source(1)).expect("a primary-key binding describes its key");
let mut different = carried.clone();
different.metric_name = "other_metric".to_string();
let carried_stream = |decode: LabelDecodeSpec| {
let polls = Arc::new(AtomicUsize::new(0));
let drops = Arc::new(AtomicUsize::new(0));
(
Box::pin(TestStream {
schema: schema(),
labels: label_source(1),
decode: Some(decode),
ordering: SeriesOrdering::KeyThenTime,
items: VecDeque::new(),
deferred: VecDeque::new(),
polls: polls.clone(),
drops: drops.clone(),
gate: None,
release: None,
}) as SendableRawSeriesStream,
polls,
drops,
)
};
let (first, _, _) = carried_stream(carried.clone());
let (same_layout, _, _) = carried_stream(carried.clone());
let merged = merge_ordered_raw_series(vec![first, same_layout])
.expect("inputs that declare the same layout merge");
assert_eq!(merged.label_decode_spec(), Some(&carried));
let (first, _, _) = carried_stream(carried.clone());
let (mismatched, mismatched_polls, mismatched_drops) = carried_stream(different);
let error = match merge_ordered_raw_series(vec![first, mismatched]) {
Ok(_) => panic!("inputs describing different layouts must be rejected"),
Err(error) => error.to_string(),
};
assert!(error.to_lowercase().contains("decode"), "{error}");
assert_eq!(mismatched_polls.load(Ordering::SeqCst), 0);
assert_eq!(mismatched_drops.load(Ordering::SeqCst), 1);
}
/// A merge whose inputs are frame-decoded has no identity to forward, and the merged stream says so
/// instead of inventing one: it reports [`NO_BOUND_LABEL_SOURCE`] while its description still states
/// that its chunks are undecoded key chunks, which is what a consumer decodes from.
#[tokio::test]
async fn merge_without_a_bound_identity_still_advertises_the_decode_layout() {
use common_query::series_frame::raw_series_frame_ordering;
use common_query::series_frame::{pack_raw_series, unpack_raw_series};
use store_api::region_engine::NO_BOUND_LABEL_SOURCE;
use crate::source::materialize_raw_labels;
let carried =
label_decode_spec(&label_source(1)).expect("a primary-key binding describes its key");
let key = read_key("rack-a", "sensor-1");
let framed_input = |times: &[i64], values: &[f64]| {
let polls = Arc::new(AtomicUsize::new(0));
let drops = Arc::new(AtomicUsize::new(0));
Box::pin(TestStream {
schema: schema(),
labels: NO_BOUND_LABEL_SOURCE,
decode: Some(carried.clone()),
ordering: SeriesOrdering::KeyThenTime,
items: VecDeque::from([Poll::Ready(Some(Ok(chunk(
key.clone(),
times,
values,
true,
))))]),
deferred: VecDeque::new(),
polls,
drops,
gate: None,
release: None,
}) as SendableRawSeriesStream
};
let merged = merge_ordered_raw_series(vec![
framed_input(&[10, 30], &[1.0, 3.0]),
framed_input(&[20, 40], &[2.0, 4.0]),
])
.expect("inputs that declare the same layout merge");
assert_eq!(merged.label_decode_spec(), Some(&carried));
assert!(
matches!(merged.label_source(), RawLabelSource::Explicit),
"no identity was established, and none is invented for the merged stream: {:?}",
NO_BOUND_LABEL_SOURCE
);
assert_eq!(merged.ordering(), SeriesOrdering::KeyThenTime);
// Packing advertises that layout under the merge's ordering, so the frame states both.
let framed = pack_raw_series(merged).unwrap();
assert_eq!(
datatypes::series::frame::raw_series_frame_label_decode(framed.schema().as_ref()).unwrap(),
Some(carried.clone())
);
assert_eq!(
raw_series_frame_ordering(framed.schema().as_ref()).unwrap(),
SeriesOrdering::KeyThenTime
);
// And it decodes: materialization reads the advertised description, not an identity.
let raw = unpack_raw_series(framed, Some(schema())).unwrap();
let rows = materialize_raw_labels(raw)
.unwrap()
.try_collect::<Vec<_>>()
.await
.unwrap();
assert_eq!(
rows.iter()
.map(|row| row.time.timestamp(0).unwrap())
.collect::<Vec<_>>(),
[10, 20, 30, 40]
);
let metric = rows[0].metric.clone().expect("labels were materialized");
assert_eq!(metric.name(), Some("room_temperature"));
assert_eq!(metric.get("rack"), Some("rack-a"));
assert_eq!(metric.get("sensor"), Some("sensor-1"));
}
/// A merged stream's label decode description survives the frame wire form: the packed schema
/// declares that description, so a consumer holding only the schema can rebuild a decoder and decode
/// the merged keys without a request-side label source.
#[tokio::test]
async fn merged_frame_declares_a_label_decode_description_that_decodes_its_keys() {
use common_query::series_frame::{pack_raw_series, unpack_raw_series};
use datatypes::series::frame::raw_series_frame_label_decode;
use mito_codec::labels::{LabelProjection, PrimaryKeyLabelDecoder};
let key = read_key("rack-a", "sensor-1");
let (left, _, _) = stream(
label_source(1),
SeriesOrdering::KeyThenTime,
[ready(chunk(key.clone(), &[10, 30], &[1.0, 3.0], true))],
);
let (right, _, _) = stream(
label_source(2),
SeriesOrdering::KeyThenTime,
[ready(chunk(key.clone(), &[20, 40], &[2.0, 4.0], true))],
);
let merged = merge_ordered_raw_series(vec![left, right]).unwrap();
let expected = merged
.label_decode_spec()
.cloned()
.expect("the merge advertises a primary-key layout");
let series_schema = merged.schema().clone();
let binding = label_source(1);
let ordering = merged.ordering();
let framed = pack_raw_series(merged).unwrap();
assert_eq!(
common_query::series_frame::raw_series_frame_ordering(framed.schema().as_ref()).unwrap(),
ordering
);
let declared = framed.schema();
assert_eq!(
raw_series_frame_label_decode(declared.as_ref()).unwrap(),
Some(expected.clone())
);
// The recovered description alone rebuilds an equivalent decoder; no metadata snapshot from
// the merge is consulted.
let mut from_schema =
PrimaryKeyLabelDecoder::from_decode_spec(&expected, LabelProjection::All).unwrap();
let mut from_binding = PrimaryKeyLabelDecoder::new(&binding, LabelProjection::All).unwrap();
assert_eq!(
from_schema.decode(&key).unwrap(),
from_binding.decode(&key).unwrap()
);
let mut decoded = unpack_raw_series(framed, Some(series_schema)).unwrap();
let first = decoded.next().await.unwrap().unwrap();
assert_eq!(first.primary_key.as_slice(), key.as_slice());
assert!(first.metric.is_none(), "key labels stay undecoded");
}
+31 -4
View File
@@ -18,10 +18,14 @@ use std::sync::{Arc, Weak};
use std::task::{Context, Poll};
use common_error::ext::BoxedError;
use common_query::series_frame::{pack_raw_series, unpack_raw_series};
use common_query::series_frame::{
ensure_frame_matches_label_source, label_decode_spec, pack_raw_series,
raw_series_frame_ordering, unpack_raw_series,
};
use datatypes::arrow::array::{Array, Float64Array, ListArray, UInt32Array};
use datatypes::arrow::buffer::{Buffer, ScalarBuffer};
use datatypes::series::SeriesOrdering;
use datatypes::series::frame::LabelDecodeSpec;
use datatypes::series::histogram::{
BucketSideView, CounterResetHint, HistogramBatch, HistogramBatchBuilder, HistogramView,
};
@@ -167,6 +171,8 @@ fn payload_chunk(
struct PayloadStream {
schema: Arc<SeriesSchema>,
labels: RawLabelSource,
/// The description this fixture declares, derived once from `labels`, as a producer does.
decode: Option<LabelDecodeSpec>,
items: VecDeque<Result<RawSeriesChunk, BoxedError>>,
}
@@ -184,6 +190,9 @@ impl RawSeriesStream for PayloadStream {
fn label_source(&self) -> &RawLabelSource {
&self.labels
}
fn label_decode_spec(&self) -> Option<&LabelDecodeSpec> {
self.decode.as_ref()
}
fn ordering(&self) -> SeriesOrdering {
SeriesOrdering::KeyThenTime
}
@@ -192,6 +201,7 @@ impl RawSeriesStream for PayloadStream {
fn payload_stream(labels: RawLabelSource, chunks: Vec<RawSeriesChunk>) -> SendableRawSeriesStream {
Box::pin(PayloadStream {
schema: payload_schema(),
decode: label_decode_spec(&labels),
labels,
items: chunks.into_iter().map(Ok).collect(),
})
@@ -243,10 +253,26 @@ async fn merges_typed_multifield_payload_through_frame_and_label_roundtrip() {
);
let merged = merge_ordered_raw_series(vec![a, b]).unwrap();
let frame_schema = merged.schema().clone();
let frame_binding = merged.label_source().clone();
let frame_ordering = merged.ordering();
assert!(
matches!(merged.label_source(), RawLabelSource::Explicit),
"a merge is not a region scan, so it reports no scan-bound identity"
);
let framed = pack_raw_series(merged).unwrap();
let raw = unpack_raw_series(framed, frame_schema, frame_binding, frame_ordering).unwrap();
// The frame states the layout the decoder uses: it matches the description an input's binding
// declares, since the merge only accepts inputs that declare one layout.
ensure_frame_matches_label_source(framed.schema().as_ref(), &label_source(1)).unwrap();
assert_eq!(
raw_series_frame_ordering(framed.schema().as_ref()).unwrap(),
frame_ordering,
"the packed frame declares the merge's ordering"
);
let raw = unpack_raw_series(framed, Some(frame_schema)).unwrap();
assert_eq!(
raw.ordering(),
frame_ordering,
"unpacking derives the ordering from the frame"
);
let rows = materialize_raw_labels(raw)
.unwrap()
.try_collect::<Vec<_>>()
@@ -442,7 +468,8 @@ async fn merge_retains_sliced_float_time_and_histogram_owners_until_output_drop(
["temperature", "latency", "observations"]
);
assert!(
matches!(merged.label_source(), RawLabelSource::PrimaryKey { metric_name, .. } if metric_name.as_ref() == "room_temperature")
matches!(merged.label_source(), RawLabelSource::Explicit),
"a merge is not a region scan, so it reports no scan-bound identity"
);
assert_eq!(merged.ordering(), SeriesOrdering::KeyThenTime);
let prefix = merged.next().await.unwrap().unwrap();
+83
View File
@@ -21,11 +21,13 @@ use std::task::{Context, Poll};
use api::v1::SemanticType;
use common_error::ext::{BoxedError, PlainError};
use common_error::status_code::StatusCode;
use common_query::series_frame::{label_decode_spec, pack_raw_series, unpack_raw_series};
use datatypes::arrow::array::{Array, Float64Array, UInt32Array};
use datatypes::arrow::buffer::{Buffer, ScalarBuffer};
use datatypes::data_type::ConcreteDataType;
use datatypes::prelude::Value;
use datatypes::schema::ColumnSchema;
use datatypes::series::frame::LabelDecodeSpec;
use datatypes::series::histogram::{
BucketSideView, CounterResetHint, HistogramBatch, HistogramBatchBuilder, HistogramView,
};
@@ -175,6 +177,9 @@ fn chunk(read_key: Buffer, metric: Option<MetricName>, final_chunk: bool) -> Raw
struct MockStream {
schema: Arc<SeriesSchema>,
source: RawLabelSource,
/// The label decode description this stream declares; a key-mode fake derives it from its
/// binding, as a frame producer would write it into the schema.
decode: Option<LabelDecodeSpec>,
ordering: SeriesOrdering,
items: VecDeque<Poll<Option<Result<RawSeriesChunk, BoxedError>>>>,
polls: Arc<AtomicUsize>,
@@ -197,6 +202,9 @@ impl RawSeriesStream for MockStream {
fn label_source(&self) -> &RawLabelSource {
&self.source
}
fn label_decode_spec(&self) -> Option<&LabelDecodeSpec> {
self.decode.as_ref()
}
fn ordering(&self) -> SeriesOrdering {
self.ordering
}
@@ -216,10 +224,12 @@ fn mock_with(
) -> (SendableRawSeriesStream, Arc<AtomicUsize>, Arc<AtomicUsize>) {
let polls = Arc::new(AtomicUsize::new(0));
let drops = Arc::new(AtomicUsize::new(0));
let decode = label_decode_spec(&source);
(
Box::pin(MockStream {
schema,
source,
decode,
ordering,
items: items.into(),
polls: polls.clone(),
@@ -807,6 +817,7 @@ async fn constructor_and_mode_errors_drop_input_without_polling_ahead() {
let raw: SendableRawSeriesStream = Box::pin(MockStream {
schema: invalid_schema,
source: RawLabelSource::Explicit,
decode: None,
ordering: SeriesOrdering::Contiguous,
items: VecDeque::new(),
polls: polls.clone(),
@@ -1038,3 +1049,75 @@ async fn dense_and_sparse_sources_materialize_complete_labels() {
assert_eq!(metric.tags().len(), 2);
}
}
/// A consumer that received only the raw series frame decodes labels with no scan-bound source:
/// the producer's description travels inside the frame schema, so unpacking needs no binding and
/// materialization builds its decoder from that description alone.
#[tokio::test]
async fn materializes_labels_from_a_frame_with_no_supplied_binding() {
for encoding in [PrimaryKeyEncoding::Dense, PrimaryKeyEncoding::Sparse] {
let binding = source(encoding);
let description =
label_decode_spec(&binding).expect("a primary-key binding describes its key");
let read_key = key(
encoding,
&[
(8, Value::String("last".into())),
(2, Value::String("hidden".into())),
(41, Value::String("first".into())),
],
);
let (raw, _, _) = mock(
binding,
vec![Poll::Ready(Some(Ok(chunk(read_key.clone(), None, true))))],
);
let framed = pack_raw_series(raw).unwrap();
assert_eq!(
datatypes::series::frame::raw_series_frame_label_decode(framed.schema().as_ref())
.unwrap(),
Some(description.clone()),
"{encoding:?}: the frame schema is the producer's only label statement"
);
// Everything the consumer keeps is the frame stream: no expected schema, no binding, no
// region metadata anywhere on the decode path.
let framed = unpack_raw_series(framed, None).unwrap();
assert_eq!(
framed.label_decode_spec(),
Some(&description),
"{encoding:?}: the parsed description survives unpacking without a binding"
);
assert!(
!matches!(framed.label_source(), RawLabelSource::PrimaryKey { .. }),
"{encoding:?}: no physical region key layout was supplied"
);
assert_eq!(
framed
.schema()
.fields
.iter()
.map(|field| field.name.to_string())
.collect::<Vec<_>>(),
schema()
.fields
.iter()
.map(|field| field.name.to_string())
.collect::<Vec<_>>(),
"{encoding:?}: the consumer held no schema, so these are the producer's declared \
fields read back from the frame"
);
let mut output = materialize_raw_labels(framed).unwrap();
let materialized = output.next().await.unwrap().unwrap();
assert_eq!(materialized.primary_key, read_key);
let metric = materialized
.metric
.expect("labels were materialized from the frame description");
assert_eq!(metric.name(), Some("requests_total"));
assert_eq!(metric.get("zeta"), Some("last"));
assert_eq!(metric.get("alpha"), Some("first"));
assert_eq!(metric.get("private"), None);
assert_eq!(metric.tags().len(), 2);
assert!(output.next().await.is_none());
}
}
+93
View File
@@ -3419,6 +3419,99 @@ mod tests {
}
}
/// The frame's label decode description is compared as an exact string, so two sides that
/// derive it from equal bindings must produce byte-identical text, and any textual difference
/// must be a mismatch rather than a silent re-decode under one side.
#[test]
fn merge_scan_compares_label_decode_metadata_as_exact_text() {
use common_query::series_frame::{
label_decode_spec, raw_series_frame_schema_with_ordering,
};
use datatypes::series::SeriesOrdering;
use datatypes::series::frame::raw_series_frame_label_decode;
let raw = raw_test_series_schema();
// Two bindings that are equal but not the same allocation: the expectation and the
// advertised schema are built independently, as on a frontend and a datanode.
let binding = || {
let mut builder =
store_api::metadata::RegionMetadataBuilder::new(RegionId::new(1024, 1));
builder
.push_column_metadata(store_api::metadata::ColumnMetadata {
column_schema: datatypes::schema::ColumnSchema::new(
"host",
ConcreteDataType::string_datatype(),
true,
),
semantic_type: api::v1::SemanticType::Tag,
column_id: 7,
})
.push_column_metadata(store_api::metadata::ColumnMetadata {
column_schema: datatypes::schema::ColumnSchema::new(
"ts",
ConcreteDataType::timestamp_millisecond_datatype(),
false,
),
semantic_type: api::v1::SemanticType::Timestamp,
column_id: 8,
})
.primary_key(vec![7]);
store_api::region_engine::RawLabelSource::PrimaryKey {
metric_name: "cpu_usage".into(),
physical_metadata: Arc::new(builder.build().unwrap()),
logical_tags: Arc::from([store_api::region_engine::BoundTag {
column_id: 7,
name: "host".into(),
}]),
}
};
let expected_binding = binding();
let actual_binding = binding();
assert_eq!(
label_decode_spec(&expected_binding),
label_decode_spec(&actual_binding)
);
let expected = raw_series_frame_schema_with_ordering(
&raw,
&expected_binding,
SeriesOrdering::KeyThenTime,
)
.unwrap();
let actual = raw_series_frame_schema_with_ordering(
&raw,
&actual_binding,
SeriesOrdering::KeyThenTime,
)
.unwrap();
let key = "greptime.series.label_decode";
assert_eq!(expected.metadata().get(key), actual.metadata().get(key));
assert!(
expected.metadata()[key].contains("\"encoding\":\"dense\""),
"the description is plain compact JSON: {}",
expected.metadata()[key]
);
assert!(
validate_remote_schema(&expected, &actual, "test").is_ok(),
"equal bindings must compare equal"
);
// Semantically equal but textually different descriptions are rejected: the reserved
// metadata is compared as exact text, not parsed and compared structurally.
let mut spaced_metadata = expected.metadata().clone();
let spaced_text = spaced_metadata[key].replacen(':', ": ", 1);
assert_ne!(spaced_text, spaced_metadata[key]);
spaced_metadata.insert(key.to_string(), spaced_text);
let spaced = ArrowSchema::new_with_metadata(expected.fields().clone(), spaced_metadata);
assert_eq!(
raw_series_frame_label_decode(&spaced)
.unwrap()
.map(|decode| decode.version),
Some(1),
"the spaced text is still a valid description"
);
assert!(validate_remote_schema(&expected, &spaced, "test").is_err());
}
fn raw_test_series_schema() -> SeriesSchema {
SeriesSchema {
fields: vec![
+35 -3
View File
@@ -31,6 +31,7 @@ use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet;
use datafusion_physical_plan::{DisplayAs, DisplayFormatType, PhysicalExpr};
use datatypes::schema::SchemaRef;
use datatypes::series::SeriesOrdering;
use datatypes::series::frame::LabelDecodeSpec;
use datatypes::series::raw::RawSeriesChunk;
use datatypes::series::samples::SeriesSchema;
use futures::Stream;
@@ -495,10 +496,20 @@ pub enum RawLabelSource {
},
/// The stream supplies a complete metric name and label set in each chunk, not a partial
/// label projection; `Some(metric)` is a complete name/label set (the name itself may be
/// absent).
/// absent), or, for a stream that has no identity at all, see
/// [`NO_BOUND_LABEL_SOURCE`].
Explicit,
}
/// The identity a stream reports when no scan-bound label source was established for it.
///
/// [`RawLabelSource`] has no "absent" variant, and [`RawLabelSource::Explicit`] is the only one that
/// carries no physical region key layout, so it stands in for "no identity" here. It is not a claim
/// that the chunks carry labels: [`RawSeriesStream::label_decode_spec`] answers that, and a stream
/// reporting this value can still return `Some` description and leave `metric` as `None` in every
/// chunk. Decoding therefore asks the description, and only an identity check asks this.
pub const NO_BOUND_LABEL_SOURCE: RawLabelSource = RawLabelSource::Explicit;
/// A stream of raw series chunks with stable schema, label binding, and ordering.
///
/// Every series has strictly increasing timestamps within and across chunks; timestamps are
@@ -508,8 +519,10 @@ pub enum RawLabelSource {
/// per-stream guarantees, not query-global order, partition ownership, or finality. `is_last`
/// marks the last chunk of that series in this stream only; merging requires recomputing
/// per-input-series finalization across all related inputs, not aggregate-group finality.
/// `RawLabelSource` keeps label interpretation self-contained through transport. Stream finality
/// applies to all related input series, not aggregate groups or vector-match state.
/// Label interpretation is carried by the stream: [`RawSeriesStream::label_source`] states the
/// scan-bound identity a producer read under, and [`RawSeriesStream::label_decode_spec`] states the
/// label layout the stream's own chunks declare. Stream finality applies to all related input
/// series, not aggregate groups or vector-match state.
pub trait RawSeriesStream: Stream<Item = Result<RawSeriesChunk, BoxedError>> + Send {
/// Returns the schema shared by all chunks; raw schemas have no evaluation grid.
fn schema(&self) -> &Arc<SeriesSchema>;
@@ -517,7 +530,26 @@ pub trait RawSeriesStream: Stream<Item = Result<RawSeriesChunk, BoxedError>> + S
/// labels are undecoded, not that the metric name is missing; `Explicit` chunks supply
/// `Some` with the complete name/label set, not a partial projection. The metric name itself
/// may be absent.
///
/// A stream that was decoded from a raw frame has no scan-bound identity and reports
/// [`NO_BOUND_LABEL_SOURCE`], which reads as `Explicit` while its chunks may still carry no
/// labels at all. That value therefore answers "is this the region I read under", never "are
/// these labels complete": [`Self::label_decode_spec`] is the stream's statement about its
/// encoding, and every consumer that decides whether to decode reads that instead.
fn label_source(&self) -> &RawLabelSource;
/// Returns the label decode description this stream declares for key-encoded labels.
///
/// Every stream whose chunks carry key-encoded labels declares the description here, whether
/// it is a storage producer or a stream reassembled from a raw frame. A consumer holding only
/// that description can therefore decode labels without a scan-bound [`RawLabelSource`]: the
/// description, not the binding, describes the encoding, and a receiver reporting
/// [`NO_BOUND_LABEL_SOURCE`] still states through this description that its chunks are
/// undecoded key chunks. A stream that carries complete labels instead answers `None`. This is
/// the description the stream declares, not an authenticated statement about where the values
/// came from.
fn label_decode_spec(&self) -> Option<&LabelDecodeSpec> {
None
}
/// Returns this stream's ordering guarantee.
fn ordering(&self) -> SeriesOrdering;
}
@@ -26,11 +26,15 @@ use datatypes::error::InvalidSeriesSnafu;
use datatypes::prelude::ConcreteDataType;
use datatypes::schema::ColumnSchema;
use datatypes::series::SeriesOrdering;
use datatypes::series::frame::{
LABEL_DECODE_VERSION, LabelDecodeColumn, LabelDecodeEncoding, LabelDecodeSpec, LabelDecodeTag,
};
use datatypes::series::raw::{RawSeriesChunk, RawTimestamps};
use datatypes::series::samples::{FieldSpec, Samples, SeriesFields, SeriesSchema};
use datatypes::series::{Label, MetricName};
use futures::{Stream, StreamExt};
use crate::codec::PrimaryKeyEncoding;
use crate::region_engine::{
PreparedSeriesSource, RawLabelSource, RawSeriesStream, SendableRawSeriesStream, SeriesInfo,
};
@@ -91,6 +95,7 @@ impl PreparedSeriesSource for MemorySource {
Ok(Box::pin(MemoryReader {
chunks,
schema: self.schema.clone(),
label_decode_spec: declared_decode_spec(&self.label_source),
label_source: self.label_source.clone(),
polls: self.polls.clone(),
drops: self.drops.clone(),
@@ -101,6 +106,8 @@ impl PreparedSeriesSource for MemorySource {
struct MemoryReader {
chunks: VecDeque<RawSeriesChunk>,
schema: Arc<SeriesSchema>,
/// The description this fixture declares, derived once from `label_source`, as a producer does.
label_decode_spec: Option<LabelDecodeSpec>,
label_source: RawLabelSource,
polls: Arc<AtomicUsize>,
drops: Arc<AtomicUsize>,
@@ -133,11 +140,58 @@ impl RawSeriesStream for MemoryReader {
&self.label_source
}
fn label_decode_spec(&self) -> Option<&LabelDecodeSpec> {
self.label_decode_spec.as_ref()
}
fn ordering(&self) -> SeriesOrdering {
SeriesOrdering::Contiguous
}
}
/// Describes the key-encoded chunks a reader of `source` yields.
///
/// This crate cannot depend on `common-query` (which depends on it), so this fixture repeats the
/// derivation `common_query::series_frame::label_decode_spec` performs for a producer: the metric
/// name, the primary-key encoding, the ordered key columns with their physical types, and the
/// logical tags in canonical order. An explicit binding already carries complete labels and
/// declares nothing.
fn declared_decode_spec(source: &RawLabelSource) -> Option<LabelDecodeSpec> {
let RawLabelSource::PrimaryKey {
metric_name,
physical_metadata,
logical_tags,
} = source
else {
return None;
};
let mut description = LabelDecodeSpec {
version: LABEL_DECODE_VERSION,
metric_name: metric_name.to_string(),
encoding: match physical_metadata.primary_key_encoding {
PrimaryKeyEncoding::Dense => LabelDecodeEncoding::Dense,
PrimaryKeyEncoding::Sparse => LabelDecodeEncoding::Sparse,
},
primary_key: physical_metadata
.primary_key_columns()
.map(|column| LabelDecodeColumn {
column_id: column.column_id,
name: column.column_schema.name.clone(),
physical_type: column.column_schema.data_type.clone(),
})
.collect(),
logical_tags: logical_tags
.iter()
.map(|tag| LabelDecodeTag {
column_id: tag.column_id,
name: tag.name.to_string(),
})
.collect(),
};
description.canonicalize();
Some(description)
}
fn metric(host: &str) -> MetricName {
MetricName::try_new(
Some(Arc::from("cpu")),
@@ -358,6 +412,25 @@ async fn prepared_source_label_binding_is_available_before_readers_and_matches_c
let reader = source.read_series(0..2).unwrap();
assert_same_label_source(source.label_source(), reader.label_source());
match (explicit, reader.label_decode_spec()) {
(true, None) => {}
(false, Some(description)) => {
assert_eq!(description.version, LABEL_DECODE_VERSION);
assert_eq!(description.metric_name, "cpu");
assert_eq!(description.encoding, LabelDecodeEncoding::Dense);
assert_eq!(description.primary_key.len(), 1);
assert_eq!(description.primary_key[0].column_id, 1);
assert_eq!(description.primary_key[0].name, "host");
assert_eq!(
description.primary_key[0].physical_type,
ConcreteDataType::string_datatype()
);
assert_eq!(description.logical_tags.len(), 1);
assert_eq!(description.logical_tags[0].column_id, 1);
assert_eq!(description.logical_tags[0].name, "host");
}
_ => panic!("unexpected declared label decode description"),
}
assert_eq!(source.polls.load(Ordering::SeqCst), 0);
let chunks = futures::TryStreamExt::try_collect::<Vec<_>>(reader)
.await
+3 -18
View File
@@ -57,7 +57,6 @@ use datafusion_physical_expr::{
use datatypes::arrow::datatypes::SchemaRef as ArrowSchemaRef;
use datatypes::compute::SortOptions;
use datatypes::series::SeriesOrdering;
use datatypes::series::frame::RawLabelMode;
use datatypes::series::samples::SeriesSchema;
use futures::{Stream, StreamExt};
use store_api::metric_engine_consts::DATA_SCHEMA_TSID_COLUMN_NAME;
@@ -593,11 +592,7 @@ impl RegionScanExec {
return Ok(None);
};
let ordering = scanner.raw_series_ordering();
let arrow_schema = raw_series_frame_schema_with_ordering(
&schema,
raw_label_mode(&label_source),
ordering,
)?;
let arrow_schema = raw_series_frame_schema_with_ordering(&schema, &label_source, ordering)?;
let partition_count = scanner.properties().num_partitions().max(1);
let properties = Arc::new(PlanProperties::new(
EquivalenceProperties::new(arrow_schema.clone()),
@@ -667,13 +662,6 @@ impl SeriesScanExec {
}
}
fn raw_label_mode(source: &RawLabelSource) -> RawLabelMode {
match source {
RawLabelSource::PrimaryKey { .. } => RawLabelMode::Key,
RawLabelSource::Explicit => RawLabelMode::Explicit,
}
}
fn same_label_source(left: &RawLabelSource, right: &RawLabelSource) -> bool {
match (left, right) {
(RawLabelSource::Explicit, RawLabelSource::Explicit) => true,
@@ -785,11 +773,8 @@ impl ExecutionPlan for SeriesScanExec {
DataFusionError::Execution("raw series capability changed after planning".into())
})?;
let current_ordering = scanner.raw_series_ordering();
let current_frame = raw_series_frame_schema_with_ordering(
&current,
raw_label_mode(&self.label_source),
current_ordering,
)?;
let current_frame =
raw_series_frame_schema_with_ordering(&current, &self.label_source, current_ordering)?;
if current_frame != self.arrow_schema
|| scanner.properties().num_partitions().max(1) != partitions
{
+13 -5
View File
@@ -36,6 +36,7 @@ use datatypes::data_type::ConcreteDataType;
use datatypes::error::InvalidSeriesSnafu;
use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
use datatypes::series::SeriesOrdering;
use datatypes::series::frame::{LabelDecodeSpec, RawLabelMode};
use datatypes::series::raw::{RawSeriesChunk, RawTimestamps};
use datatypes::series::samples::{FieldSpec, Samples, SeriesFields, SeriesSchema};
use datatypes::vectors::{Int32Vector, TimestampMillisecondVector, VectorRef};
@@ -82,6 +83,8 @@ fn raw_error(reason: &str) -> BoxedError {
struct RawStream {
schema: Arc<SeriesSchema>,
source: RawLabelSource,
/// The description this fixture declares, derived once from `source`, as a producer does.
label_decode_spec: Option<LabelDecodeSpec>,
chunks: Vec<RawSeriesChunk>,
state: Arc<Mutex<State>>,
fail_first_poll: bool,
@@ -126,6 +129,9 @@ impl RawSeriesStream for RawStream {
fn label_source(&self) -> &RawLabelSource {
&self.source
}
fn label_decode_spec(&self) -> Option<&LabelDecodeSpec> {
self.label_decode_spec.as_ref()
}
fn ordering(&self) -> SeriesOrdering {
self.ordering
}
@@ -211,16 +217,18 @@ impl RegionScanner for TestScanner {
return Ok(None);
}
let ordering = state.stream_ordering.unwrap_or(SeriesOrdering::KeyThenTime);
let source = state
.stream_source
.clone()
.unwrap_or_else(|| source.clone());
Ok(Some(Box::pin(RawStream {
ordering,
schema: state
.stream_schema
.clone()
.unwrap_or_else(|| self.raw_schema.clone().unwrap()),
source: state
.stream_source
.clone()
.unwrap_or_else(|| source.clone()),
label_decode_spec: common_query::series_frame::label_decode_spec(&source),
source,
chunks: self.chunks.clone(),
state: self.state.clone(),
fail_first_poll: state.fail_first_poll,
@@ -723,7 +731,7 @@ async fn supported_empty_stream_remains_native_with_declared_frame_schema() {
plan.schema(),
raw_series_frame_schema_with_ordering(
plan.series_schema(),
RawLabelMode::Explicit,
&RawLabelSource::Explicit,
SeriesOrdering::KeyThenTime
)
.unwrap()