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

refactor(series): unify raw and evaluated SeriesFrame codec

Browse Source 
SeriesFrame is the internal transport of the Narthex native PromQL path, and
it could only carry raw samples: an evaluated result had to fall back to a
plain row table. This makes one self-describing Arrow format carry both, so a
receiver decodes a frame from the frame alone, without the scanner, region, or
planning context its producer had.

What is unified is the physical representation and its codec, not the runtime
data structures. RawSeriesChunk, Series, and EvaluatedChunk stay as they are;
only the adapter boundaries branch on the sample shape.

Three combinations are legal and every other one is refused in both
directions:

  raw + key + timestamps          a storage raw-series scan
  raw + explicit + timestamps     a raw input whose labels are decoded
  evaluated + explicit + grid_spans   a compute node's output

An evaluated frame cannot claim a storage key, because the computation may
rewrite its labels; raw samples have no grid to be addressed on. The schema
builder and the declaration parser both enforce that, so a frame cannot be
written or read in a shape no producer could mean.

An evaluated frame declares its GridSpec once in schema metadata and carries
absolute grid positions per row as spans, so a sparse grid costs its runs and
not its holes: nothing is expanded, and no position a chunk does not hold is
filled in. A chunk with no position is legal exactly when it closes its chain,
and it is kept as one frame row, because dropping it would drop the finality
it exists to carry.

Layout is v2 and the writer and reader of the reserved greptime.series.*
metadata live in one place each, so a producer and a consumer cannot spell one
frame two ways. The unreleased v1 format is not read: the in-tree producers,
consumers, tests, and the two frozen wire fixtures all move to v2 together.

