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 (#9942)

Browse Source 
I hereby agree to the terms of the [GreptimeDB
CLA](https://github.com/GreptimeTeam/.github/blob/main/CLA.md).

## Refer to a related PR or issue link (optional)

- Spec: Narthex Unified SeriesFrame refactor, PR A of two.
- #9914, #9868, #9937, #9896, #9933, #9936, #9882 and #9886 are all
merged, so this is a plain increment on
  `narthex` with no stacked dependency.

## What's changed and what's your intention?

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` are unchanged; only the
adapter boundaries branch
on the sample shape.

### Base

- base: `narthex` @ `771347df5d9`
- HEAD: `8bc53c48714`
- **no stacked dependency**: everything this PR built on has landed

This is a plain increment on the current `narthex` tip, with focused
follow-up fixes for review
and CI. Nothing from the earlier stack is carried: #9914 is merged, so
no `MetricName` rename appears here. The codec itself, the
adapters, the tests and the two frozen wire fixtures are all that moved,
plus the call sites that
had to follow the renamed API.

### The format (`layout = v2`)

One frame family, three legal combinations. Every other combination is
refused on both the writing
and the reading side:

| content | identity | time | producer |
| --- | --- | --- | --- |
| `raw` | `key` | `timestamps` | a storage raw-series scan |
| `raw` | `explicit` | `timestamps` | a raw input whose labels are
already decoded |
| `evaluated` | `explicit` | `grid_spans` | a compute node's output |

`evaluated` + `key` is refused because a computation may rewrite its
labels, so it cannot claim a
storage key as its identity. `raw` + `grid_spans` and `evaluated` +
`timestamps` are refused because
raw samples have no grid and computed values have no timestamps of their
own.

Column layout (`series_frame_schema` in
`src/datatypes/src/series/frame.rs`):

```text
read_key: Binary (NOT NULL for key mode; NULLABLE for explicit mode)
is_last:  Boolean NOT NULL
time:     Struct<timestamps: List<Int64>>              // raw
          Struct<spans: List<Struct<start: u64, len: u64>>>  // evaluated
fields:   Struct<<field_name>: SamplePayload, ...>
name:     Utf8 NULLABLE                  // explicit only, __name__
labels:   List<Struct<name, value>>      // explicit only, no __name__, no duplicates
```

`read_key` must be populated for every raw row, including raw/explicit,
and must be NULL for
evaluated rows. `SamplePayload` (float / native histogram / mixed) keeps
its existing representation
and is not redesigned.

Protocol metadata lives under `greptime.series.*`.
`SeriesFrameDeclaration::parse` reads the
layout, identity, time and ordering declaration. Schemas explicitly
declare `contiguous` by default;
raw streams may declare `read_key_then_time`, while evaluated streams
accept only `contiguous`.
Missing ordering, unknown values and unknown reserved keys are refused
by declaration parsing,
direct decoding (including empty batches), and stream admission.

| key | value |
| --- | --- |
| `layout` | `v2` |
| `content` | `raw` / `evaluated` |
| `labels` | `key` / `explicit` |
| `time` | `timestamps` / `grid_spans` |
| `grid` | required for evaluated, forbidden for raw; canonical
`GridSpec` JSON |
| `label_decode` | required for key, forbidden for explicit |
| `ordering` | `contiguous` / `read_key_then_time` |

### Time axis

- raw: every row carries its own strictly increasing `i64` millisecond
timestamps.
- evaluated: the `GridSpec` is declared once in schema metadata
(`{start_ms, step_ms, len}`), and a row carries only the **absolute grid
positions** as spans.
A sparse grid is never expanded and its holes are never filled; `start +
len` is checked against
  the grid with checked arithmetic.
- A zero-sample closing chunk is legal for evaluated and **is kept** as
one frame row with zero
positions and `is_last = true`, because dropping it would drop the
finality it exists to carry.
A zero-sample chunk that closes nothing is refused, and a raw frame
still refuses an empty chunk.

### Series chains and `is_last`

`is_last` closes **one chunk chain**, not the end of a PromQL label set.
After a native label
rewrite, two different input series can produce **the same complete
labels**, and each closes
itself. The writer allows that and the reader does not treat "same
labels" as a re-opened series.
A chain that changes identity before closing, and EOF with an open
chain, are both errors.

### Key code locations

- `src/datatypes/src/series/frame.rs`: shared schema, declaration
parser, bounded builder,
  cursor and raw/evaluated decoder.
- `src/common/query/src/series_frame.rs`: typed pack/unpack adapters and
chain validators.
- `src/datatypes/src/series/time.rs`: canonical `GridSpec`
serialization.

### Review and CI fixes

- Repaired the two table PK-binding fixtures with real `host` and `rack`
Tag columns and matching
primary-key metadata. Production `LabelDecodeSpec::validate()` stays
strict. The mismatch test
retains all six cases, including changed tag order, and checks rejection
before polling.
- Removed the missing-ordering fallback and added one private ordering
validator shared by the
declaration parser and direct decoder. Invalid evaluated ordering cannot
bypass stream checks.
- Compatibility metadata is tested exactly as written, without merging
canonical metadata into
malformed inputs. The direct-decoder matrix covers raw/evaluated,
empty/nonempty, missing,
unknown and interleaved ordering, and valid raw versus invalid evaluated
`read_key_then_time`.
- Independent source and anti-over-design review: **APPROVE** for
`771347df5d9..8bc53c48714`. Review approval is separate from the CI
status of the new head.

### Limitations

- Not reachable from a query yet: `FusedSeriesExec` still answers
`PlainRows` by default. The
  second output format is PR B.
- Not verified: Windows; the results below are Linux.
- Deliberately **not** done, per the spec: evaluated **input** wiring in
`FusedSeriesExec`,
Complete Aggregate's SeriesFrame input, Partial/Final physical plans,
cross-region exchange,
remote NativeTree serialization, NodePath rework, #9892
storage-interface changes, #9844 remote
  read wiring.
- The unreleased v1 frame format is **not** read. In-tree producers,
consumers, tests and the two
  frozen wire fixtures all move to v2 together.

### Does this break API or data compatibility?

No published format changes. SST, WAL, manifest and the generic
Flight/Protobuf formats are
untouched. The only versioned thing here is the unreleased internal
SeriesFrame layout, which goes
from `v1` to `v2` in one step and is not read in both directions.

### Tests

The existing tests include real pack → batch stream → unpack round
trips, not only schema
snapshots:

- raw/key with its decoded PK, raw/explicit with complete labels, both
across batches and through
  Arrow IPC
- evaluated dense, sparse (multiple spans, holes not filled),
multi-field (a position can hold a
value in one field and nothing in another; NULL ≠ NaN), native histogram
and mixed payloads
  (payload kind and values survive, no silent float-ification)
- evaluated empty closing chunk kept as one row; multi-batch streams
keeping order and finality;
two consecutive chains with equal labels each closing independently; EOF
with an open chain
- malformed frames refused: non-increasing timestamps,
overlapping/out-of-order/out-of-bounds/zero
length spans, payload length mismatch, span-count overflow, missing
grid, malformed `GridSpec`,
missing label decode, contradictory metadata, wrong layout version, an
evaluated row carrying a
  storage key, an empty row that closes nothing

<!-- CURRENT_VERIFICATION_START -->
Verified on Linux at `8bc53c48714`; every command below exited 0:

```text
make fmt-check
  -> clean (rustfmt and repository source checks)
cargo nextest run -p datatypes -p common-query -p table --retries 0
  -> 690 passed, 0 skipped; includes both reported CI failures
cargo nextest run -p mito-codec -p promql -p query --retries 0
  -> 2055 passed, 2 skipped
cargo nextest run -p mito2 -E 'test(engine::scan_test::test_mito_handle_query_raw_series) | test(engine::scan_test::test_two_phase_native_raw_stream) | test(engine::scan_test::test_two_phase_raw_) | test(engine::scan_test::test_mito_raw_series_)' --retries 0
  -> 8 passed, 1733 skipped (focused raw-scan selection)
cargo check --workspace --all-targets
  -> passed
cargo clippy -p datatypes -p common-query -p table -p mito-codec -p promql -p query --all-targets -- -D warnings
  -> passed (scoped crates, default features)
```

License headers, TOML formatting and `git diff --check` also pass. These
are local Linux
results, not a claim that the new-head GitHub checks are green. Full
`make test`, workspace
all-features Clippy and Windows execution were not run locally in this
round.
<!-- CURRENT_VERIFICATION_END -->

## PR Checklist
Please convert it to a draft if some of the following conditions are not
met.

- [x] I have written the necessary rustdoc comments.
- [x] I have added the necessary unit tests and integration tests.
- [ ] This PR requires documentation updates.
- [x] API changes are backward compatible.
- [x] Schema or data changes are backward compatible.
- [ ] This PR needs to be backported to release branches, and I have
added the `backport-<target>` labels (e.g. `backport-v1.3` targets
`release/v1.3`).
This commit is contained in:
discord9
2026-10-26 14:51:05 +08:00
committed by GitHub
parent 23dc866868 8bc53c4871
commit eeca48bc43
19 changed files with 4494 additions and 1079 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/query/src/series_frame.rs
+532 -127
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,26 @@ pub fn label_decode_spec(source: &RawLabelSource) -> Option<LabelDecodeSpec> {
Some(description)
}
/// Reads the declared producer ordering, conservatively treating legacy frames as contiguous.
/// Reads the producer ordering declared by a complete v2 frame schema.
///
/// Only the two orderings representable on the wire can be returned.
pub fn raw_series_frame_ordering(
schema: &ArrowSchema,
) -> datafusion_common::Result<SeriesOrdering> {
/// Ordering is mandatory stream metadata; it is never inferred from frame contents.
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("contiguous") => Ok(SeriesOrdering::Contiguous),
Some("read_key_then_time") => Ok(SeriesOrdering::KeyThenTime),
Some(_) => Err(invalid("unknown raw series ordering")),
None | Some(_) => Err(invalid("missing or 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 +216,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 +296,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 +342,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 +397,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 +424,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 +440,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 +466,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 +498,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 +573,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 +676,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()) {
match DfRecordBatch::try_new(self.frame_schema.clone(), batch.columns().to_vec()) {
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 +745,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 +760,18 @@ 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 DfRecordBatch::try_new(declared, batch.columns().to_vec()) {
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 +785,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 +799,117 @@ 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 DfRecordBatch::try_new(declared, batch.columns().to_vec()) {
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 +948,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 +963,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 +980,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 +1003,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 +1011,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
+285 -104
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(),
@@ -672,7 +699,9 @@ async fn invalid_sequence_reports_valid_prefix_then_one_terminal_error() {
.unwrap();
let frames = frames
.into_iter()
.map(|frame| frame.with_schema(frame_schema.clone()).unwrap())
.map(|frame| {
RecordBatch::try_new(frame_schema.clone(), frame.columns().to_vec()).unwrap()
})
.collect();
let (source, _, _) = df_source(frames, frame_schema);
let mut decoded = unpack_raw_series(source, Some(schema())).unwrap();
@@ -692,7 +721,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 +953,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 +1006,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 +1049,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 +1093,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 +1108,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 +1127,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 +1157,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 +1215,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 +1278,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 +1290,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 +1300,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 +1336,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 +1450,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 +1476,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 +1501,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 +1530,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 +1607,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 +1631,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 +1823,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 +1899,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 +1953,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 +2003,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 +2021,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 +2076,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 +2175,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 +2365,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 +2399,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 +2423,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 +2444,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 +2492,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 +2502,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 +2517,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 +2582,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 +2603,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 +2621,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 +2737,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 +2852,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 +2878,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 +2888,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 +2931,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 +2978,15 @@ 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 carries its description and explicit default ordering.
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(),
"the stream schema preserves the complete canonical declaration"
);
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 +3001,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(),
"an explicit frame preserves the complete canonical declaration"
);
}
/// A key frame that omits its description is not decodable, and unpacking says so instead of falling
@@ -2899,13 +3033,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,
@@ -2922,8 +3054,18 @@ async fn unpack_round_trips_a_key_frame_and_rejects_one_without_a_description()
);
}
/// The ordering a frame declares is the only ordering a consumer has: unpacking a `KeyThenTime`
/// frame reports `KeyThenTime` although nobody passed it in.
/// An evaluated stream with no ordering declaration is rejected before the empty source is polled.
#[tokio::test]
async fn unpack_rejects_missing_ordering_on_empty_evaluated_stream() {
let frame = series_frame_schema(&evaluated_schema(), FrameLabelMode::Explicit, None).unwrap();
let mut metadata = frame.metadata().clone();
metadata.remove("greptime.series.ordering");
let frame = Arc::new(Schema::new_with_metadata(frame.fields().clone(), metadata));
let (source, polls, _) = df_source(vec![], frame);
assert!(super::unpack_evaluated_series(source, None).is_err());
assert_eq!(polls.load(Ordering::SeqCst), 0);
}
#[tokio::test]
async fn unpack_reports_the_ordering_the_frame_declares() {
let (input, _, _) = raw_custom(
@@ -2933,7 +3075,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"
);
@@ -2945,27 +3087,28 @@ async fn unpack_reports_the_ordering_the_frame_declares() {
"unpacking derives the ordering from the frame, not from a caller"
);
// 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 mut metadata = frame.metadata().clone();
metadata.remove("greptime.series.ordering");
let frame = Arc::new(Schema::new_with_metadata(frame.fields().clone(), metadata));
let (source, _, _) = df_source(vec![], frame);
assert_eq!(
unpack_raw_series(source, None).unwrap().ordering(),
SeriesOrdering::Contiguous
);
assert!(unpack_raw_series(source, None).is_err());
}
/// 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 +3125,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 +3192,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 +3202,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 +3220,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
+1170 -439
View File
File diff suppressed because it is too large Load Diff
src/datatypes/src/series/frame/tests.rs
+807 -223
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
+79 -43
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};
@@ -311,8 +311,37 @@ fn named_fixture() -> (
Arc<Mutex<State>>,
store_api::metadata::RegionMetadataRef,
) {
let schema = Arc::new(Schema::new(vec![
ColumnSchema::new("host", ConcreteDataType::string_datatype(), false),
named_fixture_with_tag_count(false)
}
fn named_fixture_with_two_tags() -> (
Arc<RegionScanExec>,
Arc<Mutex<State>>,
store_api::metadata::RegionMetadataRef,
) {
named_fixture_with_tag_count(true)
}
fn named_fixture_with_tag_count(
two_tags: bool,
) -> (
Arc<RegionScanExec>,
Arc<Mutex<State>>,
store_api::metadata::RegionMetadataRef,
) {
let mut columns = vec![ColumnSchema::new(
"host",
ConcreteDataType::string_datatype(),
false,
)];
if two_tags {
columns.push(ColumnSchema::new(
"rack",
ConcreteDataType::string_datatype(),
false,
));
}
columns.extend([
ColumnSchema::new("aux", ConcreteDataType::float64_datatype(), false),
ColumnSchema::new("value", ConcreteDataType::float64_datatype(), false),
ColumnSchema::new(
@@ -320,25 +349,34 @@ fn named_fixture() -> (
ConcreteDataType::timestamp_millisecond_datatype(),
false,
),
]));
let batch = RecordBatch::new(
schema.clone(),
vec![
Arc::new(datatypes::vectors::StringVector::from_slice(&["node"])) as VectorRef,
Arc::new(datatypes::vectors::Float64Vector::from_slice([2.0])) as VectorRef,
Arc::new(datatypes::vectors::Float64Vector::from_slice([1.0])) as VectorRef,
Arc::new(TimestampMillisecondVector::from_slice([42])) as VectorRef,
],
)
.unwrap();
]);
let schema = Arc::new(Schema::new(columns));
let mut vectors: Vec<VectorRef> =
vec![Arc::new(datatypes::vectors::StringVector::from_slice(&["node"])) as VectorRef];
if two_tags {
vectors.push(Arc::new(datatypes::vectors::StringVector::from_slice(&["r1"])) as VectorRef);
}
vectors.extend([
Arc::new(datatypes::vectors::Float64Vector::from_slice([2.0])) as VectorRef,
Arc::new(datatypes::vectors::Float64Vector::from_slice([1.0])) as VectorRef,
Arc::new(TimestampMillisecondVector::from_slice([42])) as VectorRef,
]);
let batch = RecordBatch::new(schema.clone(), vectors).unwrap();
let batches = RecordBatches::try_new(schema, vec![batch]).unwrap();
let mut builder = RegionMetadataBuilder::new(RegionId::new(1, 2));
builder
.push_column_metadata(ColumnMetadata {
column_schema: ColumnSchema::new("host", ConcreteDataType::string_datatype(), false),
builder.push_column_metadata(ColumnMetadata {
column_schema: ColumnSchema::new("host", ConcreteDataType::string_datatype(), false),
semantic_type: SemanticType::Tag,
column_id: 7,
});
if two_tags {
builder.push_column_metadata(ColumnMetadata {
column_schema: ColumnSchema::new("rack", ConcreteDataType::string_datatype(), false),
semantic_type: SemanticType::Tag,
column_id: 7,
})
column_id: 8,
});
}
builder
.push_column_metadata(ColumnMetadata {
column_schema: ColumnSchema::new("aux", ConcreteDataType::float64_datatype(), false),
semantic_type: SemanticType::Field,
@@ -357,8 +395,8 @@ fn named_fixture() -> (
),
semantic_type: SemanticType::Timestamp,
column_id: 13,
})
.primary_key(vec![7]);
});
builder.primary_key(if two_tags { vec![7, 8] } else { vec![7] });
let metadata = Arc::new(builder.build().unwrap());
let raw_schema = Arc::new(SeriesSchema {
fields: vec![
@@ -673,7 +711,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 +804,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 +819,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 +827,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 +875,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
@@ -1032,18 +1068,17 @@ async fn malformed_returned_stream_binding_fails_before_poll_even_when_empty() {
async fn primary_key_binding_mismatch_is_rejected_before_poll() {
let logical_tags: Arc<[BoundTag]> = vec![
BoundTag {
column_id: 1,
name: Arc::from("zone"),
column_id: 7,
name: Arc::from("host"),
},
BoundTag {
column_id: 2,
column_id: 8,
name: Arc::from("rack"),
},
]
.into();
for case in 0..6 {
let (scan, state) = fixture(Some(series_schema()), vec![]);
let metadata = scan.scanner.lock().unwrap().metadata();
let (scan, state, metadata) = named_fixture_with_two_tags();
let bound = RawLabelSource::PrimaryKey {
metric_name: Arc::from("cpu"),
physical_metadata: metadata.clone(),
@@ -1080,18 +1115,17 @@ async fn primary_key_binding_mismatch_is_rejected_before_poll() {
#[tokio::test]
async fn unchanged_primary_key_binding_accepts_empty_stream() {
let (scan, state) = fixture(Some(series_schema()), vec![]);
let metadata = scan.scanner.lock().unwrap().metadata();
let (scan, state, metadata) = named_fixture_with_two_tags();
let source = RawLabelSource::PrimaryKey {
metric_name: Arc::from("cpu"),
physical_metadata: metadata,
logical_tags: Arc::from([
BoundTag {
column_id: 1,
name: Arc::from("zone"),
column_id: 7,
name: Arc::from("host"),
},
BoundTag {
column_id: 2,
column_id: 8,
name: Arc::from("rack"),
},
]),
@@ -1171,7 +1205,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 +1389,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 +1520,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
+92 -20
View File
@@ -2,109 +2,181 @@
{
"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",
"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.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.ordering": "contiguous"
},
"encoding": "dense"
},
{
"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",
"greptime.series.label_decode": "{\"version\":1,\"metric_name\":\"cpu_usage\",\"encoding\":\"sparse\",\"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.label_decode": "{\"version\":1,\"metric_name\":\"cpu_usage\",\"encoding\":\"sparse\",\"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.ordering": "contiguous"
},
"encoding": "sparse"
},
{
"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"
"greptime.series.time": "timestamps",
"greptime.series.ordering": "contiguous"
},
"encoding": null
},
{
"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",
"greptime.series.label_decode": "{\"version\":2,\"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.label_decode": "{\"version\":2,\"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.ordering": "contiguous"
},
"encoding": null
},
{
"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",
"greptime.series.label_decode": "{\"version\":1,\"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.label_decode": "{\"version\":1,\"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.ordering": "contiguous"
},
"encoding": null
},
{
"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",
"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\"}],\"future\":0}"
"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\"}],\"future\":0}",
"greptime.series.ordering": "contiguous"
},
"encoding": null
},
{
"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",
"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.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.ordering": "contiguous"
},
"encoding": null
},
{
"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",
"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\"}}}],\"logical_tags\":[{\"column_id\":1,\"name\":\"host\"},{\"column_id\":2,\"name\":\"shard\"}]}"
"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\"}}}],\"logical_tags\":[{\"column_id\":1,\"name\":\"host\"},{\"column_id\":2,\"name\":\"shard\"}]}",
"greptime.series.ordering": "contiguous"
},
"encoding": null
},
{
"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",
"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.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.ordering": "contiguous"
},
"encoding": null
},
{
"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"
"greptime.series.label_decode": "greptime",
"greptime.series.ordering": "contiguous"
},
"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\"}]}",
"greptime.series.ordering": "contiguous"
},
"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",
"greptime.series.ordering": "contiguous"
},
"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}",
"greptime.series.ordering": "contiguous"
},
"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}",
"greptime.series.ordering": "contiguous"
},
"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}",
"greptime.series.ordering": "contiguous"
},
"encoding": null
}
tests/compatibility/wire/raw_series_ordering/README.md
+28 -10
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 mandatory 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:
@@ -12,8 +13,25 @@ Run the focused check:
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.
The fixture proves that missing ordering metadata and unknown values are rejected,
the two known values decode, 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.
Evaluated declarations use explicit `contiguous` ordering; 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 and verifies that declaration
parsing refuses it.
The fixture also pins rejection of evaluated `read_key_then_time` metadata. It
does not prove interoperability with an old binary or 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
+100
View File
@@ -0,0 +1,100 @@
[
{
"name": "missing_ordering_rejected",
"metadata": {
"greptime.series.layout": "v2",
"greptime.series.content": "raw",
"greptime.series.labels": "explicit",
"greptime.series.time": "timestamps"
},
"ordering": null,
"declaration": "rejected"
},
{
"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": "rejected"
},
{
"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_rejected",
"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": "rejected"
},
{
"name": "wrong_layout_rejected",
"metadata": {
"greptime.series.layout": "v3",
"greptime.series.content": "raw",
"greptime.series.labels": "explicit",
"greptime.series.time": "timestamps",
"greptime.series.ordering": "contiguous"
},
"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",
"greptime.series.ordering": "contiguous"
},
"ordering": "contiguous",
"declaration": "rejected"
}
]