diff --git a/rust/crates/sift_cli/CHANGELOG.md b/rust/crates/sift_cli/CHANGELOG.md index 59dea0a8a2..331e13cdf3 100644 --- a/rust/crates/sift_cli/CHANGELOG.md +++ b/rust/crates/sift_cli/CHANGELOG.md @@ -7,6 +7,10 @@ This project adheres to [Semantic Versioning](http://semver.org/). ### What's New +- `get_data` accepts `include_received_at` (alias `includeReceivedAt`). When + set, and the data service returns `sift_received_at`, the Parquet file + includes a receipt-time column per channel in unix nanos next to the + generated `timestamp_unix_nanos`. `upload_dataset` skips those columns. - `create_artifact` and the Sift skill now say which link relation to use: `source` for the run or asset an artifact was built from, and `derived_from` only for another artifact. Agents no longer send `derived_from` for a run and diff --git a/rust/crates/sift_cli/assets/skills/sift/SKILL.md b/rust/crates/sift_cli/assets/skills/sift/SKILL.md index fc50868944..fcb4338112 100644 --- a/rust/crates/sift_cli/assets/skills/sift/SKILL.md +++ b/rust/crates/sift_cli/assets/skills/sift/SKILL.md @@ -109,6 +109,13 @@ tool. to enable them, then restart the MCP client. - **Data:** `get_data` writes channel data to a Parquet file. `sql` queries Parquet files. `upload_dataset` streams a Parquet dataset into Sift. +- **Compare generation time with receipt time.** Call `get_data` with + `include_received_at: true` and `sample_ms: 0`. When Sift has a receipt time, + the file includes an Int64 column whose name contains `received_at="true"`, + in unix nanos, aligned with that channel's samples. `timestamp_unix_nanos` is + still when the measurement was generated. The receipt column is absent when + the service has no receipt time, including enum and bit-field channels. + `upload_dataset` skips receipt columns. - **Links:** `explore_url`. - **Docs:** `search_docs`. - **Writes:** `create_rule`, `update_rule`, `archive_rule`, `unarchive_rule`, diff --git a/rust/crates/sift_cli/assets/skills/sift/references/cli.md b/rust/crates/sift_cli/assets/skills/sift/references/cli.md index b2a88a50c0..6aac7c09ff 100644 --- a/rust/crates/sift_cli/assets/skills/sift/references/cli.md +++ b/rust/crates/sift_cli/assets/skills/sift/references/cli.md @@ -129,7 +129,9 @@ If the CLI does not support the user's file type: Do not reach for the MCP `upload_dataset` tool to import a user's file. It accepts only Sift's canonical Parquet schema — every column named ` {channel_id=""}` — so it round-trips the output of -`get_data`, and rejects a foreign file with `missing attribute block`. +`get_data`, and rejects a foreign file with `missing attribute block`. A +`received_at="true"` column from `get_data` is a receipt time, not a channel, +and upload skips it. ### Stream data diff --git a/rust/crates/sift_mcp/src/service/common/mod.rs b/rust/crates/sift_mcp/src/service/common/mod.rs index 07f82f3bbf..a29aee137b 100644 --- a/rust/crates/sift_mcp/src/service/common/mod.rs +++ b/rust/crates/sift_mcp/src/service/common/mod.rs @@ -74,9 +74,10 @@ pub fn secs_and_subsec_nanos_to_unix_nanos(sec: i64, subsec_nanos: i32) -> i64 { /// A fully-qualified Arrow column name with the channel-specific fields the /// data pipeline cares about. `Display` emits the canonical Parquet column-name -/// string: ` {channel_id="...", bit_field_element="...", run="...", units="..."}`. -/// Empty optional fields are omitted. Construct via [`ColumnName::builder`] or -/// `TryFrom` (when re-hydrating a string from a Parquet schema). +/// string: ` {channel_id="...", bit_field_element="...", run="...", units="...", received_at="true"}`. +/// Empty optional fields are omitted. `received_at="true"` marks a receipt-time +/// column rather than the channel's samples. Construct via [`ColumnName::builder`] or +/// `TryFrom<&str>` (when re-hydrating a string from a Parquet schema). #[derive(Debug, Clone, PartialEq, Eq, Hash)] pub struct ColumnName { name: String, @@ -84,6 +85,7 @@ pub struct ColumnName { bit_field_element: Option, run: Option, units: Option, + received_at: bool, } impl ColumnName { @@ -94,6 +96,20 @@ impl ColumnName { bit_field_element: None, run: None, units: None, + received_at: false, + } + } + + /// The same channel identity, marked as that channel's receipt-time column. + /// Units are dropped: the column holds unix nanos, not the measurement unit. + pub fn with_received_at(&self) -> Self { + Self { + name: self.name.clone(), + channel_id: self.channel_id.clone(), + bit_field_element: self.bit_field_element.clone(), + run: self.run.clone(), + units: None, + received_at: true, } } @@ -116,6 +132,10 @@ impl ColumnName { pub fn units(&self) -> Option<&str> { self.units.as_deref() } + + pub fn received_at(&self) -> bool { + self.received_at + } } impl fmt::Display for ColumnName { @@ -130,6 +150,9 @@ impl fmt::Display for ColumnName { if let Some(v) = &self.units { write!(f, ", units=\"{v}\"")?; } + if self.received_at { + write!(f, ", received_at=\"true\"")?; + } write!(f, "}}") } } @@ -157,6 +180,7 @@ impl TryFrom<&str> for ColumnName { let mut bit_field_element: Option = None; let mut run: Option = None; let mut units: Option = None; + let mut received_at = false; for segment in attrs.split(',') { let segment = segment.trim(); @@ -183,6 +207,12 @@ impl TryFrom<&str> for ColumnName { "bit_field_element" => bit_field_element = Some(val), "run" => run = Some(val), "units" => units = Some(val), + "received_at" => { + if val != "true" { + bail!("`received_at` must be \"true\""); + } + received_at = true; + } other => bail!("unknown attribute key `{other}`"), } } @@ -196,6 +226,7 @@ impl TryFrom<&str> for ColumnName { bit_field_element, run, units, + received_at, }) } } @@ -210,6 +241,7 @@ pub struct ColumnNameBuilder<'a> { bit_field_element: Option, run: Option, units: Option, + received_at: bool, } impl<'a> ColumnNameBuilder<'a> { @@ -228,6 +260,11 @@ impl<'a> ColumnNameBuilder<'a> { self } + pub fn received_at(mut self, value: bool) -> Self { + self.received_at = value; + self + } + pub fn build(self) -> ColumnName { ColumnName { name: self.name.to_string(), @@ -235,6 +272,7 @@ impl<'a> ColumnNameBuilder<'a> { bit_field_element: self.bit_field_element, run: self.run, units: self.units, + received_at: self.received_at, } } } diff --git a/rust/crates/sift_mcp/src/service/common/test.rs b/rust/crates/sift_mcp/src/service/common/test.rs index 5ff798ea7c..eff14118a7 100644 --- a/rust/crates/sift_mcp/src/service/common/test.rs +++ b/rust/crates/sift_mcp/src/service/common/test.rs @@ -137,6 +137,33 @@ fn column_name_try_from_missing_channel_id_errors() { ); } +#[test] +fn column_name_received_at_round_trips_and_drops_units() { + let original = ColumnName::builder("temp", "c1") + .run(Some("r42")) + .units(Some("C")) + .build() + .with_received_at(); + assert!(original.received_at()); + assert_eq!(original.units(), None); + assert_eq!( + original.to_string(), + "temp {channel_id=\"c1\", run=\"r42\", received_at=\"true\"}" + ); + let parsed = ColumnName::try_from(original.to_string().as_str()).expect("should parse"); + assert_eq!(parsed, original); +} + +#[test] +fn column_name_try_from_rejects_received_at_other_than_true() { + let err = ColumnName::try_from("temp {channel_id=\"c1\", received_at=\"false\"}") + .expect_err("only true is a receipt column"); + assert!( + err.to_string().contains("`received_at` must be \"true\""), + "unexpected error: {err}" + ); +} + #[test] fn column_name_try_from_unknown_key_errors() { let err = ColumnName::try_from("temp {channel_id=\"c1\", flavor=\"strawberry\"}") diff --git a/rust/crates/sift_mcp/src/service/data/mod.rs b/rust/crates/sift_mcp/src/service/data/mod.rs index 4b3bf2cf9e..9c0eb95dcf 100644 --- a/rust/crates/sift_mcp/src/service/data/mod.rs +++ b/rust/crates/sift_mcp/src/service/data/mod.rs @@ -29,10 +29,11 @@ use sift_rs::{ channels::v3::Channel, data::v2::{ BitFieldElementValues, BitFieldValue, BitFieldValues, BoolValue, BoolValues, BytesValue, - BytesValues, CalculatedChannelQuery, ChannelQuery, DoubleValue, DoubleValues, EnumValue, - EnumValues, FloatValue, FloatValues, GetDataRequest, GetDataResponse, Int32Value, - Int32Values, Int64Value, Int64Values, Query, StringValue, StringValues, Uint32Value, - Uint32Values, Uint64Value, Uint64Values, data_service_client::DataServiceClient, metadata, + BytesValues, CalculatedChannelQuery, ChannelQuery, DimensionIndividualWrapper, DoubleValue, + DoubleValues, EnumValue, EnumValues, FloatValue, FloatValues, GetDataRequest, + GetDataResponse, Int32Value, Int32Values, Int64Value, Int64Values, Query, StringValue, + StringValues, Uint32Value, Uint32Values, Uint64Value, Uint64Values, + data_service_client::DataServiceClient, dimension_individual_wrapper, metadata, query::Query as QueryKind, }, runs::v2::Run, @@ -175,13 +176,152 @@ pub enum TimeRange { }, } +struct OutputCursor { + data_type: DataType, + samples: std::iter::Peekable>, + received_at: Option>>, + received_index: usize, + value_builder: usize, + received_builder: Option, +} + struct ChannelColumn { json_metadata: Option, /// unix nanos to value values: Vec<(i64, ChannelValue)>, + /// Receipt time in unix nanos, aligned with `values`, when this channel + /// returned `sift_received_at`. `None` means the service sent no receipt times. + received_at: Option>>, data_type: DataType, } +const SIFT_RECEIVED_AT_LABEL: &str = "sift_received_at"; + +/// Index-aligned receipt times from a `sift_received_at` extra, or `None` when +/// that extra is absent. Entries line up with the protobuf values, including +/// samples whose generation timestamp is missing. +fn received_at_by_index( + extras: &[DimensionIndividualWrapper], + len: usize, +) -> Option>> { + let extra = extras + .iter() + .find(|extra| extra.label == SIFT_RECEIVED_AT_LABEL)?; + let Some(dimension_individual_wrapper::ValueWrapper::DimensionProtoTimestampValues(stamps)) = + &extra.value_wrapper + else { + return None; + }; + + Some( + (0..len) + .map(|index| { + stamps.values.get(index).and_then(|sample| { + sample.value.map(|Timestamp { seconds, nanos }| { + secs_and_subsec_nanos_to_unix_nanos(seconds, nanos) + }) + }) + }) + .collect(), + ) +} + +/// Drops samples with no generation timestamp and keeps receipt times aligned +/// with the samples that remain. +fn take_samples( + values: I, + extras: &[DimensionIndividualWrapper], + include_received_at: bool, + to_value: impl Fn(T) -> ChannelValue, +) -> (Vec<(i64, ChannelValue)>, Option>>) +where + I: IntoIterator, T)>, +{ + let values = values.into_iter().collect::>(); + let received = if include_received_at { + received_at_by_index(extras, values.len()) + } else { + None + }; + + let mut samples = Vec::new(); + let mut received_out = received.as_ref().map(|_| Vec::new()); + for (index, (timestamp, value)) in values.into_iter().enumerate() { + let Some(timestamp) = timestamp else { + continue; + }; + samples.push(( + secs_and_subsec_nanos_to_unix_nanos(timestamp.seconds, timestamp.nanos), + to_value(value), + )); + if let Some(out) = received_out.as_mut() { + out.push( + received + .as_ref() + .and_then(|series| series.get(index).copied()) + .flatten(), + ); + } + } + (samples, received_out) +} + +fn new_column_builder(data_type: &DataType) -> Result> { + Ok(match data_type { + DataType::Float32 => Box::new(Float32Builder::new()), + DataType::Float64 => Box::new(Float64Builder::new()), + DataType::Int32 => Box::new(Int32Builder::new()), + DataType::Int64 => Box::new(Int64Builder::new()), + DataType::UInt32 => Box::new(UInt32Builder::new()), + DataType::UInt64 => Box::new(UInt64Builder::new()), + DataType::Utf8 => Box::new(StringBuilder::new()), + DataType::Boolean => Box::new(BooleanBuilder::new()), + DataType::Binary => Box::new(BinaryBuilder::new()), + other => bail!("unsupported column data type: {other:?}"), + }) +} + +fn store_column( + columns: &mut HashMap, + column_name: ColumnName, + data_type: DataType, + json_metadata: Option, + samples: Vec<(i64, ChannelValue)>, + received_at: Option>>, +) { + let received_at = received_at.map(|mut series| { + series.truncate(samples.len()); + if series.len() < samples.len() { + series.resize(samples.len(), None); + } + series + }); + + if let Some(column) = columns.get_mut(&column_name) { + match (column.received_at.as_mut(), received_at) { + (Some(existing), Some(incoming)) => existing.extend(incoming), + (Some(existing), None) => existing.resize(existing.len() + samples.len(), None), + (None, Some(incoming)) => { + let mut series = vec![None; column.values.len()]; + series.extend(incoming); + column.received_at = Some(series); + } + (None, None) => {} + } + column.values.extend(samples); + } else { + columns.insert( + column_name, + ChannelColumn { + values: samples, + json_metadata, + data_type, + received_at, + }, + ); + } +} + enum ChannelValue { F32(f32), F64(f64), @@ -247,12 +387,15 @@ impl DataService { } /// Retrieves data for provided parameters and writes out the data in a buffered manner to the - /// provided buffer as arrow. The first column is int64, `timestamp_unix_nanos`. + /// provided buffer as arrow. The first column is int64, `timestamp_unix_nanos` (generation + /// time). When `include_received_at` is set, each channel that returns `sift_received_at` + /// also gets an Int64 column of receipt time in unix nanos. pub async fn get_data( &self, channel_inputs: &[ChannelInput], time_range: TimeRange, sample_ms: u32, + include_received_at: bool, buffer: &mut W, ) -> Result { if channel_inputs.is_empty() { @@ -388,7 +531,7 @@ impl DataService { start_time: Some(start_time), end_time: Some(end_time), page_size: GET_DATA_PAGE_SIZE, - include_received_at: None, + include_received_at: include_received_at.then_some(true), }) .await .map(|resp| resp.into_inner()) @@ -408,7 +551,7 @@ impl DataService { let BytesValues { metadata, values, - extras: _, + extras, } = BytesValues::decode(channel_page.value) .context("failed to decode channel page")?; @@ -427,30 +570,22 @@ impl DataService { .units(channel.unit.as_ref().map(|u| u.name.as_str())) .build(); - let values = values - .into_iter() - .flat_map(|BytesValue { timestamp, value }| { - timestamp.map(|Timestamp { seconds, nanos }| { - ( - secs_and_subsec_nanos_to_unix_nanos(seconds, nanos), - ChannelValue::Bytes(value), - ) - }) - }) - .collect::>(); - - if let Some(channel_column) = columns.get_mut(&column_name) { - channel_column.values.extend(values); - } else { - columns.insert( - column_name, - ChannelColumn { - values, - json_metadata: None, - data_type: DataType::Binary, - }, - ); - } + let (values, received_at) = take_samples( + values + .into_iter() + .map(|BytesValue { timestamp, value }| (timestamp, value)), + &extras, + include_received_at, + ChannelValue::Bytes, + ); + store_column( + &mut columns, + column_name, + DataType::Binary, + None, + values, + received_at, + ); } "sift.data.v2.EnumValues" => { let EnumValues { metadata, values } = @@ -501,18 +636,14 @@ impl DataService { }) .collect::>(); - if let Some(channel_column) = columns.get_mut(&column_name) { - channel_column.values.extend(values); - } else { - columns.insert( - column_name, - ChannelColumn { - values, - json_metadata: Some(enum_config_md), - data_type: DataType::Utf8, - }, - ); - } + store_column( + &mut columns, + column_name, + DataType::Utf8, + Some(enum_config_md), + values, + None, + ); } "sift.data.v2.BitFieldValues" => { let BitFieldValues { metadata, values } = @@ -559,25 +690,21 @@ impl DataService { }) .collect::>(); - if let Some(channel_column) = columns.get_mut(&column_name) { - channel_column.values.extend(values); - } else { - columns.insert( - column_name, - ChannelColumn { - values, - json_metadata: Some(bit_field_element_md.clone()), - data_type: DataType::UInt32, - }, - ); - } + store_column( + &mut columns, + column_name, + DataType::UInt32, + Some(bit_field_element_md.clone()), + values, + None, + ); } } "sift.data.v2.DoubleValues" => { let DoubleValues { metadata, values, - extras: _, + extras, } = DoubleValues::decode(channel_page.value) .context("failed to decode channel page")?; @@ -596,36 +723,28 @@ impl DataService { .units(channel.unit.as_ref().map(|u| u.name.as_str())) .build(); - let values = values - .into_iter() - .flat_map(|DoubleValue { timestamp, value }| { - timestamp.map(|Timestamp { seconds, nanos }| { - ( - secs_and_subsec_nanos_to_unix_nanos(seconds, nanos), - ChannelValue::F64(value), - ) - }) - }) - .collect::>(); - - if let Some(channel_column) = columns.get_mut(&column_name) { - channel_column.values.extend(values); - } else { - columns.insert( - column_name, - ChannelColumn { - values, - json_metadata: None, - data_type: DataType::Float64, - }, - ); - } + let (values, received_at) = take_samples( + values + .into_iter() + .map(|DoubleValue { timestamp, value }| (timestamp, value)), + &extras, + include_received_at, + ChannelValue::F64, + ); + store_column( + &mut columns, + column_name, + DataType::Float64, + None, + values, + received_at, + ); } "sift.data.v2.FloatValues" => { let FloatValues { metadata, values, - extras: _, + extras, } = FloatValues::decode(channel_page.value) .context("failed to decode channel page")?; @@ -644,36 +763,28 @@ impl DataService { .units(channel.unit.as_ref().map(|u| u.name.as_str())) .build(); - let values = values - .into_iter() - .flat_map(|FloatValue { timestamp, value }| { - timestamp.map(|Timestamp { seconds, nanos }| { - ( - secs_and_subsec_nanos_to_unix_nanos(seconds, nanos), - ChannelValue::F32(value), - ) - }) - }) - .collect::>(); - - if let Some(channel_column) = columns.get_mut(&column_name) { - channel_column.values.extend(values); - } else { - columns.insert( - column_name, - ChannelColumn { - values, - json_metadata: None, - data_type: DataType::Float32, - }, - ); - } + let (values, received_at) = take_samples( + values + .into_iter() + .map(|FloatValue { timestamp, value }| (timestamp, value)), + &extras, + include_received_at, + ChannelValue::F32, + ); + store_column( + &mut columns, + column_name, + DataType::Float32, + None, + values, + received_at, + ); } "sift.data.v2.StringValues" => { let StringValues { metadata, values, - extras: _, + extras, } = StringValues::decode(channel_page.value) .context("failed to decode channel page")?; @@ -692,36 +803,28 @@ impl DataService { .units(channel.unit.as_ref().map(|u| u.name.as_str())) .build(); - let values = values - .into_iter() - .flat_map(|StringValue { timestamp, value }| { - timestamp.map(|Timestamp { seconds, nanos }| { - ( - secs_and_subsec_nanos_to_unix_nanos(seconds, nanos), - ChannelValue::String(value), - ) - }) - }) - .collect::>(); - - if let Some(channel_column) = columns.get_mut(&column_name) { - channel_column.values.extend(values); - } else { - columns.insert( - column_name, - ChannelColumn { - values, - json_metadata: None, - data_type: DataType::Utf8, - }, - ); - } + let (values, received_at) = take_samples( + values + .into_iter() + .map(|StringValue { timestamp, value }| (timestamp, value)), + &extras, + include_received_at, + ChannelValue::String, + ); + store_column( + &mut columns, + column_name, + DataType::Utf8, + None, + values, + received_at, + ); } "sift.data.v2.BoolValues" => { let BoolValues { metadata, values, - extras: _, + extras, } = BoolValues::decode(channel_page.value) .context("failed to decode channel page")?; @@ -740,36 +843,28 @@ impl DataService { .units(channel.unit.as_ref().map(|u| u.name.as_str())) .build(); - let values = values - .into_iter() - .flat_map(|BoolValue { timestamp, value }| { - timestamp.map(|Timestamp { seconds, nanos }| { - ( - secs_and_subsec_nanos_to_unix_nanos(seconds, nanos), - ChannelValue::Bool(value), - ) - }) - }) - .collect::>(); - - if let Some(channel_column) = columns.get_mut(&column_name) { - channel_column.values.extend(values); - } else { - columns.insert( - column_name, - ChannelColumn { - values, - json_metadata: None, - data_type: DataType::Boolean, - }, - ); - } + let (values, received_at) = take_samples( + values + .into_iter() + .map(|BoolValue { timestamp, value }| (timestamp, value)), + &extras, + include_received_at, + ChannelValue::Bool, + ); + store_column( + &mut columns, + column_name, + DataType::Boolean, + None, + values, + received_at, + ); } "sift.data.v2.Int32Values" => { let Int32Values { metadata, values, - extras: _, + extras, } = Int32Values::decode(channel_page.value) .context("failed to decode channel page")?; @@ -788,36 +883,28 @@ impl DataService { .units(channel.unit.as_ref().map(|u| u.name.as_str())) .build(); - let values = values - .into_iter() - .flat_map(|Int32Value { timestamp, value }| { - timestamp.map(|Timestamp { seconds, nanos }| { - ( - secs_and_subsec_nanos_to_unix_nanos(seconds, nanos), - ChannelValue::I32(value), - ) - }) - }) - .collect::>(); - - if let Some(channel_column) = columns.get_mut(&column_name) { - channel_column.values.extend(values); - } else { - columns.insert( - column_name, - ChannelColumn { - values, - json_metadata: None, - data_type: DataType::Int32, - }, - ); - } + let (values, received_at) = take_samples( + values + .into_iter() + .map(|Int32Value { timestamp, value }| (timestamp, value)), + &extras, + include_received_at, + ChannelValue::I32, + ); + store_column( + &mut columns, + column_name, + DataType::Int32, + None, + values, + received_at, + ); } "sift.data.v2.Int64Values" => { let Int64Values { metadata, values, - extras: _, + extras, } = Int64Values::decode(channel_page.value) .context("failed to decode channel page")?; @@ -836,36 +923,28 @@ impl DataService { .units(channel.unit.as_ref().map(|u| u.name.as_str())) .build(); - let values = values - .into_iter() - .flat_map(|Int64Value { timestamp, value }| { - timestamp.map(|Timestamp { seconds, nanos }| { - ( - secs_and_subsec_nanos_to_unix_nanos(seconds, nanos), - ChannelValue::I64(value), - ) - }) - }) - .collect::>(); - - if let Some(channel_column) = columns.get_mut(&column_name) { - channel_column.values.extend(values); - } else { - columns.insert( - column_name, - ChannelColumn { - values, - json_metadata: None, - data_type: DataType::Int64, - }, - ); - } + let (values, received_at) = take_samples( + values + .into_iter() + .map(|Int64Value { timestamp, value }| (timestamp, value)), + &extras, + include_received_at, + ChannelValue::I64, + ); + store_column( + &mut columns, + column_name, + DataType::Int64, + None, + values, + received_at, + ); } "sift.data.v2.Uint32Values" => { let Uint32Values { metadata, values, - extras: _, + extras, } = Uint32Values::decode(channel_page.value) .context("failed to decode channel page")?; @@ -884,36 +963,28 @@ impl DataService { .units(channel.unit.as_ref().map(|u| u.name.as_str())) .build(); - let values = values - .into_iter() - .flat_map(|Uint32Value { timestamp, value }| { - timestamp.map(|Timestamp { seconds, nanos }| { - ( - secs_and_subsec_nanos_to_unix_nanos(seconds, nanos), - ChannelValue::U32(value), - ) - }) - }) - .collect::>(); - - if let Some(channel_column) = columns.get_mut(&column_name) { - channel_column.values.extend(values); - } else { - columns.insert( - column_name, - ChannelColumn { - values, - json_metadata: None, - data_type: DataType::UInt32, - }, - ); - } + let (values, received_at) = take_samples( + values + .into_iter() + .map(|Uint32Value { timestamp, value }| (timestamp, value)), + &extras, + include_received_at, + ChannelValue::U32, + ); + store_column( + &mut columns, + column_name, + DataType::UInt32, + None, + values, + received_at, + ); } "sift.data.v2.Uint64Values" => { let Uint64Values { metadata, values, - extras: _, + extras, } = Uint64Values::decode(channel_page.value) .context("failed to decode channel page")?; @@ -932,30 +1003,22 @@ impl DataService { .units(channel.unit.as_ref().map(|u| u.name.as_str())) .build(); - let values = values - .into_iter() - .flat_map(|Uint64Value { timestamp, value }| { - timestamp.map(|Timestamp { seconds, nanos }| { - ( - secs_and_subsec_nanos_to_unix_nanos(seconds, nanos), - ChannelValue::U64(value), - ) - }) - }) - .collect::>(); - - if let Some(channel_column) = columns.get_mut(&column_name) { - channel_column.values.extend(values); - } else { - columns.insert( - column_name, - ChannelColumn { - values, - json_metadata: None, - data_type: DataType::UInt64, - }, - ); - } + let (values, received_at) = take_samples( + values + .into_iter() + .map(|Uint64Value { timestamp, value }| (timestamp, value)), + &extras, + include_received_at, + ChannelValue::U64, + ); + store_column( + &mut columns, + column_name, + DataType::UInt64, + None, + values, + received_at, + ); } _ => bail!("queried an unsupported channel type"), } @@ -999,25 +1062,25 @@ impl DataService { let mut array_builders = Vec::>::new(); let mut time_col_builder = Int64Builder::new(); + let mut cursors = Vec::with_capacity(columns.len()); - for (column_name, column) in &columns { + for (column_name, column) in columns { let Some((_, first_val)) = column.values.first() else { bail!("unexpected empty column encountered") }; - - match &column.data_type { - DataType::Float32 => array_builders.push(Box::new(Float32Builder::new())), - DataType::Float64 => array_builders.push(Box::new(Float64Builder::new())), - DataType::Int32 => array_builders.push(Box::new(Int32Builder::new())), - DataType::Int64 => array_builders.push(Box::new(Int64Builder::new())), - DataType::UInt32 => array_builders.push(Box::new(UInt32Builder::new())), - DataType::UInt64 => array_builders.push(Box::new(UInt64Builder::new())), - DataType::Utf8 => array_builders.push(Box::new(StringBuilder::new())), - DataType::Boolean => array_builders.push(Box::new(BooleanBuilder::new())), - DataType::Binary => array_builders.push(Box::new(BinaryBuilder::new())), - _ => bail!("unsupported column data type: {:?}", column.data_type), + if let Some(received) = &column.received_at { + if received.len() != column.values.len() { + bail!( + "received timestamps ({}) do not align with samples ({}) for {column_name}", + received.len(), + column.values.len(), + ); + } } + let value_builder = array_builders.len(); + array_builders.push(new_column_builder(&column.data_type)?); + let mut field = Field::new(column_name.clone(), column.data_type.clone(), true); if let Some(json_metadata) = column.json_metadata.as_ref() { @@ -1039,15 +1102,32 @@ impl DataService { } fields.push(field); + + let received_builder = if column.received_at.is_some() { + let index = array_builders.len(); + array_builders.push(Box::new(Int64Builder::new())); + fields.push(Field::new( + column_name.with_received_at(), + DataType::Int64, + true, + )); + Some(index) + } else { + None + }; + + cursors.push(OutputCursor { + data_type: column.data_type, + samples: column.values.into_iter().peekable(), + received_at: column.received_at, + received_index: 0, + value_builder, + received_builder, + }); } let schema = Arc::new(Schema::new(fields)); - let mut col_iters = columns - .into_iter() - .map(|(_, col)| (col.data_type, col.values.into_iter().peekable())) - .collect::>(); - let mut rows_since_flush = 0; let mut size_since_flush = 0; @@ -1072,9 +1152,9 @@ impl DataService { } loop { - let maybe_current_ts = col_iters + let maybe_current_ts = cursors .iter_mut() - .filter_map(|(_, it)| it.peek().map(|(ts, _)| *ts)) + .filter_map(|cursor| cursor.samples.peek().map(|(ts, _)| *ts)) .min(); let Some(current_ts) = maybe_current_ts else { @@ -1083,22 +1163,36 @@ impl DataService { time_col_builder.append_value(current_ts); - for (i, (data_type, col_iter)) in col_iters.iter_mut().enumerate() { - let builder = &mut array_builders[i]; - - let Some((ts, val)) = col_iter.peek() else { - Self::append_null_to_builder(data_type, builder)?; - continue; - }; - - if *ts == current_ts { - Self::append_to_builder(val, builder)?; + for cursor in &mut cursors { + let data_type = cursor.data_type.clone(); + let value_builder = cursor.value_builder; + let received_builder = cursor.received_builder; + let peeked_ts = cursor.samples.peek().map(|(ts, _)| *ts); + + if peeked_ts == Some(current_ts) { + let (_, val) = cursor.samples.next().expect("peeked sample"); + let received = received_builder.map(|_| { + let value = cursor + .received_at + .as_ref() + .and_then(|series| series.get(cursor.received_index).copied()) + .flatten(); + cursor.received_index += 1; + value + }); size_since_flush += val.data_size(); - - // Advance the iterator - col_iter.next(); + if received.flatten().is_some() { + size_since_flush += mem::size_of::(); + } + Self::append_to_builder(&val, &mut array_builders[value_builder])?; + if let Some(index) = received_builder { + Self::append_i64_option(&mut array_builders[index], received.flatten())?; + } } else { - Self::append_null_to_builder(data_type, builder)?; + Self::append_null_to_builder(&data_type, &mut array_builders[value_builder])?; + if let Some(index) = received_builder { + Self::append_null_to_builder(&DataType::Int64, &mut array_builders[index])?; + } } } rows_since_flush += 1; @@ -1124,6 +1218,15 @@ impl DataService { }) } + fn append_i64_option(builder: &mut Box, value: Option) -> Result<()> { + builder + .as_any_mut() + .downcast_mut::() + .context("failed to downcast array builder to int64")? + .append_option(value); + Ok(()) + } + fn append_null_to_builder( data_type: &DataType, builder: &mut Box, diff --git a/rust/crates/sift_mcp/src/service/data/test.rs b/rust/crates/sift_mcp/src/service/data/test.rs index de10092e37..5231c7eb55 100644 --- a/rust/crates/sift_mcp/src/service/data/test.rs +++ b/rust/crates/sift_mcp/src/service/data/test.rs @@ -109,6 +109,80 @@ fn double_page(channel_id: &str, channel_name: &str, samples: Vec<(i64, f64)>) - } } +fn received_extra(received_nanos: &[Option]) -> sift_rs::data::v2::DimensionIndividualWrapper { + use sift_rs::data::v2::{ + DimensionIndividualWrapper, DimensionProtoTimestampValues, + dimension_individual_wrapper::{Type, ValueWrapper}, + dimension_proto_timestamp_values::DimensionProtoTimestampValue, + }; + + DimensionIndividualWrapper { + label: "sift_received_at".into(), + r#type: Type::AdditionalTimestampNanos as i32, + value_wrapper: Some(ValueWrapper::DimensionProtoTimestampValues( + DimensionProtoTimestampValues { + values: received_nanos + .iter() + .copied() + .map(|nanos| DimensionProtoTimestampValue { + value: nanos.map(|n| { + let (seconds, nanos) = unix_nanos_to_secs_and_subsec_nanos(n); + Timestamp { seconds, nanos } + }), + }) + .collect(), + }, + )), + } +} + +/// `samples` keeps samples with no generation timestamp so receipt times stay +/// aligned with the protobuf values array, which is how the API returns them. +fn double_page_with_received( + channel_id: &str, + channel_name: &str, + samples: Vec<(Option, f64)>, + received: &[Option], +) -> Any { + let values = samples + .into_iter() + .map(|(ts_nanos, value)| DoubleValue { + timestamp: ts_nanos.map(|n| { + let (seconds, nanos) = unix_nanos_to_secs_and_subsec_nanos(n); + Timestamp { seconds, nanos } + }), + value, + }) + .collect(); + + let payload = DoubleValues { + metadata: Some(Metadata { + channel: Some(metadata::Channel { + channel_id: channel_id.into(), + name: channel_name.into(), + ..Default::default() + }), + ..Default::default() + }), + values, + extras: vec![received_extra(received)], + }; + + Any { + type_url: "sift.data.v2.DoubleValues".into(), + value: Bytes::from(payload.encode_to_vec()), + } +} + +fn column_index(batch: &RecordBatch, fragment: &str) -> usize { + batch + .schema() + .fields() + .iter() + .position(|field| field.name().contains(fragment)) + .unwrap_or_else(|| panic!("missing column containing `{fragment}`")) +} + /// Like `double_page`, but sets `Metadata.sampled_ms` — the rate the service /// says it actually applied, which is what the caller is told about. fn double_page_sampled( @@ -165,7 +239,7 @@ async fn get_data_empty_inputs_errors() { let mut buffer = Vec::new(); let err = service - .get_data(&[], asset_range(0, 1_000_000_000), 0, &mut buffer) + .get_data(&[], asset_range(0, 1_000_000_000), 0, false, &mut buffer) .await .expect_err("expected error on empty channel_inputs"); @@ -195,6 +269,7 @@ async fn get_data_writes_double_channel_to_parquet() { &[raw_channel("c1")], asset_range(0, 3_000_000_000), 0, + false, &mut buffer, ) .await @@ -240,6 +315,7 @@ async fn get_data_merges_disjoint_timestamps_with_nulls() { &[raw_channel("c1"), raw_channel("c2")], asset_range(0, 4_000_000_000), 0, + false, &mut buffer, ) .await @@ -311,6 +387,7 @@ async fn get_data_paginates_until_token_empty() { &[raw_channel("c1")], asset_range(0, 3_000_000_000), 0, + false, &mut buffer, ) .await @@ -346,6 +423,7 @@ async fn get_data_requests_large_pages() { &[raw_channel("c1")], asset_range(0, 3_000_000_000), 0, + false, &mut buffer, ) .await @@ -365,6 +443,7 @@ async fn get_data_propagates_grpc_error() { &[raw_channel("c1")], asset_range(0, 1_000_000_000), 0, + false, &mut buffer, ) .await @@ -391,7 +470,7 @@ async fn get_data_run_without_start_time_errors() { let mut buffer = Vec::new(); let err = service - .get_data(&[raw_channel("c1")], time_range, 0, &mut buffer) + .get_data(&[raw_channel("c1")], time_range, 0, false, &mut buffer) .await .expect_err("expected error for run with no start_time"); @@ -421,6 +500,7 @@ async fn get_data_identifies_channels_that_returned_no_samples() { ], asset_range(0, 4_000_000_000), 0, + false, &mut buffer, ) .await @@ -467,6 +547,7 @@ async fn get_data_does_not_report_a_calculated_channel_that_returned_data() { }], asset_range(0, 4_000_000_000), 0, + false, &mut buffer, ) .await @@ -499,6 +580,7 @@ async fn get_data_reports_nothing_empty_when_every_channel_has_samples() { &[named_raw_channel("c1", "a"), named_raw_channel("c2", "b")], asset_range(0, 4_000_000_000), 0, + false, &mut buffer, ) .await @@ -526,6 +608,7 @@ async fn get_data_reports_empty_registration_ids_and_calculation_keys() { ], asset_range(0, 4_000_000_000), 0, + false, &mut Vec::new(), ) .await @@ -558,6 +641,7 @@ async fn get_data_no_samples_error_identifies_every_channel() { ], asset_range(0, 4_000_000_000), 0, + false, &mut buffer, ) .await @@ -587,7 +671,13 @@ async fn get_data_no_samples_error_caps_the_channel_list() { let (service, _h) = service_with_mock(mock).await; let mut buffer = Vec::new(); let err = service - .get_data(&inputs, asset_range(0, 4_000_000_000), 0, &mut buffer) + .get_data( + &inputs, + asset_range(0, 4_000_000_000), + 0, + false, + &mut buffer, + ) .await .expect_err("expected an error when nothing returned samples"); @@ -775,6 +865,7 @@ async fn get_data_sends_saved_calculation_query() { &[saved_calculation("thrust_margin", "$1 * 2", "c1")], asset_range(0, 3_000_000_000), 0, + false, &mut buffer, ) .await @@ -814,6 +905,7 @@ async fn get_data_reports_the_applied_sample_rate() { &[raw_channel("c1")], asset_range(0, 3_000_000_000), 100, + false, &mut buffer, ) .await @@ -848,6 +940,7 @@ async fn get_data_reports_raw_when_nothing_was_decimated() { &[raw_channel("c1")], asset_range(0, 3_000_000_000), 0, + false, &mut buffer, ) .await @@ -882,6 +975,7 @@ async fn get_data_flags_a_file_that_mixes_decimated_and_raw_channels() { &[raw_channel("c1"), raw_channel("c2")], asset_range(0, 3_000_000_000), 100, + false, &mut buffer, ) .await @@ -893,3 +987,252 @@ async fn get_data_flags_a_file_that_mixes_decimated_and_raw_channels() { assert!(output.applied_sample_ms.decimated()); assert!(output.applied_sample_ms.mixed()); } + +#[tokio::test] +async fn get_data_ignores_received_at_unless_asked() { + let mut mock = MockDataServiceImpl::new(); + mock.expect_get_data().times(1).returning(|req| { + assert_eq!(req.get_ref().include_received_at, None); + Ok(Response::new(GetDataResponse { + data: vec![double_page_with_received( + "c1", + "temp", + vec![(Some(1_000_000_000), 10.0)], + &[Some(1_500_000_000)], + )], + next_page_token: String::new(), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + let mut buffer = Vec::new(); + service + .get_data( + &[raw_channel("c1")], + asset_range(0, 2_000_000_000), + 0, + false, + &mut buffer, + ) + .await + .expect("get_data failed"); + + let batches = read_parquet(buffer); + let batch = &batches[0]; + assert_eq!(batch.num_columns(), 2); + assert!( + batch + .schema() + .fields() + .iter() + .all(|field| !field.name().contains("received_at")) + ); +} + +#[tokio::test] +async fn get_data_writes_received_at_beside_the_channel() { + let mut mock = MockDataServiceImpl::new(); + mock.expect_get_data().times(1).returning(|req| { + assert_eq!(req.get_ref().include_received_at, Some(true)); + Ok(Response::new(GetDataResponse { + data: vec![double_page_with_received( + "c1", + "temp", + vec![(Some(1_000_000_000), 10.0), (Some(2_000_000_000), 11.0)], + &[Some(1_500_000_000), None], + )], + next_page_token: String::new(), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + let mut buffer = Vec::new(); + service + .get_data( + &[named_raw_channel("c1", "temp")], + asset_range(0, 3_000_000_000), + 0, + true, + &mut buffer, + ) + .await + .expect("get_data failed"); + + let batches = read_parquet(buffer); + let batch = &batches[0]; + assert_eq!(batch.num_columns(), 3); + let received_idx = column_index(batch, "received_at=\"true\""); + let schema = batch.schema(); + let field = schema.field(received_idx); + assert_eq!(field.data_type(), &DataType::Int64); + assert!(field.is_nullable()); + assert_eq!( + field.name(), + "temp {channel_id=\"c1\", received_at=\"true\"}" + ); + + let received = batch.column(received_idx).as_primitive::(); + assert_eq!(received.value(0), 1_500_000_000); + assert!(received.is_null(1)); + + let value_idx = batch + .schema() + .fields() + .iter() + .position(|field| { + field.name().contains("channel_id=\"c1\"") && !field.name().contains("received_at") + }) + .expect("missing value column"); + let values = batch.column(value_idx).as_primitive::(); + assert_eq!(values.value(0), 10.0); + assert_eq!(values.value(1), 11.0); +} + +/// Receipt times are aligned with the protobuf values, including samples the +/// writer drops because they have no generation timestamp. Dropping one must +/// not shift the next sample's receipt time onto it. +#[tokio::test] +async fn get_data_keeps_received_at_aligned_when_a_timestamp_is_missing() { + let mut mock = MockDataServiceImpl::new(); + mock.expect_get_data().returning(|_| { + Ok(Response::new(GetDataResponse { + data: vec![double_page_with_received( + "c1", + "temp", + vec![ + (Some(1_000_000_000), 10.0), + (None, 11.0), + (Some(3_000_000_000), 12.0), + ], + &[Some(100), Some(200), Some(300)], + )], + next_page_token: String::new(), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + let mut buffer = Vec::new(); + service + .get_data( + &[raw_channel("c1")], + asset_range(0, 4_000_000_000), + 0, + true, + &mut buffer, + ) + .await + .expect("get_data failed"); + + let batches = read_parquet(buffer); + let batch = &batches[0]; + assert_eq!(batch.num_rows(), 2); + let timestamps = batch.column(0).as_primitive::(); + assert_eq!(timestamps.values(), &[1_000_000_000, 3_000_000_000]); + let received = batch + .column(column_index(batch, "received_at=\"true\"")) + .as_primitive::(); + assert_eq!(received.values(), &[100, 300]); +} + +#[tokio::test] +async fn get_data_nulls_received_at_where_the_channel_has_no_sample() { + let mut mock = MockDataServiceImpl::new(); + mock.expect_get_data().returning(|_| { + Ok(Response::new(GetDataResponse { + data: vec![ + double_page_with_received( + "c1", + "temp", + vec![(Some(1_000_000_000), 1.0), (Some(3_000_000_000), 3.0)], + &[Some(11), Some(33)], + ), + double_page("c2", "pressure", vec![(2_000_000_000, 20.0)]), + ], + next_page_token: String::new(), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + let mut buffer = Vec::new(); + service + .get_data( + &[raw_channel("c1"), raw_channel("c2")], + asset_range(0, 4_000_000_000), + 0, + true, + &mut buffer, + ) + .await + .expect("get_data failed"); + + let batches = read_parquet(buffer); + let batch = &batches[0]; + assert_eq!(batch.num_rows(), 3); + assert_eq!(batch.num_columns(), 4); + assert!( + batch + .schema() + .fields() + .iter() + .filter(|field| field.name().contains("received_at")) + .all(|field| field.name().contains("channel_id=\"c1\"")) + ); + + let received = batch + .column(column_index(batch, "received_at=\"true\"")) + .as_primitive::(); + assert_eq!(received.value(0), 11); + assert!(received.is_null(1)); + assert_eq!(received.value(2), 33); +} + +#[tokio::test] +async fn get_data_appends_received_at_across_pages_and_backfills_gaps() { + let mut mock = MockDataServiceImpl::new(); + mock.expect_get_data().returning(|req| { + let req = req.into_inner(); + assert_eq!(req.include_received_at, Some(true)); + let (pages, next) = match req.page_token.as_str() { + "" => ( + vec![double_page("c1", "temp", vec![(1_000_000_000, 1.0)])], + "page-2".to_string(), + ), + "page-2" => ( + vec![double_page_with_received( + "c1", + "temp", + vec![(Some(2_000_000_000), 2.0)], + &[Some(2_500_000_000)], + )], + String::new(), + ), + other => return Err(Status::invalid_argument(format!("bad token: {other}"))), + }; + Ok(Response::new(GetDataResponse { + data: pages, + next_page_token: next, + })) + }); + + let (service, _h) = service_with_mock(mock).await; + let mut buffer = Vec::new(); + service + .get_data( + &[raw_channel("c1")], + asset_range(0, 3_000_000_000), + 0, + true, + &mut buffer, + ) + .await + .expect("get_data failed"); + + let batches = read_parquet(buffer); + let batch = &batches[0]; + assert_eq!(batch.num_rows(), 2); + let received = batch + .column(column_index(batch, "received_at=\"true\"")) + .as_primitive::(); + assert!(received.is_null(0)); + assert_eq!(received.value(1), 2_500_000_000); +} diff --git a/rust/crates/sift_mcp/src/service/ingest/mod.rs b/rust/crates/sift_mcp/src/service/ingest/mod.rs index 290ad8cfc8..e88398c01f 100644 --- a/rust/crates/sift_mcp/src/service/ingest/mod.rs +++ b/rust/crates/sift_mcp/src/service/ingest/mod.rs @@ -113,17 +113,27 @@ impl IngestService { } let mut channel_configs = Vec::with_capacity(fields.len() - 1); - let mut column_encodings = Vec::with_capacity(fields.len() - 1); + // `None` marks a receipt-time column. It rides along in the file and is + // not a channel, so it is neither hashed into the flow nor ingested. + let mut column_roles = Vec::with_capacity(fields.len() - 1); let mut hasher = DefaultHasher::new(); // Skip the time column for field in fields.iter().skip(1) { + if ColumnName::try_from(field.name().as_str()).is_ok_and(|name| name.received_at()) { + column_roles.push(None); + continue; + } let (config, encoding) = Self::arrow_field_to_channel_config(field)?; let wf = config.encode_to_vec(); hasher.write(&wf); channel_configs.push(config); - column_encodings.push(encoding); + column_roles.push(Some(encoding)); + } + + if channel_configs.is_empty() { + bail!("parquet file must have at least one channel column") } let hash = format!( @@ -201,11 +211,11 @@ impl IngestService { let columns = record_batch.columns(); let data_columns = &columns[1..]; - if data_columns.len() != column_encodings.len() { + if data_columns.len() != column_roles.len() { bail!( "parquet record batch has {} data columns but channel config expected {}", data_columns.len(), - column_encodings.len(), + column_roles.len(), ); } @@ -221,7 +231,10 @@ impl IngestService { let mut channel_values = Vec::with_capacity(data_columns.len()); - for (col, encoding) in data_columns.iter().zip(column_encodings.iter()) { + for (col, role) in data_columns.iter().zip(column_roles.iter()) { + let Some(encoding) = role else { + continue; + }; let val = Self::channel_value_from_arrow_array(encoding, col, i)?; channel_values.push(val); } diff --git a/rust/crates/sift_mcp/src/tool/data/mod.rs b/rust/crates/sift_mcp/src/tool/data/mod.rs index 820fef2cd9..126561bcec 100644 --- a/rust/crates/sift_mcp/src/tool/data/mod.rs +++ b/rust/crates/sift_mcp/src/tool/data/mod.rs @@ -164,6 +164,10 @@ pub struct GetDataParams { channel_regex: Option, channel_id: Option, channel_ids: Option>, + /// When true, request `sift_received_at` and write it as a column when the + /// service returns it. `includeReceivedAt` is accepted as an alias. + #[serde(default, alias = "includeReceivedAt")] + include_received_at: bool, output: PathBuf, } @@ -233,6 +237,11 @@ impl SiftMcpServer { has no channel id, so its column carries the calculated channel's name in both places. - Enum and BitField channels carry their decode config in field metadata under the `enum_config` and `bit_field_elements` keys respectively. + - When `include_received_at` is true and the service returns receipt times, each such channel + gains a second Int64 column ` {channel_id=\"...\", received_at=\"true\", ...}` + of unix nanos. Cells are null where that sample has no receipt time, or where the channel + has no sample on the row. `timestamp_unix_nanos` stays the generation time. Enum and + bit-field channels have no receipt time. The column is omitted when the service returns none. - A requested channel that produced no samples has NO column at all, not an all-null one. The tool result reports these so they never have to be inferred from the schema: `unmatched_channel_names` lists requested names served by neither a raw channel nor a saved @@ -275,6 +284,11 @@ impl SiftMcpServer { without name-based resolution or calculated-channel fallback. - `channel_ids`: optional non-empty array of exact raw channel IDs on the specified asset. Selects only those registrations. Every ID must exist on the asset; duplicate IDs are queried once. + - `include_received_at`: optional bool, default false. `includeReceivedAt` is accepted as an alias. + Set it true to request `sift_received_at` and write receipt time beside each channel that has + it, so you can compare generation time (`timestamp_unix_nanos`) with the time Sift received + the measurement. Leave it false unless you need that comparison. Latency needs `sample_ms = 0`: + a decimated file only has receipt time for the samples LTTB kept. - `output`: filesystem path for the Parquet file. Replaced only when the call succeeds; a failed call leaves no partial file and keeps any existing one. The tool creates missing parent directories. @@ -340,6 +354,7 @@ impl SiftMcpServer { start_time_unix_nanos, end_time_unix_nanos, sample_ms, + include_received_at, output, }) = params; @@ -627,7 +642,13 @@ impl SiftMcpServer { let data_output = match self .data_service - .get_data(&channel_inputs, time_range, sample_ms, output_file.file()) + .get_data( + &channel_inputs, + time_range, + sample_ms, + include_received_at, + output_file.file(), + ) .await { Ok(data_output) => data_output, @@ -850,6 +871,9 @@ impl SiftMcpServer { - `channel_id` is REQUIRED inside the block; the bracketed attributes are optional. This is the canonical form produced by `get_data`. Enum and BitField channels are recognized via field metadata under the `enum_config` and `bit_field_elements` keys respectively. + - A column with `received_at=\"true\"` is a receipt timestamp from `get_data`, not a channel. + Those columns are skipped and are not created in Sift. The file still needs at least one + channel column. Output: - `{ \"input\": \"\", \"asset_name\": \"...\", \"asset_id\": \"...\", \"asset_url\": string|null, diff --git a/rust/crates/sift_mcp/src/tool/data/test.rs b/rust/crates/sift_mcp/src/tool/data/test.rs index 97af67512b..91d23723a7 100644 --- a/rust/crates/sift_mcp/src/tool/data/test.rs +++ b/rust/crates/sift_mcp/src/tool/data/test.rs @@ -106,10 +106,36 @@ fn get_data_params(channel_regex: &str) -> Parameters { channel_regex: Some(channel_regex.into()), channel_id: None, channel_ids: None, + include_received_at: false, output: std::env::temp_dir().join("sift-mcp-get-data-test-never-written.parquet"), }) } +#[test] +fn get_data_params_accept_include_received_at_aliases() { + let base = serde_json::json!({ + "asset_name": "bench", + "start_time_unix_nanos": 0, + "end_time_unix_nanos": 1, + "sample_ms": 0, + "channel_names": ["temp"], + "output": "out.parquet", + }); + + let omitted: GetDataParams = serde_json::from_value(base.clone()).unwrap(); + assert!(!omitted.include_received_at); + + let mut snake = base.clone(); + snake["include_received_at"] = serde_json::json!(true); + let snake: GetDataParams = serde_json::from_value(snake).unwrap(); + assert!(snake.include_received_at); + + let mut camel = base; + camel["includeReceivedAt"] = serde_json::json!(true); + let camel: GetDataParams = serde_json::from_value(camel).unwrap(); + assert!(camel.include_received_at); +} + #[tokio::test] async fn get_data_fetches_exact_channel_ids() { for selector in [ @@ -611,6 +637,7 @@ fn named_params( channel_regex: None, channel_id: None, channel_ids: None, + include_received_at: false, output, }) } @@ -1189,6 +1216,7 @@ async fn get_data_reports_channel_names_that_matched_nothing() { channel_regex: None, channel_id: None, channel_ids: None, + include_received_at: false, output: dir.path().join("out.parquet"), })) .await @@ -1254,6 +1282,7 @@ async fn get_data_reports_no_unmatched_names_for_a_regex_selection() { channel_regex: Some("press.*".into()), channel_id: None, channel_ids: None, + include_received_at: false, output: dir.path().join("out.parquet"), })) .await @@ -1317,6 +1346,7 @@ async fn get_data_reports_matched_channels_that_returned_no_samples() { channel_regex: None, channel_id: None, channel_ids: None, + include_received_at: false, output: dir.path().join("out.parquet"), })) .await @@ -1385,6 +1415,7 @@ async fn no_data_error_reports_both_the_empty_and_the_unmatched_channels() { channel_regex: None, channel_id: None, channel_ids: None, + include_received_at: false, output: dir.path().join("out.parquet"), })) .await @@ -1568,6 +1599,7 @@ async fn get_data_persist_failure_keeps_the_channel_report() { channel_regex: None, channel_id: None, channel_ids: None, + include_received_at: false, output: output.clone(), })) .await @@ -1624,6 +1656,7 @@ async fn get_data_leaves_no_file_when_the_data_call_fails() { channel_regex: None, channel_id: None, channel_ids: None, + include_received_at: false, output: dir.path().join("out.parquet"), })) .await @@ -1681,6 +1714,7 @@ async fn get_data_leaves_no_file_when_the_request_is_cancelled() { channel_regex: None, channel_id: None, channel_ids: None, + include_received_at: false, output: output.clone(), }); let request = tokio::spawn(async move { server.get_data(params).await }); @@ -1771,6 +1805,7 @@ async fn get_data_writes_into_a_directory_that_does_not_exist_yet() { channel_regex: None, channel_id: None, channel_ids: None, + include_received_at: false, output: nested.clone(), })) .await