Signed-off-by: discord9 <discord9@users.noreply.github.com>
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
(cherry picked from commit 176f6f6479)
This commit is contained in:
discord9
2026-10-26 12:57:11 +08:00
parent 771347df5d
commit b20e22b426
19 changed files with 4230 additions and 1017 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/query/src/series_frame.rs
+534 -125
View File
@@ -27,8 +27,12 @@
//! 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.
//! [`SeriesOrdering::Contiguous`] and [`SeriesOrdering::KeyThenTime`] are accepted for raw frames and
//! only `Contiguous` for evaluated ones; `Interleaved` is rejected before any poll.
//!
//! An evaluated frame is the same physical family: [`pack_evaluated_series`] and
//! [`unpack_evaluated_series`] carry computed values under complete labels and grid positions, so a
//! compute consumer needs neither a storage key nor a scan binding to read them.
use std::pin::Pin;
use std::sync::Arc;
@@ -40,13 +44,12 @@ 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_label_decode, raw_series_frame_schema_with_label_decode,
raw_series_frame_series_schema,
FrameContent, FrameLabelMode, LABEL_DECODE_VERSION, LabelDecodeColumn, LabelDecodeEncoding,
LabelDecodeSpec, LabelDecodeTag, SeriesFrameBuilder, SeriesFrameCursor, SeriesFrameDeclaration,
SeriesFrameDecoder, series_frame_label_decode, series_frame_schema,
};
use datatypes::series::raw::RawSeriesChunk;
use datatypes::series::samples::SeriesSchema;
use datatypes::series::samples::{Series, SeriesSchema};
use datatypes::series::{SeriesLabels, SeriesOrdering};
use futures_util::Stream;
use store_api::codec::PrimaryKeyEncoding;
@@ -63,10 +66,19 @@ mod tests;
const ORDERING_KEY: &str = "greptime.series.ordering";
/// A stream of evaluated series chunks and their finality.
///
/// The pair is what an evaluated producer has and an evaluated consumer needs: the values on their
/// absolute grid positions, and whether this chunk closed the chain it belongs to. It is deliberately
/// not a third runtime chunk type - `Series` is the runtime value, and finality is the one fact a
/// chunk has to carry besides it.
pub type SendableEvaluatedSeriesStream =
Pin<Box<dyn Stream<Item = Result<(Series, bool), BoxedError>> + Send>>;
/// 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. The label binding
/// has no 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.
@@ -75,6 +87,11 @@ pub fn raw_series_frame_schema_with_ordering(
label_source: &RawLabelSource,
ordering: SeriesOrdering,
) -> datafusion_common::Result<ArrowSchemaRef> {
if schema.eval_grid.is_some() {
return Err(invalid(
"a raw frame cannot carry a schema with an evaluation grid",
));
}
frame_schema_with(
schema,
label_mode(label_source),
@@ -85,26 +102,30 @@ pub fn raw_series_frame_schema_with_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.
/// The mode and description pair is checked by [`series_frame_schema`], 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,
mode: FrameLabelMode,
decode: Option<&LabelDecodeSpec>,
ordering: SeriesOrdering,
) -> datafusion_common::Result<ArrowSchemaRef> {
// An evaluated frame has no storage key to be ordered by, so the only claim it can honour is
// that each chain is contiguous. Writing a stronger one would produce a frame the reader refuses
// and a plan that fails where it could have been refused.
if schema.eval_grid.is_some() {
ensure_contiguous_ordering(ordering)?;
}
let ordering = match ordering {
SeriesOrdering::Contiguous => "contiguous",
SeriesOrdering::KeyThenTime => "read_key_then_time",
SeriesOrdering::Interleaved => {
return Err(invalid(
"interleaved raw series ordering has no wire representation",
"interleaved series ordering has no wire representation",
));
}
};
let schema =
raw_series_frame_schema_with_label_decode(schema, mode, decode).map_err(codec_error)?;
let schema = series_frame_schema(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(
@@ -163,28 +184,34 @@ pub fn label_decode_spec(source: &RawLabelSource) -> Option<LabelDecodeSpec> {
Some(description)
}
/// Reads the declared producer ordering, conservatively treating legacy frames as contiguous.
/// Reads the declared producer ordering, treating an absent declaration as `contiguous`.
///
/// Only the two orderings representable on the wire can be returned.
pub fn raw_series_frame_ordering(
schema: &ArrowSchema,
) -> datafusion_common::Result<SeriesOrdering> {
/// The absent case is deliberately the weakest guarantee rather than an error, and it is safe for
/// exactly that reason: `contiguous` says only that each series' chunks are not interleaved with
/// another's, so adopting it for a frame that never said anything can lose an optimization but
/// cannot accept a stream whose real ordering is weaker. A producer that does make a stronger claim
/// has to state it, and a producer that cannot keep one is refused: `Interleaved` is not a
/// representable value, and neither is anything this revision does not know.
///
/// This is also what the frame's own declaration has always meant, which is what keeps a reader of
/// `v1` and `v2` text agreeing about the same bytes.
pub fn series_frame_ordering(schema: &ArrowSchema) -> datafusion_common::Result<SeriesOrdering> {
match schema.metadata().get(ORDERING_KEY).map(String::as_str) {
None | Some("contiguous") => Ok(SeriesOrdering::Contiguous),
Some("read_key_then_time") => Ok(SeriesOrdering::KeyThenTime),
Some(_) => Err(invalid("unknown raw series ordering")),
Some(_) => Err(invalid("unknown series ordering")),
}
}
/// Packs a storage raw-series stream into DataFusion record batches incrementally.
///
/// Only [`SeriesOrdering::Contiguous`] and [`SeriesOrdering::KeyThenTime`] streams are
/// accepted; an `Interleaved` declaration is rejected before the first poll, including for an
/// empty stream. Rows accumulate across chunks into batches bounded by the builder's row and
/// estimated payload limits; one legal oversized row may be emitted alone, and these limits are a
/// 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.
/// Only [`SeriesOrdering::Contiguous`] and [`SeriesOrdering::KeyThenTime`] streams are accepted; an
/// `Interleaved` declaration is rejected before the first poll, including for an empty stream. Rows
/// accumulate across chunks into batches bounded by the builder's row and estimated payload limits;
/// one legal oversized row may be emitted alone, and these limits are a 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
@@ -197,27 +224,71 @@ pub fn pack_raw_series(
ensure_supported_ordering(ordering)?;
let schema = source.schema().clone();
let decode = source.label_decode_spec().cloned();
let mode = declared_label_mode(&decode);
let labels = 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(
// every ready batch is published under it, including when no row is ever produced.
let frame_schema = frame_schema_with(&schema, labels, decode.as_ref(), ordering)?;
let builder = SeriesFrameBuilder::try_new(
&schema,
mode,
RawSeriesFrameBuilder::DEFAULT_MAX_ROWS,
RawSeriesFrameBuilder::DEFAULT_MAX_BYTES,
labels,
decode.as_ref(),
SeriesFrameBuilder::DEFAULT_MAX_ROWS,
SeriesFrameBuilder::DEFAULT_MAX_BYTES,
)
.map_err(codec_error)?;
Ok(Box::pin(PackStream {
source: Some(source),
schema,
mode,
frame_schema,
labels,
batches: PackBatches::new(builder, frame_schema),
ordering,
validator: SequenceValidator::new(ordering),
builder: Some(builder),
cursor: None,
}))
}
/// Packs a stream of evaluated chunks into DataFusion record batches incrementally.
///
/// An evaluated frame names its series by their complete labels and addresses time on the evaluation
/// grid `schema` declares, so there is nothing to decode and no storage key to carry: the frame
/// declares `explicit` identity and `grid_spans` time, and the grid is written once instead of
/// repeating timestamps on every row.
///
/// Only [`SeriesOrdering::Contiguous`] is declared, because that is the one ordering this layer can
/// promise and check: each chunk chain is contiguous with itself. A caller that has reason to claim
/// [`SeriesOrdering::KeyThenTime`] about computed values would have to justify it, and a consumer
/// must never infer it from the presence of labels.
///
/// Bounds, backpressure, and error handling are the raw path's: rows accumulate into batches bounded
/// by the builder's row and estimated payload limits, a ready batch is yielded before the source is
/// polled again, and a `Pending` source still delivers what is buffered.
pub fn pack_evaluated_series(
source: SendableEvaluatedSeriesStream,
schema: Arc<SeriesSchema>,
) -> datafusion_common::Result<DfSendableRecordBatchStream> {
if schema.eval_grid.is_none() {
return Err(invalid(
"an evaluated frame needs a schema with an evaluation grid",
));
}
let frame_schema = frame_schema_with(
&schema,
FrameLabelMode::Explicit,
None,
SeriesOrdering::Contiguous,
)?;
let builder = SeriesFrameBuilder::try_new(
&schema,
FrameLabelMode::Explicit,
None,
SeriesFrameBuilder::DEFAULT_MAX_ROWS,
SeriesFrameBuilder::DEFAULT_MAX_BYTES,
)
.map_err(codec_error)?;
Ok(Box::pin(PackEvaluatedStream {
source: Some(source),
schema,
batches: PackBatches::new(builder, frame_schema),
validator: EvaluatedSequenceValidator::new(),
}))
}
@@ -233,31 +304,40 @@ pub fn pack_raw_series(
///
/// 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
/// 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.
/// A frame that declares evaluated content is refused here, before anything is read: a raw consumer
/// reads timestamps rather than grid positions, so the two contents are not interchangeable.
///
/// `expected` is a check, never an input. When it is given it must name the same logical fields in
/// the same order and the same evaluation grid as the frame declares - which for a raw frame is that
/// both sides carry none - 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: Option<Arc<SeriesSchema>>,
) -> datafusion_common::Result<SendableRawSeriesStream> {
let frame_schema = source.schema();
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)?);
let declaration = SeriesFrameDeclaration::parse(frame_schema.as_ref()).map_err(codec_error)?;
if declaration.content() != FrameContent::Raw {
return Err(invalid("expected a raw frame, found an evaluated one"));
}
let labels = declaration.labels();
let declared = declaration.label_decode().cloned();
let schema = Arc::new(declaration.series().clone());
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)?;
let ordering = series_frame_ordering(frame_schema.as_ref())?;
ensure_supported_ordering(ordering)?;
let decoder =
SeriesFrameDecoder::try_new(&schema, labels, declared.as_ref()).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)?;
decoder.decode_raw(&empty).map_err(codec_error)?;
Ok(Box::pin(UnpackStream {
source: Some(source),
schema,
@@ -270,20 +350,53 @@ pub fn unpack_raw_series(
}))
}
/// Unpacks DataFusion batches declared by [`pack_evaluated_series`].
///
/// The frame declares its grid, its field list, and its label mode, so the returned stream carries
/// the [`SeriesSchema`] the rows are addressed on and needs no planning-side copy of it. The grid is
/// the frame's own declaration: the timestamps stay unmaterialized until a consumer asks for one.
///
/// A frame that declares raw samples, keyed identity, or an ordering other than `contiguous` is
/// refused before the stream is returned, as is a grid the declaration check cannot make sense of.
/// `expected` is a check and never an input, exactly as on the raw path: when given it must name the
/// same fields and the same grid the frame declares.
pub fn unpack_evaluated_series(
source: DfSendableRecordBatchStream,
expected: Option<Arc<SeriesSchema>>,
) -> datafusion_common::Result<SendableEvaluatedSeriesStream> {
let frame_schema = source.schema();
let declaration = SeriesFrameDeclaration::parse(frame_schema.as_ref()).map_err(codec_error)?;
if declaration.content() != FrameContent::Evaluated {
return Err(invalid("expected an evaluated frame, found a raw one"));
}
ensure_contiguous_ordering(series_frame_ordering(frame_schema.as_ref())?)?;
let schema = Arc::new(declaration.series().clone());
if let Some(expected) = &expected {
ensure_declared_schema(&schema, expected)?;
}
let decoder = SeriesFrameDecoder::try_new(&schema, FrameLabelMode::Explicit, None)
.map_err(codec_error)?;
let empty = DfRecordBatch::new_empty(frame_schema.clone());
decoder.decode_evaluated(&empty).map_err(codec_error)?;
Ok(Box::pin(UnpackEvaluatedStream {
source: Some(source),
frame_schema,
validator: EvaluatedSequenceValidator::new(),
pending: Vec::new().into_iter(),
decoder,
}))
}
/// 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.
/// [`SeriesSchema`] holds: the logical fields and their order, and the evaluation grid. A raw frame
/// carries no grid and an evaluated one carries exactly its own, so the grid comparison is also what
/// keeps a raw expectation from decoding an evaluated frame.
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
@@ -292,10 +405,16 @@ fn ensure_declared_schema(
.any(|(expected, declared)| expected.name != declared.name)
{
return Err(invalid(
"raw frame fields do not match the expected series schema",
"frame fields do not match the expected series schema",
));
}
Ok(())
match (&expected.eval_grid, &declared.eval_grid) {
(None, None) => Ok(()),
(Some(expected), Some(declared)) if expected.spec() == declared.spec() => Ok(()),
_ => Err(invalid(
"frame evaluation grid does not match the expected series schema",
)),
}
}
/// Checks that a frame declares the label decode layout a scan-bound binding describes.
@@ -313,15 +432,14 @@ fn ensure_declared_schema(
/// [`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.
/// The comparison is plain equality. The writer canonicalizes the description (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)?;
let declared = 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",
@@ -330,19 +448,19 @@ pub fn ensure_frame_matches_label_source(
Ok(())
}
fn label_mode(source: &RawLabelSource) -> RawLabelMode {
fn label_mode(source: &RawLabelSource) -> FrameLabelMode {
match source {
RawLabelSource::PrimaryKey { .. } => RawLabelMode::Key,
RawLabelSource::Explicit => RawLabelMode::Explicit,
RawLabelSource::PrimaryKey { .. } => FrameLabelMode::Key,
RawLabelSource::Explicit => FrameLabelMode::Explicit,
}
}
/// 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 {
fn declared_label_mode(declared: &Option<LabelDecodeSpec>) -> FrameLabelMode {
match declared {
Some(_) => RawLabelMode::Key,
None => RawLabelMode::Explicit,
Some(_) => FrameLabelMode::Key,
None => FrameLabelMode::Explicit,
}
}
@@ -356,6 +474,21 @@ fn ensure_supported_ordering(ordering: SeriesOrdering) -> datafusion_common::Res
}
}
/// Rejects an evaluated frame that claims more order than its producer promised.
///
/// Computed values have no storage key to be ordered by, so `read_key_then_time` is not a claim this
/// layer can honour; `Interleaved` is not consumable at all. Treating either as `contiguous` would
/// silently weaken a guarantee a consumer may rely on.
fn ensure_contiguous_ordering(ordering: SeriesOrdering) -> datafusion_common::Result<()> {
match ordering {
SeriesOrdering::Contiguous => Ok(()),
SeriesOrdering::KeyThenTime => Err(invalid(
"an evaluated frame cannot claim key-then-time ordering",
)),
SeriesOrdering::Interleaved => Err(invalid("interleaved series ordering is not supported")),
}
}
fn codec_error(error: datatypes::error::Error) -> DataFusionError {
DataFusionError::External(Box::new(BoxedError::new(error)))
}
@@ -373,9 +506,10 @@ fn invalid(reason: impl Into<String>) -> DataFusionError {
)))
}
/// Cross-chunk validation state. The producer must keep each series' chunks contiguous; this
/// validator checks adjacent chunks only and never sees `Interleaved` streams. It retains only
/// the previous complete owned key (not an Arrow buffer), timestamp, metric, and finality.
/// Cross-chunk validation state for raw chains. The producer must keep each series' chunks
/// contiguous; this validator checks adjacent chunks only and never sees `Interleaved` streams. It
/// retains only the previous complete owned key (not an Arrow buffer), timestamp, metric, and
/// finality.
struct SequenceValidator {
ordering: SeriesOrdering,
previous_key: Option<Vec<u8>>,
@@ -447,30 +581,95 @@ impl SequenceValidator {
}
}
struct PackStream {
source: Option<SendableRawSeriesStream>,
schema: Arc<SeriesSchema>,
mode: RawLabelMode,
frame_schema: datatypes::arrow::datatypes::SchemaRef,
ordering: SeriesOrdering,
validator: SequenceValidator,
builder: Option<RawSeriesFrameBuilder>,
cursor: Option<RawSeriesChunkCursor>,
/// Cross-chunk validation state for evaluated chains.
///
/// `is_last` closes one chunk chain and nothing more. It is not the end of a label set: a range
/// function or a label rewrite can make two input series show the same complete labels, and each of
/// them closes itself, so a closed chain may be followed by another chain whose labels happen to be
/// equal. Only the still-open chain is remembered, which is what keeps such a contribution from being
/// refused as a re-opened series while still refusing a chain that changes identity before it closes.
struct EvaluatedSequenceValidator {
/// The labels of the open chain and the last grid position it carried.
open: Option<(SeriesLabels, usize)>,
}
impl PackStream {
/// Releases the source and drops every buffered owner, leaving the stream terminal; batches
/// already yielded are unaffected.
fn finish(&mut self) {
self.source = None;
self.cursor = None;
self.builder = None;
self.validator = SequenceValidator::new(self.ordering);
impl EvaluatedSequenceValidator {
fn new() -> Self {
Self { open: None }
}
fn finish(&self) -> datafusion_common::Result<()> {
if self.open.is_some() {
return Err(invalid(
"evaluated series stream ended before a chain closed",
));
}
Ok(())
}
fn validate(&mut self, series: &Series, is_last: bool) -> datafusion_common::Result<()> {
let (Some(first), Some(last)) = (series.time.first_position(), series.time.last_position())
else {
// No carried position: only a closing chunk may be empty, and it closes the chain it
// belongs to without advancing it.
if !is_last {
return Err(invalid(
"an evaluated chunk without positions must close its series",
));
}
match &self.open {
Some((labels, _)) if labels == &series.metric => self.open = None,
Some(_) => {
return Err(invalid(
"evaluated series identity changed before its chain closed",
));
}
None => {}
}
return Ok(());
};
if let Some((labels, previous)) = &self.open {
if labels != &series.metric {
return Err(invalid(
"evaluated series identity changed before its chain closed",
));
}
if first <= *previous {
return Err(invalid(
"evaluated series positions are not increasing across chunks",
));
}
}
self.open = (!is_last).then(|| (series.metric.clone(), last));
Ok(())
}
}
/// The batching half of a pack stream: the builder, the cursor installed in it, and the declared
/// schema every ready batch is published under.
struct PackBatches {
frame_schema: ArrowSchemaRef,
builder: Option<SeriesFrameBuilder>,
cursor: Option<SeriesFrameCursor>,
}
impl PackBatches {
fn new(builder: SeriesFrameBuilder, frame_schema: ArrowSchemaRef) -> Self {
Self {
frame_schema,
builder: Some(builder),
cursor: None,
}
}
/// Installs the cursor of the chunk just read, replacing whatever was left of the previous one.
fn install(&mut self, cursor: SeriesFrameCursor) {
self.cursor = Some(cursor);
}
/// Advances the installed cursor until it yields one ready batch or is exhausted, dropping an
/// exhausted cursor. `None` means no batch is ready, so the source may be polled again.
fn advance_cursor(&mut self) -> datatypes::error::Result<Option<DfRecordBatch>> {
fn advance(&mut self) -> datatypes::error::Result<Option<DfRecordBatch>> {
let ready = match (self.cursor.as_mut(), self.builder.as_mut()) {
(Some(cursor), Some(builder)) => cursor.append_to(builder)?,
_ => return Ok(None),
@@ -485,38 +684,63 @@ impl PackStream {
}
/// Flushes the rows buffered so far, if any, so a pending source can still deliver data.
fn flush_buffered(&mut self) -> datatypes::error::Result<Option<DfRecordBatch>> {
fn flush(&mut self) -> datatypes::error::Result<Option<DfRecordBatch>> {
match self.builder.as_mut() {
Some(builder) => builder.flush(),
None => Ok(None),
}
}
/// Publishes one ready builder batch under the declared frame schema, which is the shared
/// builder's schema plus the producer's ordering metadata. A batch whose fields cannot be
/// declared releases the stream instead of yielding rows with an undeclared layout.
fn declare_batch(
/// Releases the buffered owners, leaving the stream terminal; batches already yielded are
/// unaffected.
fn release(&mut self) {
self.cursor = None;
self.builder = None;
}
/// Publishes one ready batch under the declared frame schema, which is the builder's own layout
/// plus the producer's ordering metadata. A batch whose fields cannot be declared releases the
/// stream instead of yielding rows with an undeclared layout.
fn declare(
&mut self,
batch: DfRecordBatch,
) -> Poll<Option<datafusion_common::Result<DfRecordBatch>>> {
match batch.with_schema(self.frame_schema.clone()) {
Ok(batch) => Poll::Ready(Some(Ok(batch))),
Err(error) => {
self.finish();
self.release();
Poll::Ready(Some(Err(DataFusionError::from(error))))
}
}
}
}
struct PackStream {
source: Option<SendableRawSeriesStream>,
schema: Arc<SeriesSchema>,
labels: FrameLabelMode,
batches: PackBatches,
ordering: SeriesOrdering,
validator: SequenceValidator,
}
impl PackStream {
/// Releases the source and drops every buffered owner, leaving the stream terminal.
fn finish(&mut self) {
self.source = None;
self.batches.release();
self.validator = SequenceValidator::new(self.ordering);
}
}
impl Stream for PackStream {
type Item = datafusion_common::Result<DfRecordBatch>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let this = &mut *self;
loop {
match this.advance_cursor() {
Ok(Some(batch)) => return this.declare_batch(batch),
match this.batches.advance() {
Ok(Some(batch)) => return this.batches.declare(batch),
Ok(None) => {}
Err(error) => {
this.finish();
@@ -529,9 +753,9 @@ impl Stream for PackStream {
};
match polled {
Poll::Pending => {
let flushed = this.flush_buffered();
let flushed = this.batches.flush();
return match flushed {
Ok(Some(batch)) => this.declare_batch(batch),
Ok(Some(batch)) => this.batches.declare(batch),
Ok(None) => Poll::Pending,
Err(error) => {
this.finish();
@@ -544,10 +768,16 @@ impl Stream for PackStream {
this.finish();
return Poll::Ready(Some(Err(error)));
}
let flushed = this.flush_buffered();
let flushed = this.batches.flush();
// The declared schema is read before the release below, so the last batch is
// published under the same schema as every batch before it.
let declared = this.batches.frame_schema.clone();
this.finish();
return match flushed {
Ok(Some(batch)) => this.declare_batch(batch),
Ok(Some(batch)) => match batch.with_schema(declared) {
Ok(batch) => Poll::Ready(Some(Ok(batch))),
Err(error) => Poll::Ready(Some(Err(DataFusionError::from(error)))),
},
Ok(None) => Poll::Ready(None),
Err(error) => Poll::Ready(Some(Err(codec_error(error)))),
};
@@ -561,8 +791,8 @@ impl Stream for PackStream {
this.finish();
return Poll::Ready(Some(Err(error)));
}
match RawSeriesChunkCursor::try_new(&this.schema, this.mode, chunk) {
Ok(cursor) => this.cursor = Some(cursor),
match SeriesFrameCursor::try_new_raw(&this.schema, this.labels, chunk) {
Ok(cursor) => this.batches.install(cursor),
Err(error) => {
this.finish();
return Poll::Ready(Some(Err(codec_error(error))));
@@ -575,21 +805,115 @@ impl Stream for PackStream {
}
impl datafusion::physical_plan::RecordBatchStream for PackStream {
fn schema(&self) -> datatypes::arrow::datatypes::SchemaRef {
self.frame_schema.clone()
fn schema(&self) -> ArrowSchemaRef {
self.batches.frame_schema.clone()
}
}
/// The pack stream of [`pack_evaluated_series`].
struct PackEvaluatedStream {
source: Option<SendableEvaluatedSeriesStream>,
schema: Arc<SeriesSchema>,
batches: PackBatches,
validator: EvaluatedSequenceValidator,
}
impl PackEvaluatedStream {
/// Releases the source and drops every buffered owner, leaving the stream terminal.
fn finish(&mut self) {
self.source = None;
self.batches.release();
self.validator = EvaluatedSequenceValidator::new();
}
}
impl Stream for PackEvaluatedStream {
type Item = datafusion_common::Result<DfRecordBatch>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let this = &mut *self;
loop {
match this.batches.advance() {
Ok(Some(batch)) => return this.batches.declare(batch),
Ok(None) => {}
Err(error) => {
this.finish();
return Poll::Ready(Some(Err(codec_error(error))));
}
}
let polled = match this.source.as_mut() {
Some(source) => source.as_mut().poll_next(cx),
None => return Poll::Ready(None),
};
match polled {
Poll::Pending => {
let flushed = this.batches.flush();
return match flushed {
Ok(Some(batch)) => this.batches.declare(batch),
Ok(None) => Poll::Pending,
Err(error) => {
this.finish();
Poll::Ready(Some(Err(codec_error(error))))
}
};
}
Poll::Ready(None) => {
if let Err(error) = this.validator.finish() {
this.finish();
return Poll::Ready(Some(Err(error)));
}
let flushed = this.batches.flush();
// The declared schema is read before the release below, so the last batch is
// published under the same schema as every batch before it.
let declared = this.batches.frame_schema.clone();
this.finish();
return match flushed {
Ok(Some(batch)) => match batch.with_schema(declared) {
Ok(batch) => Poll::Ready(Some(Ok(batch))),
Err(error) => Poll::Ready(Some(Err(DataFusionError::from(error)))),
},
Ok(None) => Poll::Ready(None),
Err(error) => Poll::Ready(Some(Err(codec_error(error)))),
};
}
Poll::Ready(Some(Err(error))) => {
this.finish();
return Poll::Ready(Some(Err(upstream_error(error))));
}
Poll::Ready(Some(Ok((series, is_last)))) => {
if let Err(error) = this.validator.validate(&series, is_last) {
this.finish();
return Poll::Ready(Some(Err(error)));
}
match SeriesFrameCursor::try_new_evaluated(&this.schema, series, is_last) {
Ok(cursor) => this.batches.install(cursor),
Err(error) => {
this.finish();
return Poll::Ready(Some(Err(codec_error(error))));
}
}
}
}
}
}
}
impl datafusion::physical_plan::RecordBatchStream for PackEvaluatedStream {
fn schema(&self) -> ArrowSchemaRef {
self.batches.frame_schema.clone()
}
}
struct UnpackStream {
source: Option<DfSendableRecordBatchStream>,
schema: Arc<SeriesSchema>,
frame_schema: datatypes::arrow::datatypes::SchemaRef,
frame_schema: ArrowSchemaRef,
/// 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>,
decoder: RawSeriesFrameDecoder,
decoder: SeriesFrameDecoder,
}
impl RawSeriesStream for UnpackStream {
@@ -628,7 +952,6 @@ impl Stream for UnpackStream {
type Item = Result<RawSeriesChunk, BoxedError>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
const MAX_EMPTY_BATCHES_PER_POLL: usize = 32;
let this = &mut *self;
let mut empty_batches = 0;
loop {
@@ -644,12 +967,7 @@ impl Stream for UnpackStream {
Poll::Ready(None) => {
let result = this.validator.finish();
this.finish();
return Poll::Ready(result.err().map(|error| {
Err(BoxedError::new(crate::error::Error::GeneralDataFusion {
error,
location: snafu::Location::default(),
}))
}));
return Poll::Ready(result.err().map(|error| Err(unpack_error(error))));
}
Poll::Ready(Some(Err(error))) => {
this.finish();
@@ -666,7 +984,7 @@ impl Stream for UnpackStream {
return Err(invalid("raw frame schema mismatch"));
}
this.decoder
.decode(&batch)
.decode_raw(&batch)
.map_err(codec_error)
.and_then(|chunks| {
for chunk in &chunks {
@@ -689,12 +1007,7 @@ impl Stream for UnpackStream {
}
Err(error) => {
this.finish();
return Poll::Ready(Some(Err(BoxedError::new(
crate::error::Error::GeneralDataFusion {
error,
location: snafu::Location::default(),
},
))));
return Poll::Ready(Some(Err(unpack_error(error))));
}
}
}
@@ -702,3 +1015,99 @@ impl Stream for UnpackStream {
}
}
}
/// The unpack stream of [`unpack_evaluated_series`].
struct UnpackEvaluatedStream {
source: Option<DfSendableRecordBatchStream>,
frame_schema: ArrowSchemaRef,
validator: EvaluatedSequenceValidator,
pending: std::vec::IntoIter<(Series, bool)>,
decoder: SeriesFrameDecoder,
}
impl UnpackEvaluatedStream {
fn finish(&mut self) {
self.source = None;
self.pending = Vec::new().into_iter();
self.validator = EvaluatedSequenceValidator::new();
}
}
impl Stream for UnpackEvaluatedStream {
type Item = Result<(Series, bool), BoxedError>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let this = &mut *self;
let mut empty_batches = 0;
loop {
if let Some(chunk) = this.pending.next() {
return Poll::Ready(Some(Ok(chunk)));
}
let polled = match this.source.as_mut() {
Some(source) => source.as_mut().poll_next(cx),
None => return Poll::Ready(None),
};
match polled {
Poll::Pending => return Poll::Pending,
Poll::Ready(None) => {
let result = this.validator.finish();
this.finish();
return Poll::Ready(result.err().map(|error| Err(unpack_error(error))));
}
Poll::Ready(Some(Err(error))) => {
this.finish();
return Poll::Ready(Some(Err(BoxedError::new(
crate::error::Error::GeneralDataFusion {
error,
location: snafu::Location::default(),
},
))));
}
Poll::Ready(Some(Ok(batch))) => {
let result = (|| {
if batch.schema() != this.frame_schema {
return Err(invalid("evaluated frame schema mismatch"));
}
this.decoder
.decode_evaluated(&batch)
.map_err(codec_error)
.and_then(|chunks| {
for (series, is_last) in &chunks {
this.validator.validate(series, *is_last)?;
}
Ok(chunks)
})
})();
match result {
Ok(chunks) => {
if chunks.is_empty() {
empty_batches += 1;
if empty_batches == MAX_EMPTY_BATCHES_PER_POLL {
cx.waker().wake_by_ref();
return Poll::Pending;
}
} else {
this.pending = chunks.into_iter();
}
}
Err(error) => {
this.finish();
return Poll::Ready(Some(Err(unpack_error(error))));
}
}
}
}
}
}
}
/// How many empty batches one poll reads before yielding, so a stream of them cannot hold the task.
const MAX_EMPTY_BATCHES_PER_POLL: usize = 32;
/// Wraps a decode or validation error as the boxed error an unpack stream reports.
fn unpack_error(error: DataFusionError) -> BoxedError {
BoxedError::new(crate::error::Error::GeneralDataFusion {
error,
location: snafu::Location::default(),
})
}
src/common/query/src/series_frame/tests.rs
+267 -96
View File
@@ -12,6 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
mod evaluated;
mod lifecycle;
use std::collections::{HashMap, VecDeque};
@@ -34,7 +35,9 @@ use datatypes::arrow::record_batch::RecordBatch;
use datatypes::series::Label;
use datatypes::series::SeriesLabels;
use datatypes::series::SeriesOrdering;
use datatypes::series::frame::{LabelDecodeSpec, RawLabelMode, raw_series_frame_schema};
use datatypes::series::frame::{
FrameLabelMode, LabelDecodeSpec, raw_series_frame_schema, series_frame_schema,
};
use datatypes::series::histogram::{BucketSideView, CounterResetHint, HistogramBatchBuilder};
use datatypes::series::raw::{RawSeriesChunk, RawTimestamps};
use datatypes::series::samples::{
@@ -46,8 +49,8 @@ use store_api::region_engine::{
};
use super::{
ensure_frame_matches_label_source, label_decode_spec, pack_raw_series,
raw_series_frame_ordering, unpack_raw_series,
ensure_frame_matches_label_source, label_decode_spec, pack_raw_series, series_frame_ordering,
unpack_raw_series,
};
fn schema() -> Arc<SeriesSchema> {
@@ -59,6 +62,28 @@ fn schema() -> Arc<SeriesSchema> {
eval_grid: None,
})
}
/// The one grid every evaluated fixture is addressed on: 512 positions, 1000 ms apart.
fn evaluated_grid() -> Arc<datatypes::series::time::EvalGrid> {
Arc::new(
datatypes::series::time::EvalGrid::try_new(
datatypes::series::time::GridSpec::try_new(0, 1000, 512).unwrap(),
)
.unwrap(),
)
}
/// A schema with one `value` field addressed on [`evaluated_grid`], which is what makes it an
/// evaluated declaration.
fn evaluated_schema() -> Arc<SeriesSchema> {
Arc::new(SeriesSchema {
fields: vec![FieldSpec {
name: Arc::from("value"),
}]
.into(),
eval_grid: Some(evaluated_grid()),
})
}
fn metric() -> SeriesLabels {
SeriesLabels::try_new(Some(Arc::from("cpu")), vec![]).unwrap()
}
@@ -629,7 +654,8 @@ async fn invalid_sequence_reports_valid_prefix_then_one_terminal_error() {
assert_eq!(batch.num_rows(), case.prefix.len(), "{}", case.name);
let chunks = datatypes::series::frame::unpack_raw_series_frame(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
None,
&batch,
)
.unwrap();
@@ -656,7 +682,8 @@ async fn invalid_sequence_reports_valid_prefix_then_one_terminal_error() {
frames.extend(
datatypes::series::frame::pack_raw_series_chunk(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
None,
item,
)
.unwrap(),
@@ -692,7 +719,8 @@ async fn invalid_sequence_reports_valid_prefix_then_one_terminal_error() {
// missing final key is reported once, at EOF.
let frame = datatypes::series::frame::pack_raw_series_chunk(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
None,
&chunk(b"unfinished", &[7], false),
)
.unwrap();
@@ -923,7 +951,8 @@ async fn explicit_metric_must_remain_stable_within_key() {
);
let chunks = datatypes::series::frame::unpack_raw_series_frame(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
None,
&batch,
)
.unwrap();
@@ -975,7 +1004,7 @@ async fn empty_streams_keep_self_describing_schema() {
assert_eq!(
frame_schema.field(0).name(),
"read_key",
"the v1 wire column name is unchanged"
"the wire column name is unchanged"
);
assert!(packed.next().await.is_none());
// A declared-but-empty stream is still validated against the expected schema before any
@@ -1018,7 +1047,7 @@ async fn upstream_error_status_is_preserved_once_and_releases_source() {
assert_eq!(drops.load(Ordering::SeqCst), 1);
let batch_schema = datatypes::series::frame::raw_series_frame_schema(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
)
.unwrap();
let plain = PlainError::new("upstream df".into(), status);
@@ -1062,7 +1091,8 @@ async fn dropping_adapter_releases_upstream_and_pending_item_is_resumed() {
);
let chunks = datatypes::series::frame::unpack_raw_series_frame(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
None,
&first,
)
.unwrap();
@@ -1076,7 +1106,8 @@ async fn dropping_adapter_releases_upstream_and_pending_item_is_resumed() {
);
let chunks = datatypes::series::frame::unpack_raw_series_frame(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
None,
&second,
)
.unwrap();
@@ -1094,18 +1125,25 @@ async fn unpack_rejects_mismatched_batch_schema_before_polling() {
assert_eq!(drops.load(Ordering::SeqCst), 1);
}
/// A schema with an evaluation grid is a legal *evaluated* frame declaration, so it is no longer a
/// reason to refuse a stream at construction; it is still no schema to carry *raw* chunks under, so
/// the first raw chunk is rejected instead of the stream.
#[tokio::test]
async fn invalid_grid_schema_rejected_for_empty_raw_stream() {
let mut invalid = (*schema()).clone();
invalid.eval_grid = Some(Arc::new(
async fn grid_schema_is_legal_but_cannot_carry_a_raw_chunk() {
let mut grid_schema = (*schema()).clone();
grid_schema.eval_grid = Some(Arc::new(
datatypes::series::time::EvalGrid::try_new(
datatypes::series::time::GridSpec::try_new(0, 10, 1).unwrap(),
)
.unwrap(),
));
let grid_schema = Arc::new(grid_schema);
// An empty stream under a grid schema constructs: the declaration is legal and no row disagrees
// with it.
let polls = Arc::new(AtomicUsize::new(0));
let source = Box::pin(RawMock {
schema: Arc::new(invalid),
schema: grid_schema.clone(),
items: VecDeque::new(),
binding: RawLabelSource::Explicit,
label_decode_spec: None,
@@ -1117,8 +1155,33 @@ async fn invalid_grid_schema_rejected_for_empty_raw_stream() {
panic_on_poll: false,
emitted: 0,
});
assert!(pack_raw_series(source).is_err());
assert_eq!(polls.load(Ordering::SeqCst), 0);
let mut packed = pack_raw_series(source).unwrap();
assert!(packed.next().await.is_none());
assert_eq!(polls.load(Ordering::SeqCst), 1);
// A raw chunk cannot be addressed on a grid, so the chunk that arrives under this declaration is
// what fails.
let polls = Arc::new(AtomicUsize::new(0));
let source = Box::pin(RawMock {
schema: grid_schema,
items: VecDeque::from([Ok(chunk(b"a", &[1], true))]),
binding: RawLabelSource::Explicit,
label_decode_spec: None,
ordering: SeriesOrdering::Contiguous,
polls: polls.clone(),
drops: Arc::new(AtomicUsize::new(0)),
pending_after: None,
pending_reported: false,
panic_on_poll: false,
emitted: 0,
});
let mut packed = pack_raw_series(source).unwrap();
let error = packed.next().await.unwrap().unwrap_err();
assert_eq!(
crate::error::datafusion_status_code::<crate::error::Error>(&error, None),
StatusCode::InvalidArguments
);
assert!(packed.next().await.is_none());
}
#[tokio::test]
@@ -1150,19 +1213,21 @@ async fn unpack_batch_sequence_error_is_atomic_for_all_rows() {
async fn unpack_rejects_late_invalid_frame_row_before_prefix() {
let frame_schema = datatypes::series::frame::raw_series_frame_schema(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
)
.unwrap();
let first = datatypes::series::frame::pack_raw_series_chunk(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
None,
&chunk(b"a", &[1], true),
)
.unwrap()
.remove(0);
let mut second = datatypes::series::frame::pack_raw_series_chunk(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
None,
&chunk(b"b", &[2], true),
)
.unwrap()
@@ -1211,7 +1276,7 @@ async fn unpack_rejects_late_invalid_frame_row_before_prefix() {
async fn frame_application_metadata_is_stable_and_schema_change_rejected() {
let base_schema = datatypes::series::frame::raw_series_frame_schema(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
)
.unwrap();
let mut application_metadata = base_schema.metadata().clone();
@@ -1223,7 +1288,8 @@ async fn frame_application_metadata_is_stable_and_schema_change_rejected() {
);
let first = datatypes::series::frame::pack_raw_series_chunk(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
None,
&chunk(b"a", &[1], true),
)
.unwrap()
@@ -1232,7 +1298,8 @@ async fn frame_application_metadata_is_stable_and_schema_change_rejected() {
.unwrap();
let stable_second = datatypes::series::frame::pack_raw_series_chunk(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
None,
&chunk(b"b", &[2], true),
)
.unwrap()
@@ -1267,7 +1334,8 @@ async fn frame_application_metadata_is_stable_and_schema_change_rejected() {
let changed_schema = Arc::new((*base_schema).clone().with_metadata(changed));
let bad_batch = datatypes::series::frame::pack_raw_series_chunk(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
None,
&chunk(b"b", &[2], true),
)
.unwrap()
@@ -1380,7 +1448,7 @@ fn unpack_yields_at_empty_batch_budget_before_observing_eof() {
const BUDGET: usize = 32;
let frame_schema = datatypes::series::frame::raw_series_frame_schema(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
)
.unwrap();
let batches = vec![RecordBatch::new_empty(frame_schema.clone()); BUDGET];
@@ -1406,7 +1474,7 @@ fn unpack_yields_at_empty_batch_budget_before_observing_eof() {
fn unpack_propagates_upstream_pending_without_extra_wake() {
let frame_schema = datatypes::series::frame::raw_series_frame_schema(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
)
.unwrap();
let (source, polls, _) = df_pending(
@@ -1431,7 +1499,7 @@ fn unpack_reports_bad_batch_at_budget_boundary_before_yielding() {
const BUDGET: usize = 32;
let frame_schema = datatypes::series::frame::raw_series_frame_schema(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
)
.unwrap();
let mut batches = vec![RecordBatch::new_empty(frame_schema.clone()); BUDGET - 1];
@@ -1460,7 +1528,7 @@ fn unpack_surfaces_upstream_error_after_budget_yield_once() {
const BUDGET: usize = 32;
let frame_schema = datatypes::series::frame::raw_series_frame_schema(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
)
.unwrap();
let error = datafusion_common::DataFusionError::External(Box::new(BoxedError::new(
@@ -1537,6 +1605,10 @@ async fn unpack_drains_decoded_rows_before_polling_another_batch() {
assert!(unpacked.next().await.is_none());
}
/// An `expected` schema the raw frame does not declare is rejected before any poll, whether it names
/// no field at all or carries an evaluation grid. A grid schema is a legal *evaluated* declaration
/// now, so it is refused here because a raw frame carries no grid, not because the schema itself is
/// impossible.
#[tokio::test]
async fn unpack_preflights_empty_and_grid_schemas_without_polling() {
let invalid_schemas = [
@@ -1557,7 +1629,7 @@ async fn unpack_preflights_empty_and_grid_schemas_without_polling() {
for expected in invalid_schemas {
let frame_schema = datatypes::series::frame::raw_series_frame_schema(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
)
.unwrap();
let (source, polls, drops) = df_source(vec![], frame_schema);
@@ -1749,7 +1821,8 @@ async fn pack_bounds_ready_rows_at_default_row_limit_before_eof() {
for batch in [&first, &second] {
let decoded = datatypes::series::frame::unpack_raw_series_frame(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
None,
batch,
)
.unwrap();
@@ -1824,7 +1897,8 @@ async fn pack_splits_one_chunk_into_bounded_batches_without_repolling_the_source
for batch in [&first, &second] {
let decoded = datatypes::series::frame::unpack_raw_series_frame(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
None,
batch,
)
.unwrap();
@@ -1877,7 +1951,8 @@ async fn dropping_pack_mid_chunk_keeps_last_batch_and_releases_source() {
drop(owners);
let decoded = datatypes::series::frame::unpack_raw_series_frame(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
None,
&batch,
)
.unwrap();
@@ -1926,7 +2001,8 @@ async fn pack_flushes_ready_rows_at_pending_then_remainder_at_eof() {
assert_eq!(wake.0.load(Ordering::SeqCst), 1, "upstream wakes the task");
let decoded = datatypes::series::frame::unpack_raw_series_frame(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
None,
&first,
)
.unwrap();
@@ -1943,7 +2019,8 @@ async fn pack_flushes_ready_rows_at_pending_then_remainder_at_eof() {
assert_eq!(drops.load(Ordering::SeqCst), 1);
let decoded = datatypes::series::frame::unpack_raw_series_frame(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
None,
&second,
)
.unwrap();
@@ -1997,7 +2074,8 @@ async fn pack_emits_oversized_legal_rows_whole() {
] {
let chunks = datatypes::series::frame::unpack_raw_series_frame(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
None,
batch,
)
.unwrap();
@@ -2095,7 +2173,7 @@ async fn multirow_ipc_roundtrip_preserves_cross_chunk_variable_segments() {
assert_eq!(
reader.schema(),
batches[0].schema(),
"IPC preserves the v1 layout, metadata, and field order"
"IPC preserves the layout, metadata, and field order"
);
let frame_schema = reader.schema();
let ipc_batches = reader.map(Result::unwrap).collect::<Vec<_>>();
@@ -2285,8 +2363,8 @@ async fn public_chunk_with_empty_times_is_rejected_without_panic() {
}
/// `Interleaved` is rejected by the constructor before any source poll, for a pack source and for
/// an empty unpack stream; the wire-schema helper rejects it before assembling a schema, since no
/// v1 metadata value represents it. No adjacent-only fallback is installed.
/// an empty unpack stream; the wire-schema helper rejects it before assembling a schema, since the
/// wire carries only the two representable spellings. No adjacent-only fallback is installed.
#[tokio::test]
async fn interleaved_ordering_is_rejected_before_any_poll() {
assert!(
@@ -2319,7 +2397,7 @@ async fn interleaved_ordering_is_rejected_before_any_poll() {
// a frame that declares one; that spelling has no wire representation and is rejected.
let frame = datatypes::series::frame::raw_series_frame_schema(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
)
.unwrap();
let mut metadata = frame.metadata().clone();
@@ -2343,17 +2421,20 @@ async fn interleaved_ordering_is_rejected_before_any_poll() {
/// The unpack stream rejects a batch whose schema differs from the stream's declared schema,
/// including drift in metadata the codec itself does not interpret and an empty batch that
/// carries the drift; the reserved layout metadata is rejected as well.
/// carries the drift; the reserved layout metadata is rejected as well. Since the current layout is
/// `v2`, the drift case rewrites the layout to `v1`: any value other than this revision's is refused,
/// so a stale or a future spelling cannot be decoded under the current columns.
#[tokio::test]
async fn unpack_rejects_unrelated_metadata_drift_and_empty_schema_changes() {
let base = datatypes::series::frame::raw_series_frame_schema(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
)
.unwrap();
let first = datatypes::series::frame::pack_raw_series_chunk(
&schema(),
datatypes::series::frame::RawLabelMode::Explicit,
datatypes::series::frame::FrameLabelMode::Explicit,
None,
&chunk(b"a", &[1], true),
)
.unwrap()
@@ -2361,8 +2442,9 @@ async fn unpack_rejects_unrelated_metadata_drift_and_empty_schema_changes() {
let mut unrelated = base.metadata().clone();
unrelated.insert("app.unrelated".into(), "x".into());
// The current layout is `v2`, so a frame that still declares `v1` is the drift here.
let mut reserved = base.metadata().clone();
reserved.insert("greptime.series.layout".into(), "v2".into());
reserved.insert("greptime.series.layout".into(), "v1".into());
let mut renamed = base.metadata().clone();
renamed.insert("app.renamed".into(), "y".into());
@@ -2408,6 +2490,9 @@ async fn unpack_rejects_unrelated_metadata_drift_and_empty_schema_changes() {
}
}
/// The ordering and layout fixture: literal metadata maps are read the way a receiver reads them,
/// so the same text pins both what `series_frame_ordering` answers and whether the declaration as a
/// whole is accepted.
#[test]
fn raw_series_ordering_metadata_fixture() {
#[derive(serde::Deserialize)]
@@ -2415,10 +2500,12 @@ fn raw_series_ordering_metadata_fixture() {
name: String,
metadata: HashMap<String, String>,
ordering: Option<String>,
/// `accepted` or `rejected`: whether the whole declaration is one this revision reads.
declaration: String,
}
let cases: Vec<Case> = serde_json::from_str(include_str!(
"../../../../../tests/compatibility/wire/raw_series_ordering/v1_metadata.json"
"../../../../../tests/compatibility/wire/raw_series_ordering/v2_metadata.json"
))
.unwrap();
for case in cases {
@@ -2428,22 +2515,52 @@ fn raw_series_ordering_metadata_fixture() {
Some(value) => panic!("{} has invalid fixture ordering {value}", case.name),
None => None,
};
let frame_schema = raw_series_frame_schema(&schema(), RawLabelMode::Explicit).unwrap();
// The columns have to match the content the case declares, so an evaluated case is built on
// an evaluated layout rather than read through a raw one.
let evaluated = case
.metadata
.get("greptime.series.content")
.is_some_and(|content| content == "evaluated");
let frame_schema = if evaluated {
series_frame_schema(&evaluated_schema(), FrameLabelMode::Explicit, None).unwrap()
} else {
raw_series_frame_schema(&schema(), FrameLabelMode::Explicit).unwrap()
};
let frame_schema = frame_schema.as_ref().clone().with_metadata(case.metadata);
match literal_ordering {
Some(expected) => assert_eq!(
raw_series_frame_ordering(&frame_schema)
series_frame_ordering(&frame_schema)
.unwrap_or_else(|error| panic!("{}: {error}", case.name)),
expected,
"{}",
case.name
),
None => assert!(
raw_series_frame_ordering(&frame_schema).is_err(),
series_frame_ordering(&frame_schema).is_err(),
"{}",
case.name
),
}
let declared = datatypes::series::frame::SeriesFrameDeclaration::parse(&frame_schema);
match case.declaration.as_str() {
"accepted" => {
let declared = declared.unwrap_or_else(|error| panic!("{}: {error}", case.name));
if evaluated {
assert_eq!(
declared.content(),
datatypes::series::frame::FrameContent::Evaluated,
"{}",
case.name
);
}
}
"rejected" => assert!(
declared.is_err(),
"{}: the declaration must be refused",
case.name
),
other => panic!("{} has invalid fixture declaration {other}", case.name),
}
}
}
@@ -2463,14 +2580,14 @@ async fn pack_declares_ordering_on_every_batch_and_ipc_roundtrips_it() {
// replaced single-row packing emitted one batch per chunk.
let expected_batch_count = if is_empty { 0 } else { 1 };
let canonical_schema =
raw_series_frame_schema(&schema(), RawLabelMode::Explicit).unwrap();
raw_series_frame_schema(&schema(), FrameLabelMode::Explicit).unwrap();
let mut expected_metadata = canonical_schema.metadata().clone();
expected_metadata.insert(
"greptime.series.ordering".to_string(),
match ordering {
SeriesOrdering::Contiguous => "contiguous",
SeriesOrdering::KeyThenTime => "read_key_then_time",
// The v1 wire carries exactly these two spellings, so the loop above
// The wire carries exactly these two spellings, so the loop above
// admits only the two representable modes; `Interleaved` is rejected
// before any poll (see `interleaved_ordering_is_rejected_before_any_poll`).
SeriesOrdering::Interleaved => {
@@ -2484,7 +2601,7 @@ async fn pack_declares_ordering_on_every_batch_and_ipc_roundtrips_it() {
let declared = packed.schema();
assert_eq!(declared.fields(), canonical_schema.fields());
assert_eq!(declared.metadata(), &expected_metadata);
assert_eq!(raw_series_frame_ordering(&declared).unwrap(), ordering);
assert_eq!(series_frame_ordering(&declared).unwrap(), ordering);
let mut batches = Vec::new();
while let Some(batch) = packed.next().await {
let batch = batch.unwrap();
@@ -2502,10 +2619,7 @@ async fn pack_declares_ordering_on_every_batch_and_ipc_roundtrips_it() {
}
let reader = StreamReader::try_new(Cursor::new(ipc), None).unwrap();
assert_eq!(reader.schema(), declared);
assert_eq!(
raw_series_frame_ordering(&reader.schema()).unwrap(),
ordering
);
assert_eq!(series_frame_ordering(&reader.schema()).unwrap(), ordering);
let ipc_batches = reader.map(Result::unwrap).collect::<Vec<_>>();
assert_eq!(ipc_batches.len(), batches.len());
assert_eq!(ipc_batches.len(), expected_batch_count);
@@ -2621,17 +2735,33 @@ fn two_tag_binding(metric: &str) -> RawLabelSource {
}
}
/// A key frame whose metadata carries exactly the supplied description and nothing else.
fn key_frame_with(description: &datatypes::series::frame::LabelDecodeSpec) -> SchemaRef {
let layout = raw_series_frame_schema(&schema(), RawLabelMode::Key).unwrap();
/// A raw key frame declaring the supplied description, as a producer with a primary-key binding
/// writes it.
fn key_layout() -> SchemaRef {
let description =
label_decode_spec(&key_binding("cpu")).expect("a primary-key binding describes its key");
series_frame_schema(&schema(), FrameLabelMode::Key, Some(&description)).unwrap()
}
/// A key frame whose metadata carries exactly the supplied description text and nothing else.
///
/// The text is written verbatim, not canonicalized, so a case can pin what a writer must have
/// ordered and what a reader refuses to reorder.
fn key_frame_with_encoded(encoded: &str) -> SchemaRef {
let layout = key_layout();
let mut metadata = layout.metadata().clone();
metadata.insert(
"greptime.series.label_decode".to_string(),
serde_json::to_string(description).unwrap(),
encoded.to_string(),
);
Arc::new(Schema::new_with_metadata(layout.fields().clone(), metadata))
}
/// A key frame whose `greptime.series.label_decode` entry is the supplied description.
fn key_frame_with(description: &datatypes::series::frame::LabelDecodeSpec) -> SchemaRef {
key_frame_with_encoded(&serde_json::to_string(description).unwrap())
}
/// Logical tag order is fixed by the writer, so the wire form has one spelling and the comparison
/// is plain equality. The ordered key columns are the encoding, so the same columns in another
/// order are a different layout and are rejected.
@@ -2720,7 +2850,7 @@ async fn frame_check_rejects_a_binding_the_frame_does_not_describe() {
"an explicit identity expects a frame that carries complete labels"
);
let explicit = raw_series_frame_schema(&schema(), RawLabelMode::Explicit).unwrap();
let explicit = raw_series_frame_schema(&schema(), FrameLabelMode::Explicit).unwrap();
assert!(
ensure_frame_matches_label_source(explicit.as_ref(), &key_binding("cpu")).is_err(),
"a frame declaring no layout cannot satisfy a primary-key identity"
@@ -2746,7 +2876,7 @@ async fn unpack_retains_the_declared_description_without_a_binding() {
"no physical region key layout is invented for a frame that carries none"
);
let explicit = raw_series_frame_schema(&schema(), RawLabelMode::Explicit).unwrap();
let explicit = raw_series_frame_schema(&schema(), FrameLabelMode::Explicit).unwrap();
let (source, _, _) = df_source(vec![], explicit);
let unpacked = unpack_raw_series(source, None).unwrap();
assert_eq!(
@@ -2756,17 +2886,6 @@ async fn unpack_retains_the_declared_description_without_a_binding() {
);
}
/// A key frame whose `greptime.series.label_decode` entry is exactly the supplied text.
fn key_frame_with_encoded(encoded: &str) -> SchemaRef {
let layout = raw_series_frame_schema(&schema(), RawLabelMode::Key).unwrap();
let mut metadata = layout.metadata().clone();
metadata.insert(
"greptime.series.label_decode".to_string(),
encoded.to_string(),
);
Arc::new(Schema::new_with_metadata(layout.fields().clone(), metadata))
}
/// The description is validated whether or not a binding is supplied, so trusting the frame never
/// means skipping validation: an unknown version, an unknown field, a malformed entry, a logical
/// tag outside the key, a key frame without a description, and an explicit frame carrying one are
@@ -2810,14 +2929,21 @@ async fn unpack_rejects_invalid_descriptions_without_a_binding() {
);
}
let key_without_description = raw_series_frame_schema(&schema(), RawLabelMode::Key).unwrap();
// A key frame nobody can decode: the description is removed from the frame the producer wrote.
// Construction cannot express this, which is the point - the reader is what refuses it.
let mut without_description = key_layout().metadata().clone();
without_description.remove("greptime.series.label_decode");
let key_without_description = Arc::new(Schema::new_with_metadata(
key_layout().fields().clone(),
without_description,
));
let (source, _, _) = df_source(vec![], key_without_description);
assert!(
unpack_raw_series(source, None).is_err(),
"a key frame nobody can decode is rejected instead of being trusted"
);
let explicit = raw_series_frame_schema(&schema(), RawLabelMode::Explicit).unwrap();
let explicit = raw_series_frame_schema(&schema(), FrameLabelMode::Explicit).unwrap();
let mut metadata = explicit.metadata().clone();
metadata.insert(
"greptime.series.label_decode".to_string(),
@@ -2850,14 +2976,16 @@ async fn pack_declares_the_label_decode_description_of_its_binding() {
);
let mut packed = pack_raw_series(input).unwrap();
let declared = packed.schema();
let canonical = raw_series_frame_schema(&schema(), RawLabelMode::Key).unwrap();
// A key frame's canonical layout already carries its description, so the producer adds only the
// ordering on top of it.
let canonical = key_layout();
assert_eq!(
declared.metadata().len(),
canonical.metadata().len() + 2,
"a key frame adds the producer ordering and the label decode description"
canonical.metadata().len() + 1,
"a key frame adds the producer ordering to the description it must declare"
);
assert_eq!(
datatypes::series::frame::raw_series_frame_label_decode(&declared).unwrap(),
datatypes::series::frame::series_frame_label_decode(&declared).unwrap(),
label_decode_spec(&binding),
"the declared description is derived from the binding, not from the first row"
);
@@ -2872,11 +3000,16 @@ async fn pack_declares_the_label_decode_description_of_its_binding() {
let packed = pack_raw_series(input).unwrap();
let declared = packed.schema();
assert_eq!(
datatypes::series::frame::raw_series_frame_label_decode(&declared).unwrap(),
datatypes::series::frame::series_frame_label_decode(&declared).unwrap(),
None,
"explicit labels are already complete"
);
assert_eq!(declared.metadata().len(), canonical.metadata().len() + 1);
let canonical_explicit = raw_series_frame_schema(&schema(), FrameLabelMode::Explicit).unwrap();
assert_eq!(
declared.metadata().len(),
canonical_explicit.metadata().len() + 1,
"an explicit frame adds only the producer ordering"
);
}
/// A key frame that omits its description is not decodable, and unpacking says so instead of falling
@@ -2899,13 +3032,11 @@ async fn unpack_round_trips_a_key_frame_and_rejects_one_without_a_description()
let chunk = decoded.next().await.unwrap().unwrap();
assert_eq!(chunk.primary_key.as_slice(), b"encoded-complete-key");
// A frame that omits the description is not decodable from the schema alone.
let canonical = raw_series_frame_schema(&schema(), RawLabelMode::Key).unwrap();
let mut metadata = canonical.metadata().clone();
metadata.insert(
"greptime.series.ordering".to_string(),
"contiguous".to_string(),
);
// A frame that omits the description is not decodable from the schema alone. The producer
// cannot write one, so the entry is dropped from the frame it did write: the reader is what has
// to refuse it.
let mut metadata = declared.metadata().clone();
metadata.remove("greptime.series.label_decode");
let stripped = Arc::new(datatypes::arrow::datatypes::Schema::new_with_metadata(
declared.fields().clone(),
metadata,
@@ -2933,7 +3064,7 @@ async fn unpack_reports_the_ordering_the_frame_declares() {
);
let packed = pack_raw_series(input).unwrap();
assert_eq!(
raw_series_frame_ordering(packed.schema().as_ref()).unwrap(),
series_frame_ordering(packed.schema().as_ref()).unwrap(),
SeriesOrdering::KeyThenTime,
"the producer declaration reaches the wire"
);
@@ -2946,7 +3077,7 @@ async fn unpack_reports_the_ordering_the_frame_declares() {
);
// A frame that declares no ordering is a legacy contiguous frame.
let frame = raw_series_frame_schema(&schema(), RawLabelMode::Explicit).unwrap();
let frame = raw_series_frame_schema(&schema(), FrameLabelMode::Explicit).unwrap();
let (source, _, _) = df_source(vec![], frame);
assert_eq!(
unpack_raw_series(source, None).unwrap().ordering(),
@@ -2955,17 +3086,19 @@ async fn unpack_reports_the_ordering_the_frame_declares() {
}
/// Reserved layout metadata this revision does not write is rejected instead of being read as the
/// current layout, so a frame cannot be decoded under a contract it did not declare.
/// current layout, so a frame cannot be decoded under a contract it did not declare. The current
/// layout is `v2`; a `v1` frame and every unknown value are refused alike.
#[tokio::test]
async fn unpack_rejects_unknown_layout_metadata() {
let cases = [
("greptime.series.layout", "v2"),
("greptime.series.layout", "v1"),
("greptime.series.layout", "v3"),
("greptime.series.content", "series"),
("greptime.series.time", "timestamps_v2"),
("greptime.series.ordering", "hash_then_time"),
];
for (key, value) in cases {
let frame = raw_series_frame_schema(&schema(), RawLabelMode::Explicit).unwrap();
let frame = raw_series_frame_schema(&schema(), FrameLabelMode::Explicit).unwrap();
let mut metadata = frame.metadata().clone();
metadata.insert(key.to_string(), value.to_string());
let frame = Arc::new(Schema::new_with_metadata(frame.fields().clone(), metadata));
@@ -2982,7 +3115,7 @@ async fn unpack_rejects_unknown_layout_metadata() {
/// same frame unpacks without it because the frame is the only source of decode information.
#[tokio::test]
async fn unpack_rejects_an_expected_schema_the_frame_does_not_declare() {
let frame = raw_series_frame_schema(&schema(), RawLabelMode::Explicit).unwrap();
let frame = raw_series_frame_schema(&schema(), FrameLabelMode::Explicit).unwrap();
let other_field = Arc::new(SeriesSchema {
fields: vec![FieldSpec {
@@ -3049,7 +3182,7 @@ async fn unpack_pack_unpack_preserves_the_description_and_the_ordering() {
let (input, _, _) = raw_custom(vec![key_chunk()], SeriesOrdering::KeyThenTime, binding);
let packed = pack_raw_series(input).unwrap();
assert_eq!(
datatypes::series::frame::raw_series_frame_label_decode(packed.schema().as_ref()).unwrap(),
datatypes::series::frame::series_frame_label_decode(packed.schema().as_ref()).unwrap(),
Some(description.clone())
);
@@ -3059,13 +3192,12 @@ async fn unpack_pack_unpack_preserves_the_description_and_the_ordering() {
let repacked = pack_raw_series(unpacked).unwrap();
assert_eq!(
datatypes::series::frame::raw_series_frame_label_decode(repacked.schema().as_ref())
.unwrap(),
datatypes::series::frame::series_frame_label_decode(repacked.schema().as_ref()).unwrap(),
Some(description.clone()),
"the repacked frame declares what the unpacked stream declared"
);
assert_eq!(
raw_series_frame_ordering(repacked.schema().as_ref()).unwrap(),
series_frame_ordering(repacked.schema().as_ref()).unwrap(),
SeriesOrdering::KeyThenTime,
"the repacked frame keeps the ordering it was unpacked under"
);
@@ -3078,3 +3210,42 @@ async fn unpack_pack_unpack_preserves_the_description_and_the_ordering() {
assert!(chunk.metric.is_none(), "key chunks stay undecoded");
assert!(again.next().await.is_none());
}
/// Each typed schema entry point refuses the content it cannot describe.
///
/// The raw entry point is handed an evaluated series schema and the shared builder is handed an
/// evaluated schema with a key-then-time ordering. Both would otherwise produce a frame whose
/// declaration contradicts its own columns, and the reader would be the first to notice — after the
/// producer had already published it.
#[test]
fn typed_schema_entry_points_refuse_the_content_they_cannot_describe() {
assert!(
super::raw_series_frame_schema_with_ordering(
&evaluated_schema(),
&RawLabelSource::Explicit,
SeriesOrdering::Contiguous,
)
.is_err(),
"a raw frame cannot carry a schema with an evaluation grid"
);
assert!(
super::frame_schema_with(
&evaluated_schema(),
FrameLabelMode::Explicit,
None,
SeriesOrdering::KeyThenTime,
)
.is_err(),
"an evaluated frame has no storage key to be ordered by"
);
assert!(
super::frame_schema_with(
&evaluated_schema(),
FrameLabelMode::Explicit,
None,
SeriesOrdering::Contiguous,
)
.is_ok(),
"contiguous is the one ordering an evaluated frame can honour"
);
}
src/common/query/src/series_frame/tests/evaluated.rs
+1250
View File
File diff suppressed because it is too large Load Diff
src/datatypes/src/series/frame.rs
+1153 -435
View File
File diff suppressed because it is too large Load Diff
src/datatypes/src/series/frame/tests.rs
+670 -216
View File
File diff suppressed because it is too large Load Diff
src/datatypes/src/series/time.rs
+7 -1
View File
@@ -17,6 +17,7 @@ use std::sync::Arc;
use arrow::buffer::ScalarBuffer;
use once_cell::sync::OnceCell;
use serde::{Deserialize, Serialize};
use snafu::OptionExt;
use crate::error::{InvalidSeriesSnafu, Result};
@@ -25,7 +26,12 @@ use crate::error::{InvalidSeriesSnafu, Result};
mod tests;
/// Value description of a non-empty, fixed-step evaluation grid.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
///
/// The serde form is the wire form a series frame writes under `greptime.series.grid`: field names
/// in declaration order, no optional members. Two nodes compare that text byte for byte, so the
/// only place that spells it is a series frame's own grid metadata entry.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct GridSpec {
/// First evaluation timestamp in milliseconds.
pub start_ms: i64,
src/mito-codec/src/labels/tests.rs
+3 -3
View File
@@ -629,13 +629,13 @@ fn frame_label_decode_description_decodes_like_the_bound_source() {
);
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(
let frame_schema = datatypes::series::frame::series_frame_schema(
&frame_series_schema(),
datatypes::series::frame::RawLabelMode::Key,
datatypes::series::frame::FrameLabelMode::Key,
Some(&description),
)
.unwrap();
let recovered = datatypes::series::frame::raw_series_frame_label_decode(&frame_schema)
let recovered = datatypes::series::frame::series_frame_label_decode(&frame_schema)
.unwrap()
.expect("the frame declares its decode description");
assert_eq!(recovered, description);
src/mito2/src/engine/scan_test.rs
+7 -9
View File
@@ -5093,7 +5093,7 @@ async fn test_mito_handle_query_raw_series_candidate_projection_and_mvcc() {
assert_eq!(raw_visible, visible);
// Exercise the production table physical consumer over a fresh real Mito scanner.
use common_query::series_frame::{raw_series_frame_ordering, unpack_raw_series};
use common_query::series_frame::{series_frame_ordering, unpack_raw_series};
use datafusion::execution::TaskContext;
use datafusion::physical_plan::ExecutionPlan;
use table::table::scan::RegionScanExec;
@@ -5127,7 +5127,7 @@ async fn test_mito_handle_query_raw_series_candidate_projection_and_mvcc() {
.unwrap();
// Unpack under the frame's own declaration. A legacy producer declares `KeyThenTime`, and
// only that declaration makes the decoder check that keys increase.
let frame_ordering = raw_series_frame_ordering(native_frames.schema().as_ref()).unwrap();
let frame_ordering = series_frame_ordering(native_frames.schema().as_ref()).unwrap();
assert_eq!(
SeriesOrdering::KeyThenTime,
frame_ordering,
@@ -5233,9 +5233,7 @@ async fn test_mito_handle_query_raw_series_candidate_projection_and_mvcc() {
/// `promql::source::merge::tests::rejects_unsupported_ordering_and_empty_input_without_polling`.
#[tokio::test]
async fn test_two_phase_native_raw_stream_declares_contiguous_frames() {
use common_query::series_frame::{
pack_raw_series, raw_series_frame_ordering, unpack_raw_series,
};
use common_query::series_frame::{pack_raw_series, series_frame_ordering, unpack_raw_series};
use datatypes::series::SeriesOrdering;
use datatypes::series::samples::SeriesSchema;
use store_api::region_engine::{BoundTag, QueryScanContext, RawLabelSource};
@@ -5374,7 +5372,7 @@ async fn test_two_phase_native_raw_stream_declares_contiguous_frames() {
chunks,
)))
.unwrap();
let ordering = raw_series_frame_ordering(packed.schema().as_ref()).unwrap();
let ordering = series_frame_ordering(packed.schema().as_ref()).unwrap();
assert_eq!(SeriesOrdering::Contiguous, ordering);
let validated = unpack_raw_series(packed, Some(schema))
.unwrap()
@@ -6560,7 +6558,7 @@ async fn test_mito_raw_series_int64_predicate_uses_original_precision() {
assert_eq!(vec![(2, 20.0)], output_values);
// Validate the typed integer gate through RegionScanExec's actual physical consumer.
use common_query::series_frame::{raw_series_frame_ordering, unpack_raw_series};
use common_query::series_frame::{series_frame_ordering, unpack_raw_series};
use datafusion::execution::TaskContext;
use datafusion::physical_plan::ExecutionPlan;
use datatypes::series::SeriesOrdering;
@@ -6597,7 +6595,7 @@ async fn test_mito_raw_series_int64_predicate_uses_original_precision() {
.execute(0, Arc::new(TaskContext::default()))
.unwrap();
// Unpack under the frame's own declaration so the decoder checks the legacy key order.
let ordering = raw_series_frame_ordering(frames.schema().as_ref()).unwrap();
let ordering = series_frame_ordering(frames.schema().as_ref()).unwrap();
assert_eq!(
SeriesOrdering::KeyThenTime,
ordering,
@@ -6655,7 +6653,7 @@ async fn test_mito_raw_series_int64_predicate_uses_original_precision() {
let empty_frames = empty_plan
.execute(0, Arc::new(TaskContext::default()))
.unwrap();
let empty_ordering = raw_series_frame_ordering(empty_frames.schema().as_ref()).unwrap();
let empty_ordering = series_frame_ordering(empty_frames.schema().as_ref()).unwrap();
assert_eq!(
SeriesOrdering::KeyThenTime,
empty_ordering,
src/promql/src/extension_plan/fused_series/tests.rs
+47 -17
View File
@@ -63,7 +63,11 @@ use datafusion::physical_plan::{
RecordBatchStream, SendableRecordBatchStream, collect, displayable,
};
use datafusion::prelude::SessionContext;
use datatypes::series::frame::{RawLabelMode, pack_raw_series_chunk, raw_series_frame_schema};
use datatypes::prelude::ConcreteDataType;
use datatypes::series::frame::{
FrameLabelMode, LABEL_DECODE_VERSION, LabelDecodeColumn, LabelDecodeEncoding, LabelDecodeSpec,
LabelDecodeTag, pack_raw_series_chunk, raw_series_frame_schema, series_frame_schema,
};
use datatypes::series::raw::{RawSeriesChunk, RawTimestamps};
use datatypes::series::samples::{FieldSpec, Samples, Series, SeriesFields, SeriesSchema};
use datatypes::series::time::{EvalGrid, GridSpan, GridSpec, TimeAxis};
@@ -488,7 +492,7 @@ fn binding(tree: &NativeTree) -> FusedSeriesBinding {
/// The schema of a frame carrying series of `schema` with their labels explicit.
fn frame_schema_of(schema: &SeriesSchema) -> SchemaRef {
raw_series_frame_schema(schema, RawLabelMode::Explicit).expect("explicit labels are a frame")
raw_series_frame_schema(schema, FrameLabelMode::Explicit).expect("explicit labels are a frame")
}
/// The schema of a frame carrying these series, which is what the node declares as its input.
@@ -496,6 +500,30 @@ fn frame_schema() -> SchemaRef {
frame_schema_of(&raw_schema())
}
/// The description a frame with its labels encoded in a dense [`HOST`] key declares.
fn host_label_decode() -> LabelDecodeSpec {
LabelDecodeSpec {
version: LABEL_DECODE_VERSION,
metric_name: METRIC.to_string(),
encoding: LabelDecodeEncoding::Dense,
primary_key: vec![LabelDecodeColumn {
column_id: 1,
name: HOST.to_string(),
physical_type: ConcreteDataType::string_datatype(),
}],
logical_tags: vec![LabelDecodeTag {
column_id: 1,
name: HOST.to_string(),
}],
}
}
/// The schema of a frame carrying series of `schema` whose labels are encoded in their key.
fn keyed_frame_schema_of(schema: &SeriesSchema) -> SchemaRef {
series_frame_schema(schema, FrameLabelMode::Key, Some(&host_label_decode()))
.expect("key labels are a frame")
}
/// The columns this node produces: the field, the tags and the grid timestamps.
fn output_schema() -> SchemaRef {
Arc::new(Schema::new(vec![
@@ -975,7 +1003,7 @@ fn metric(labels: &[(&str, &str)], name: Option<&str>) -> SeriesLabels {
/// One frame batch carrying `chunk` of `schema`, packed by the scan's own encoder.
fn frame_of(schema: &SeriesSchema, chunk: &RawSeriesChunk) -> RecordBatch {
let mut batches = pack_raw_series_chunk(schema, RawLabelMode::Explicit, chunk)
let mut batches = pack_raw_series_chunk(schema, FrameLabelMode::Explicit, None, chunk)
.expect("a chunk of this schema packs into a frame");
assert_eq!(1, batches.len(), "one small chunk is one frame batch");
batches.remove(0)
@@ -2270,8 +2298,7 @@ async fn a_pending_boundary_loses_no_rows() {
/// before the first batch, not when the missing labels reach the output.
#[test]
fn a_frame_whose_labels_are_encoded_is_rejected_when_the_plan_is_built() {
let keyed =
raw_series_frame_schema(&raw_schema(), RawLabelMode::Key).expect("key labels are a frame");
let keyed = keyed_frame_schema_of(&raw_schema());
let error = plan(rate_tree(), memory_exec_with(keyed, Vec::new()))
.expect_err("the frames carry no labels to read tags from");
assert!(
@@ -2283,18 +2310,19 @@ fn a_frame_whose_labels_are_encoded_is_rejected_when_the_plan_is_built() {
/// A frame that carries an evaluation grid is refused when the plan is built.
///
/// A frame holds raw series: its rows are samples, not grid positions, and the grid every output
/// row is addressed on is the tree's own. A schema that declares a grid is one this node would
/// read positions out of, so it is refused rather than decoded. The source below is one partition,
/// which is the shape the node accepts, so what refuses the plan here is the frame's own
/// declaration and not the node's partition count: the refusal has to name the grid.
/// row is addressed on is the tree's own. A frame that declares a grid is one whose rows this node
/// would read as positions on it, so the shared decoder refuses it rather than reading samples out
/// of it. The source below is one partition, which is the shape the node accepts, so what refuses
/// the plan here is the frame's own declaration and not the node's partition count: the refusal has
/// to name the frame.
#[test]
fn a_frame_with_an_evaluation_grid_is_rejected_when_the_plan_is_built() {
let mut metadata = frame_schema().metadata().clone();
metadata.insert("greptime.series.grid".to_string(), "v1".to_string());
let gridded = Arc::new(Schema::new_with_metadata(
frame_schema().fields().clone(),
metadata,
));
let gridded_schema = Arc::new(SeriesSchema {
fields: raw_schema().fields.clone(),
eval_grid: Some(grid()),
});
let gridded = series_frame_schema(&gridded_schema, FrameLabelMode::Explicit, None)
.expect("a grid is a frame");
// One empty partition: every check but the frame's passes.
let source = memory_exec_with(gridded, Vec::new());
assert_eq!(
@@ -2305,7 +2333,7 @@ fn a_frame_with_an_evaluation_grid_is_rejected_when_the_plan_is_built() {
let error = plan(rate_tree(), source).expect_err("the frame declares a grid");
let message = error_chain(&error);
assert!(
message.contains("raw series schema cannot have an evaluation grid"),
message.contains("expected a raw frame, found an evaluated one"),
"refused for another reason: {message}"
);
assert!(
@@ -2326,7 +2354,9 @@ fn a_frame_of_another_series_schema_is_rejected_when_the_plan_is_built() {
let error = plan(rate_tree(), memory_exec_with(other, Vec::new()))
.expect_err("the tree's input reads one field");
assert!(
error.to_string().contains("raw frame"),
error
.to_string()
.contains("frame fields do not match the expected series schema"),
"refused for another reason: {error}"
);
}
src/promql/src/source/merge/tests.rs
+8 -8
View File
@@ -24,7 +24,7 @@ use api::v1::SemanticType;
use common_error::ext::{BoxedError, ErrorExt, PlainError};
use common_error::status_code::StatusCode;
use common_query::series_frame::{
label_decode_spec, pack_raw_series, raw_series_frame_ordering, unpack_raw_series,
label_decode_spec, pack_raw_series, series_frame_ordering, unpack_raw_series,
};
use datatypes::arrow::array::Float64Array;
use datatypes::arrow::buffer::{Buffer, ScalarBuffer};
@@ -458,7 +458,7 @@ async fn merges_two_unpacked_frames_through_the_frame_codec() {
);
let schema = Arc::clone(stream.schema());
let packed = pack_raw_series(stream).unwrap();
let ordering = raw_series_frame_ordering(packed.schema().as_ref()).unwrap();
let ordering = series_frame_ordering(packed.schema().as_ref()).unwrap();
assert_eq!(SeriesOrdering::KeyThenTime, ordering);
unpack_raw_series(packed, Some(schema)).unwrap()
};
@@ -1889,7 +1889,7 @@ async fn merge_rejects_inputs_that_describe_different_decode_layouts() {
/// 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::series_frame_ordering;
use common_query::series_frame::{pack_raw_series, unpack_raw_series};
use store_api::region_engine::NO_BOUND_LABEL_SOURCE;
@@ -1937,11 +1937,11 @@ async fn merge_without_a_bound_identity_still_advertises_the_decode_layout() {
// 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(),
datatypes::series::frame::series_frame_label_decode(framed.schema().as_ref()).unwrap(),
Some(carried.clone())
);
assert_eq!(
raw_series_frame_ordering(framed.schema().as_ref()).unwrap(),
series_frame_ordering(framed.schema().as_ref()).unwrap(),
SeriesOrdering::KeyThenTime
);
@@ -1970,7 +1970,7 @@ async fn merge_without_a_bound_identity_still_advertises_the_decode_layout() {
#[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 datatypes::series::frame::series_frame_label_decode;
use mito_codec::labels::{LabelProjection, PrimaryKeyLabelDecoder};
let key = read_key("rack-a", "sensor-1");
@@ -1994,12 +1994,12 @@ async fn merged_frame_declares_a_label_decode_description_that_decodes_its_keys(
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(),
common_query::series_frame::series_frame_ordering(framed.schema().as_ref()).unwrap(),
ordering
);
let declared = framed.schema();
assert_eq!(
raw_series_frame_label_decode(declared.as_ref()).unwrap(),
series_frame_label_decode(declared.as_ref()).unwrap(),
Some(expected.clone())
);
src/promql/src/source/merge/tests/payload.rs
+3 -3
View File
@@ -19,8 +19,8 @@ use std::task::{Context, Poll};
use common_error::ext::BoxedError;
use common_query::series_frame::{
ensure_frame_matches_label_source, label_decode_spec, pack_raw_series,
raw_series_frame_ordering, unpack_raw_series,
ensure_frame_matches_label_source, label_decode_spec, pack_raw_series, series_frame_ordering,
unpack_raw_series,
};
use datatypes::arrow::array::{Array, Float64Array, ListArray, UInt32Array};
use datatypes::arrow::buffer::{Buffer, ScalarBuffer};
@@ -263,7 +263,7 @@ async fn merges_typed_multifield_payload_through_frame_and_label_roundtrip() {
// 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(),
series_frame_ordering(framed.schema().as_ref()).unwrap(),
frame_ordering,
"the packed frame declares the merge's ordering"
);
src/promql/src/source/tests.rs
+1 -2
View File
@@ -1073,8 +1073,7 @@ async fn materializes_labels_from_a_frame_with_no_supplied_binding() {
);
let framed = pack_raw_series(raw).unwrap();
assert_eq!(
datatypes::series::frame::raw_series_frame_label_decode(framed.schema().as_ref())
.unwrap(),
datatypes::series::frame::series_frame_label_decode(framed.schema().as_ref()).unwrap(),
Some(description.clone()),
"{encoding:?}: the frame schema is the producer's only label statement"
);
src/query/src/dist_plan/merge_scan.rs
+56 -18
View File
@@ -1423,7 +1423,7 @@ mod tests {
};
use datatypes::prelude::{ConcreteDataType, VectorRef};
use datatypes::schema::{ColumnSchema, Schema};
use datatypes::series::frame::{self, RawLabelMode};
use datatypes::series::frame::{self, FrameLabelMode};
use datatypes::series::raw::{RawSeriesChunk, RawTimestamps};
use datatypes::series::samples::{
DenseSamplesRef, FieldSpec, SampleRef, Samples, SeriesFields, SeriesSchema,
@@ -3361,7 +3361,8 @@ mod tests {
async fn merge_scan_remote_schema_identity_rejects_empty_raw_frame_metadata_mismatch() {
let region_id = RegionId::new(1024, 1);
let raw = raw_test_series_schema();
let frame = frame::raw_series_frame_schema(&raw, RawLabelMode::Key).unwrap();
let decode = raw_test_label_decode();
let frame = frame::series_frame_schema(&raw, FrameLabelMode::Key, Some(&decode)).unwrap();
let metadata = frame.metadata().clone();
let mut cases = Vec::new();
for (key, value) in [
@@ -3428,7 +3429,7 @@ mod tests {
label_decode_spec, raw_series_frame_schema_with_ordering,
};
use datatypes::series::SeriesOrdering;
use datatypes::series::frame::raw_series_frame_label_decode;
use datatypes::series::frame::series_frame_label_decode;
let raw = raw_test_series_schema();
// Two bindings that are equal but not the same allocation: the expectation and the
@@ -3503,7 +3504,7 @@ mod tests {
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)
series_frame_label_decode(&spaced)
.unwrap()
.map(|decode| decode.version),
Some(1),
@@ -3512,6 +3513,27 @@ mod tests {
assert!(validate_remote_schema(&expected, &spaced, "test").is_err());
}
/// The key layout a key frame over [`raw_test_series_schema`] declares.
///
/// A key-encoded frame cannot be declared without the description that decodes its key, and the
/// merge-scan tests never decode a key with it, so it only has to be self-consistent.
fn raw_test_label_decode() -> frame::LabelDecodeSpec {
frame::LabelDecodeSpec {
version: frame::LABEL_DECODE_VERSION,
metric_name: "cpu_usage".to_string(),
encoding: frame::LabelDecodeEncoding::Dense,
primary_key: vec![frame::LabelDecodeColumn {
column_id: 1,
name: "host".to_string(),
physical_type: ConcreteDataType::string_datatype(),
}],
logical_tags: vec![frame::LabelDecodeTag {
column_id: 1,
name: "host".to_string(),
}],
}
}
fn raw_test_series_schema() -> SeriesSchema {
SeriesSchema {
fields: vec![
@@ -3531,8 +3553,9 @@ mod tests {
async fn merge_scan_remote_schema_identity_rejects_later_raw_frame_inner_metadata_mismatch() {
let region_id = RegionId::new(1024, 1);
let raw = raw_test_series_schema();
let mode = RawLabelMode::Key;
let frame = frame::raw_series_frame_schema(&raw, mode).unwrap();
let mode = FrameLabelMode::Key;
let decode = raw_test_label_decode();
let frame = frame::series_frame_schema(&raw, mode, Some(&decode)).unwrap();
let mut changed_metadata = frame.metadata().clone();
changed_metadata.insert("greptime.series.marker".to_string(), "changed".to_string());
let changed_schema = Arc::new(ArrowSchema::new_with_metadata(
@@ -3554,13 +3577,22 @@ mod tests {
)
.unwrap()
};
let first = frame::pack_raw_series_chunk(&raw, mode, &chunk([1.0, 2.0], [10, 20], false))
.unwrap()
.remove(0);
let mut second =
frame::pack_raw_series_chunk(&raw, mode, &chunk([3.0, 4.0], [30, 40], true))
.unwrap()
.remove(0);
let first = frame::pack_raw_series_chunk(
&raw,
mode,
Some(&decode),
&chunk([1.0, 2.0], [10, 20], false),
)
.unwrap()
.remove(0);
let mut second = frame::pack_raw_series_chunk(
&raw,
mode,
Some(&decode),
&chunk([3.0, 4.0], [30, 40], true),
)
.unwrap()
.remove(0);
second = DfRecordBatch::try_new(changed_schema, second.columns().to_vec()).unwrap();
let advertised = Arc::new(Schema::try_from(frame.clone()).unwrap());
let first = RecordBatch::from_df_record_batch(advertised.clone(), first);
@@ -3578,7 +3610,7 @@ mod tests {
);
let mut stream = exec.execute(0, Arc::new(TaskContext::default())).unwrap();
let emitted = stream.next().await.unwrap().unwrap();
let decoded = frame::unpack_raw_series_frame(&raw, mode, &emitted).unwrap();
let decoded = frame::unpack_raw_series_frame(&raw, mode, Some(&decode), &emitted).unwrap();
assert_eq!(decoded[0].time.timestamp(0), Some(10));
assert!(matches!(
decoded[0].fields.get(0).unwrap().sample(0),
@@ -3593,7 +3625,9 @@ mod tests {
async fn merge_scan_remote_schema_identity_rejects_raw_frame_field_metadata_mismatch() {
let region_id = RegionId::new(1024, 1);
let raw = raw_test_series_schema();
let expected = frame::raw_series_frame_schema(&raw, RawLabelMode::Key).unwrap();
let expected =
frame::series_frame_schema(&raw, FrameLabelMode::Key, Some(&raw_test_label_decode()))
.unwrap();
let frame_metadata = expected.metadata().clone();
let fields = expected.fields().to_vec();
let mut field_cases = Vec::new();
@@ -3678,7 +3712,9 @@ mod tests {
async fn merge_scan_remote_schema_identity_preserves_valid_raw_frame_payload() {
let region_id = RegionId::new(1024, 1);
let raw = raw_test_series_schema();
let expected_frame = frame::raw_series_frame_schema(&raw, RawLabelMode::Key).unwrap();
let decode = raw_test_label_decode();
let expected_frame =
frame::series_frame_schema(&raw, FrameLabelMode::Key, Some(&decode)).unwrap();
let mut expected_metadata = expected_frame.metadata().clone();
expected_metadata.insert("application.note".to_string(), "expected".to_string());
let expected = Arc::new(ArrowSchema::new_with_metadata(
@@ -3702,7 +3738,8 @@ mod tests {
true,
)
.unwrap();
let packed = frame::pack_raw_series_chunk(&raw, RawLabelMode::Key, &chunk).unwrap();
let packed =
frame::pack_raw_series_chunk(&raw, FrameLabelMode::Key, Some(&decode), &chunk).unwrap();
let inner = packed.into_iter().next().unwrap();
let original_columns = inner.columns().to_vec();
let batch = RecordBatch::from_df_record_batch(remote_schema.clone(), inner);
@@ -3722,7 +3759,8 @@ mod tests {
for (original, returned) in original_columns.iter().zip(result[0].columns()) {
assert!(Arc::ptr_eq(original, returned));
}
let decoded = frame::unpack_raw_series_frame(&raw, RawLabelMode::Key, &result[0]);
let decoded =
frame::unpack_raw_series_frame(&raw, FrameLabelMode::Key, Some(&decode), &result[0]);
let decoded = decoded.unwrap();
assert_eq!(decoded.len(), 1);
assert_eq!(decoded[0].primary_key.as_slice(), &[1, 2, 3]);
src/table/src/table/scan/series_tests.rs
+13 -13
View File
@@ -20,7 +20,7 @@ use std::task::{Context, Poll};
use api::v1::SemanticType;
use common_error::ext::BoxedError;
use common_query::request::RawSeriesRequest;
use common_query::series_frame::raw_series_frame_ordering;
use common_query::series_frame::series_frame_ordering;
use common_recordbatch::{
QueryMemoryTracker, RecordBatch, RecordBatches, SendableRecordBatchStream,
};
@@ -36,7 +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::frame::{FrameLabelMode, LabelDecodeSpec};
use datatypes::series::raw::{RawSeriesChunk, RawTimestamps};
use datatypes::series::samples::{FieldSpec, Samples, SeriesFields, SeriesSchema};
use datatypes::vectors::{Int32Vector, TimestampMillisecondVector, VectorRef};
@@ -673,7 +673,8 @@ async fn native_raw_frame_executes_and_decodes_bound_data() {
assert_eq!(frames[0].schema(), plan.schema());
let decoded = datatypes::series::frame::unpack_raw_series_frame(
&schema,
RawLabelMode::Explicit,
FrameLabelMode::Explicit,
None,
&frames[0],
)
.unwrap();
@@ -765,10 +766,7 @@ async fn planned_ordering_is_preserved_for_empty_and_nonempty_frames() {
.unwrap()
.unwrap();
let planned_schema = plan.schema();
assert_eq!(
ordering,
raw_series_frame_ordering(&planned_schema).unwrap()
);
assert_eq!(ordering, series_frame_ordering(&planned_schema).unwrap());
{
let state = state.lock().unwrap();
assert_eq!(0, state.raw_calls);
@@ -783,7 +781,7 @@ async fn planned_ordering_is_preserved_for_empty_and_nonempty_frames() {
);
assert_eq!(
ordering,
raw_series_frame_ordering(stream.schema().as_ref()).unwrap()
series_frame_ordering(stream.schema().as_ref()).unwrap()
);
let frames = stream.try_collect::<Vec<_>>().await.unwrap();
assert_eq!(usize::from(nonempty), frames.len());
@@ -791,7 +789,7 @@ async fn planned_ordering_is_preserved_for_empty_and_nonempty_frames() {
assert_eq!(planned_schema.as_ref(), frame.schema().as_ref());
assert_eq!(
ordering,
raw_series_frame_ordering(frame.schema().as_ref()).unwrap()
series_frame_ordering(frame.schema().as_ref()).unwrap()
);
}
let state = state.lock().unwrap();
@@ -839,7 +837,7 @@ async fn planned_and_stream_ordering_mismatches_fail_without_legacy_fallback() {
.unwrap();
assert_eq!(
planned,
raw_series_frame_ordering(plan.schema().as_ref()).unwrap()
series_frame_ordering(plan.schema().as_ref()).unwrap()
);
state.lock().unwrap().stream_ordering = Some(actual);
let error = plan
@@ -1171,7 +1169,8 @@ async fn error_after_emitted_frame_is_not_replayed() {
assert_eq!(frame.schema(), plan.schema());
let decoded = datatypes::series::frame::unpack_raw_series_frame(
plan.series_schema(),
RawLabelMode::Explicit,
FrameLabelMode::Explicit,
None,
&frame,
)
.unwrap();
@@ -1354,7 +1353,8 @@ async fn tracked_native_frame_holds_and_releases_quota_on_cancel() {
assert_eq!(
datatypes::series::frame::unpack_raw_series_frame(
plan.series_schema(),
RawLabelMode::Explicit,
FrameLabelMode::Explicit,
None,
&frame,
)
.unwrap()[0]
@@ -1484,7 +1484,7 @@ async fn raw_series_admission_captures_scanner_ordering_before_execution() {
.unwrap();
assert_eq!(plan.raw_series_ordering(), ordering);
assert_eq!(
raw_series_frame_ordering(plan.schema().as_ref()).unwrap(),
series_frame_ordering(plan.schema().as_ref()).unwrap(),
ordering
);
{
tests/compatibility/wire/raw_series_label_decode/README.md
+19 -7
View File
@@ -1,13 +1,14 @@
# Raw-series label decode description compatibility
`v1_metadata.json` is a hand-authored, frozen set of Arrow schema metadata maps
for the v1 raw-series frame. It records the established layout/content/labels/
time metadata plus the new `greptime.series.label_decode` entry, whose value is a
JSON label decode description with an explicit `version`. The Rust unit test
parses these literal maps through `raw_series_frame_label_decode`; the fixture is
not produced by the schema builder. `v1_physical_types.json`, described under
`v2_metadata.json` is a hand-authored, frozen set of Arrow schema metadata maps
for the v2 series frame. It records the established layout/content/labels/time
metadata plus the `greptime.series.label_decode` entry, whose value is a JSON
label decode description with an explicit `version`. The Rust unit test parses
these literal maps through `series_frame_label_decode`; the fixture is not
produced by the schema builder. `v1_physical_types.json`, described under
[Physical types](#physical-types), is the one generated fixture here: it records
the serde text of every `ConcreteDataType` variant.
the serde text of every `ConcreteDataType` variant, which the frame layout version
does not change, so it keeps its original name.
Run the focused check:
@@ -22,6 +23,17 @@ column, and that the reader rejects a key frame without a description, an explic
frame that declares one, an unknown version, a missing or unknown field, a wrong
field type, and a non-JSON value.
It also pins the declarations the unified frame family refuses. Layout is now
`v2`: a frame declaring `v1` or any other value is refused, and so is a
`greptime.series.*` key this revision does not register. The three modes are not
independent - `content` fixes `time`, and `labels` decides whether a description
must be present - so `content = evaluated` with `time = timestamps`, `content =
evaluated` with `labels = key`, and `content = raw` with a
`greptime.series.grid` are all contradictions and are refused rather than
half-trusted. An evaluated frame is therefore always `explicit` + `grid_spans` and
always declares its grid; a raw frame is always `timestamps` and never declares
one.
A timestamp-typed tag column has its own case in the crate's unit tests rather
than here, so this fixture stays a single fixed layout.
tests/compatibility/wire/raw_series_label_decode/v1_metadata.json → tests/compatibility/wire/raw_series_label_decode/v2_metadata.json
+67 -10
View File
@@ -2,7 +2,7 @@
{
"name": "key_frame_declares_dense_description",
"metadata": {
"greptime.series.layout": "v1",
"greptime.series.layout": "v2",
"greptime.series.content": "raw",
"greptime.series.labels": "key",
"greptime.series.time": "timestamps",
@@ -13,7 +13,7 @@
{
"name": "key_frame_declares_sparse_description",
"metadata": {
"greptime.series.layout": "v1",
"greptime.series.layout": "v2",
"greptime.series.content": "raw",
"greptime.series.labels": "key",
"greptime.series.time": "timestamps",
@@ -24,7 +24,7 @@
{
"name": "missing_description_rejected",
"metadata": {
"greptime.series.layout": "v1",
"greptime.series.layout": "v2",
"greptime.series.content": "raw",
"greptime.series.labels": "key",
"greptime.series.time": "timestamps"
@@ -34,7 +34,7 @@
{
"name": "unknown_version_rejected",
"metadata": {
"greptime.series.layout": "v1",
"greptime.series.layout": "v2",
"greptime.series.content": "raw",
"greptime.series.labels": "key",
"greptime.series.time": "timestamps",
@@ -45,7 +45,7 @@
{
"name": "missing_field_rejected",
"metadata": {
"greptime.series.layout": "v1",
"greptime.series.layout": "v2",
"greptime.series.content": "raw",
"greptime.series.labels": "key",
"greptime.series.time": "timestamps",
@@ -56,7 +56,7 @@
{
"name": "unknown_field_rejected",
"metadata": {
"greptime.series.layout": "v1",
"greptime.series.layout": "v2",
"greptime.series.content": "raw",
"greptime.series.labels": "key",
"greptime.series.time": "timestamps",
@@ -67,7 +67,7 @@
{
"name": "wrong_field_type_rejected",
"metadata": {
"greptime.series.layout": "v1",
"greptime.series.layout": "v2",
"greptime.series.content": "raw",
"greptime.series.labels": "key",
"greptime.series.time": "timestamps",
@@ -78,7 +78,7 @@
{
"name": "unlisted_logical_tag_rejected",
"metadata": {
"greptime.series.layout": "v1",
"greptime.series.layout": "v2",
"greptime.series.content": "raw",
"greptime.series.labels": "key",
"greptime.series.time": "timestamps",
@@ -89,7 +89,7 @@
{
"name": "explicit_frame_with_description_rejected",
"metadata": {
"greptime.series.layout": "v1",
"greptime.series.layout": "v2",
"greptime.series.content": "raw",
"greptime.series.labels": "explicit",
"greptime.series.time": "timestamps",
@@ -100,12 +100,69 @@
{
"name": "malformed_json_rejected",
"metadata": {
"greptime.series.layout": "v1",
"greptime.series.layout": "v2",
"greptime.series.content": "raw",
"greptime.series.labels": "key",
"greptime.series.time": "timestamps",
"greptime.series.label_decode": "greptime"
},
"encoding": null
},
{
"name": "wrong_layout_rejected",
"metadata": {
"greptime.series.layout": "v3",
"greptime.series.content": "raw",
"greptime.series.labels": "key",
"greptime.series.time": "timestamps",
"greptime.series.label_decode": "{\"version\":1,\"metric_name\":\"cpu_usage\",\"encoding\":\"dense\",\"primary_key\":[{\"column_id\":1,\"name\":\"host\",\"physical_type\":{\"String\":{\"size_type\":\"Utf8\"}}},{\"column_id\":2,\"name\":\"shard\",\"physical_type\":{\"UInt32\":{}}}],\"logical_tags\":[{\"column_id\":1,\"name\":\"host\"},{\"column_id\":2,\"name\":\"shard\"}]}"
},
"encoding": null
},
{
"name": "unregistered_reserved_metadata_rejected",
"metadata": {
"greptime.series.layout": "v2",
"greptime.series.content": "raw",
"greptime.series.labels": "key",
"greptime.series.time": "timestamps",
"greptime.series.label_decode": "{\"version\":1,\"metric_name\":\"cpu_usage\",\"encoding\":\"dense\",\"primary_key\":[{\"column_id\":1,\"name\":\"host\",\"physical_type\":{\"String\":{\"size_type\":\"Utf8\"}}},{\"column_id\":2,\"name\":\"shard\",\"physical_type\":{\"UInt32\":{}}}],\"logical_tags\":[{\"column_id\":1,\"name\":\"host\"},{\"column_id\":2,\"name\":\"shard\"}]}",
"greptime.series.future": "x"
},
"encoding": null
},
{
"name": "evaluated_with_timestamps_rejected",
"metadata": {
"greptime.series.layout": "v2",
"greptime.series.content": "evaluated",
"greptime.series.labels": "explicit",
"greptime.series.time": "timestamps",
"greptime.series.grid": "{\"start_ms\":0,\"step_ms\":10000,\"len\":4}"
},
"encoding": null
},
{
"name": "evaluated_with_key_labels_rejected",
"metadata": {
"greptime.series.layout": "v2",
"greptime.series.content": "evaluated",
"greptime.series.labels": "key",
"greptime.series.time": "grid_spans",
"greptime.series.label_decode": "{\"version\":1,\"metric_name\":\"cpu_usage\",\"encoding\":\"dense\",\"primary_key\":[{\"column_id\":1,\"name\":\"host\",\"physical_type\":{\"String\":{\"size_type\":\"Utf8\"}}},{\"column_id\":2,\"name\":\"shard\",\"physical_type\":{\"UInt32\":{}}}],\"logical_tags\":[{\"column_id\":1,\"name\":\"host\"},{\"column_id\":2,\"name\":\"shard\"}]}",
"greptime.series.grid": "{\"start_ms\":0,\"step_ms\":10000,\"len\":4}"
},
"encoding": null
},
{
"name": "raw_with_grid_rejected",
"metadata": {
"greptime.series.layout": "v2",
"greptime.series.content": "raw",
"greptime.series.labels": "explicit",
"greptime.series.time": "timestamps",
"greptime.series.grid": "{\"start_ms\":0,\"step_ms\":10000,\"len\":4}"
},
"encoding": null
}
]
tests/compatibility/wire/raw_series_ordering/README.md
+27 -9
View File
@@ -1,10 +1,11 @@
# Raw-series ordering schema metadata compatibility
`v1_metadata.json` is a hand-authored, frozen set of Arrow schema metadata maps
for the v1 raw-series frame. It records the established layout/content/labels/
time metadata and the new optional ordering entry. The Rust unit test parses
these literal maps through `raw_series_frame_ordering`; the fixture is not
produced by the schema builder.
`v2_metadata.json` is a hand-authored, frozen set of Arrow schema metadata maps
for the v2 series frame. It records the established layout/content/labels/time
metadata and the optional ordering entry, for raw frames and for evaluated ones.
The Rust unit test parses these literal maps through `series_frame_ordering` and
through `SeriesFrameDeclaration::parse`; the fixture is not produced by the
schema builder.
Run the focused check:
@@ -13,7 +14,24 @@ cargo nextest run -p common-query -E 'test(raw_series_ordering_metadata_fixture)
```
The fixture proves that absent ordering metadata defaults to `Contiguous`, the
two known values decode, and an unknown value is rejected. It does not prove
interoperability with an old binary, nor does it test an end-to-end client or
service. Separate pack/IPC tests check the declared ordering schema and batch
schema agreement for both orderings, including empty streams.
two known values decode, an unknown value is rejected, and that a whole
declaration is accepted or refused consistently with its layout and content: a
`v3` layout, an unregistered `greptime.series.*` key, a raw frame that declares a
grid, and any content/time or content/labels contradiction are refusals. The
evaluated cases are read as declarations only; the streaming round trips for
evaluated frames are covered by the crate's evaluated tests.
Because the declaration is cross-checked, a frame that names `evaluated` content
is only an evaluated frame when it also declares `explicit` labels, `grid_spans`
time, and a `greptime.series.grid`. `read_key_then_time` parses as a declaration
(a raw frame may legitimately claim it), but no writer produces one on an
evaluated frame: computed values have no storage key to be ordered by, so both the
schema builder and `unpack_evaluated_series` refuse it. The fixture keeps a
hand-written evaluated frame that states it, because a reader must still refuse
such a frame rather than decode it under an ordering its producer could not have
guaranteed.
The fixture does not prove interoperability with an old binary, nor does it test
an end-to-end client or service. Separate pack/IPC tests check the declared
ordering schema and batch schema agreement for both orderings, including empty
streams, and check the same for the evaluated frame across Arrow IPC.
tests/compatibility/wire/raw_series_ordering/v1_metadata.json
-45
View File
@@ -1,45 +0,0 @@
[
{
"name": "missing_defaults_to_contiguous",
"metadata": {
"greptime.series.layout": "v1",
"greptime.series.content": "raw",
"greptime.series.labels": "explicit",
"greptime.series.time": "timestamps"
},
"ordering": "contiguous"
},
{
"name": "known_contiguous",
"metadata": {
"greptime.series.layout": "v1",
"greptime.series.content": "raw",
"greptime.series.labels": "explicit",
"greptime.series.time": "timestamps",
"greptime.series.ordering": "contiguous"
},
"ordering": "contiguous"
},
{
"name": "known_read_key_then_time",
"metadata": {
"greptime.series.layout": "v1",
"greptime.series.content": "raw",
"greptime.series.labels": "explicit",
"greptime.series.time": "timestamps",
"greptime.series.ordering": "read_key_then_time"
},
"ordering": "read_key_then_time"
},
{
"name": "unknown_ordering_rejected",
"metadata": {
"greptime.series.layout": "v1",
"greptime.series.content": "raw",
"greptime.series.labels": "explicit",
"greptime.series.time": "timestamps",
"greptime.series.ordering": "future_value"
},
"ordering": null
}
]
tests/compatibility/wire/raw_series_ordering/v2_metadata.json
+98
View File
@@ -0,0 +1,98 @@
[
{
"name": "missing_defaults_to_contiguous",
"metadata": {
"greptime.series.layout": "v2",
"greptime.series.content": "raw",
"greptime.series.labels": "explicit",
"greptime.series.time": "timestamps"
},
"ordering": "contiguous",
"declaration": "accepted"
},
{
"name": "known_contiguous",
"metadata": {
"greptime.series.layout": "v2",
"greptime.series.content": "raw",
"greptime.series.labels": "explicit",
"greptime.series.time": "timestamps",
"greptime.series.ordering": "contiguous"
},
"ordering": "contiguous",
"declaration": "accepted"
},
{
"name": "known_read_key_then_time",
"metadata": {
"greptime.series.layout": "v2",
"greptime.series.content": "raw",
"greptime.series.labels": "explicit",
"greptime.series.time": "timestamps",
"greptime.series.ordering": "read_key_then_time"
},
"ordering": "read_key_then_time",
"declaration": "accepted"
},
{
"name": "unknown_ordering_rejected",
"metadata": {
"greptime.series.layout": "v2",
"greptime.series.content": "raw",
"greptime.series.labels": "explicit",
"greptime.series.time": "timestamps",
"greptime.series.ordering": "future_value"
},
"ordering": null,
"declaration": "accepted"
},
{
"name": "evaluated_contiguous",
"metadata": {
"greptime.series.layout": "v2",
"greptime.series.content": "evaluated",
"greptime.series.labels": "explicit",
"greptime.series.time": "grid_spans",
"greptime.series.grid": "{\"start_ms\":0,\"step_ms\":10000,\"len\":4}",
"greptime.series.ordering": "contiguous"
},
"ordering": "contiguous",
"declaration": "accepted"
},
{
"name": "evaluated_read_key_then_time_still_reads",
"metadata": {
"greptime.series.layout": "v2",
"greptime.series.content": "evaluated",
"greptime.series.labels": "explicit",
"greptime.series.time": "grid_spans",
"greptime.series.grid": "{\"start_ms\":0,\"step_ms\":10000,\"len\":4}",
"greptime.series.ordering": "read_key_then_time"
},
"ordering": "read_key_then_time",
"declaration": "accepted"
},
{
"name": "wrong_layout_rejected",
"metadata": {
"greptime.series.layout": "v3",
"greptime.series.content": "raw",
"greptime.series.labels": "explicit",
"greptime.series.time": "timestamps"
},
"ordering": "contiguous",
"declaration": "rejected"
},
{
"name": "unregistered_reserved_metadata_rejected",
"metadata": {
"greptime.series.layout": "v2",
"greptime.series.content": "raw",
"greptime.series.labels": "explicit",
"greptime.series.time": "timestamps",
"greptime.series.future": "x"
},
"ordering": "contiguous",
"declaration": "rejected"
}
]