feat(parquet): rfc 0022 green pt1 — promoted attribute columns in the writer

crates/ourios-parquet/src/lib.rs

@@ -32,6 +32,7 @@ pub mod audit_writer;
 pub mod compaction;
 pub mod manifest;
 pub mod partition;
+pub mod promoted;
 pub mod reader;
 pub mod record_batch;
 pub mod store;
@@ -50,12 +51,13 @@ pub use partition::{
     PartitionKey, TimestampOverflowError, effective_time_unix_nano, hour_partition_in_window,
     percent_decode_tenant, percent_encode_tenant,
 };
+pub use promoted::{PromotedAttributes, SERVICE_NAME_KEY};
 pub use reader::{Reader, ReaderError, ShapeValidation, batch_to_mined_records};
-pub use record_batch::{BatchError, mined_records_to_batch};
+pub use record_batch::{BatchError, mined_records_to_batch, mined_records_to_batch_with_promoted};
 pub use store::{S3Config, Store, StoreConfig, StoreError};
 pub use writer::{
     DEFAULT_ZSTD_LEVEL, ROW_GROUP_FLUSH_BYTES, Writer, WriterError, WrittenFile,
-    encode_records_to_parquet,
+    encode_records_to_parquet, encode_records_to_parquet_with_promoted,
 };
 
 use std::sync::Arc;
@@ -208,6 +210,21 @@ pub fn data_schema() -> SchemaRef {
     ]))
 }
 
+/// [`data_schema`] plus the RFC 0022 promoted attribute columns for
+/// `promoted` — additive `OPTIONAL` Utf8 fields appended after the
+/// §3.2 base columns (`resource.<key>` columns first,
+/// `resource.service.name` always leading, then `attr.<key>`). This
+/// is the writer's declared schema; [`data_schema`] stays the base
+/// shape readers address by name (§3.9 tolerates both absent and
+/// unknown columns, so files written under any promoted set coexist).
+#[must_use]
+pub fn data_schema_with_promoted(promoted: &PromotedAttributes) -> SchemaRef {
+    let base = data_schema();
+    let mut fields: Vec<Field> = base.fields().iter().map(|f| f.as_ref().clone()).collect();
+    fields.extend(promoted.fields());
+    Arc::new(ArrowSchema::new(fields))
+}
+
 /// Build the audit-event file Arrow schema per RFC 0005 §3.7.
 ///
 /// Both `event_kind` (`UInt8` Arrow — Parquet stores it physically

crates/ourios-parquet/src/promoted.rs

@@ -0,0 +1,193 @@
+//! RFC 0022 — promoted attribute columns.
+//!
+//! A promoted key is projected at write time from its canonical-JSON
+//! attribute column into a dedicated `OPTIONAL` Utf8 column named
+//! literally after the DSL path (`resource.<key>` / `attr.<key>`), so
+//! attribute predicates can prune row groups instead of scanning JSON
+//! (RFC 0022 §3.1). The JSON columns remain the source of truth: a
+//! promoted cell is a query-only projection, never read back into a
+//! [`MinedRecord`](ourios_core::record::MinedRecord).
+
+use arrow_schema::{DataType, Field};
+use ourios_core::otlp::{KeyValue, any_value};
+
+/// The resource key that is always promoted (RFC 0022 §3.1): the
+/// `Required`, `Stable` identity attribute of the `OTel` `service`
+/// resource entity, surfaced in the DSL as the bare `service` field.
+pub const SERVICE_NAME_KEY: &str = "service.name";
+
+/// Column-name prefix for promoted resource-attribute keys.
+const RESOURCE_PREFIX: &str = "resource.";
+/// Column-name prefix for promoted log-attribute keys.
+const ATTR_PREFIX: &str = "attr.";
+
+/// The effective promoted attribute key set (RFC 0022 §3.1/§3.2).
+///
+/// `service.name` is implicit and non-removable: `resource_keys()`
+/// always yields it first, regardless of the configured set. The
+/// configured keys come from `storage.promoted_attributes` (an RFC
+/// 0020 schema extension) and are deduplicated preserving first
+/// occurrence.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct PromotedAttributes {
+    resource: Vec<String>,
+    log: Vec<String>,
+}
+
+impl Default for PromotedAttributes {
+    /// The empty configured set — `service.name` only.
+    fn default() -> Self {
+        Self::new(std::iter::empty::<String>(), std::iter::empty::<String>())
+    }
+}
+
+impl PromotedAttributes {
+    /// Build the effective set from the configured resource and log
+    /// keys. The implicit `service.name` is prepended to the resource
+    /// keys; duplicates (including a configured `service.name`)
+    /// collapse, preserving first occurrence.
+    pub fn new(
+        resource: impl IntoIterator<Item = String>,
+        log: impl IntoIterator<Item = String>,
+    ) -> Self {
+        fn dedup_preserving_order(
+            implicit: impl IntoIterator<Item = String>,
+            keys: impl IntoIterator<Item = String>,
+        ) -> Vec<String> {
+            let mut seen = std::collections::HashSet::new();
+            implicit
+                .into_iter()
+                .chain(keys)
+                .filter(|k| seen.insert(k.clone()))
+                .collect()
+        }
+        Self {
+            resource: dedup_preserving_order([SERVICE_NAME_KEY.to_string()], resource),
+            log: dedup_preserving_order([], log),
+        }
+    }
+
+    /// The promoted resource-attribute keys, `service.name` first.
+    #[must_use]
+    pub fn resource_keys(&self) -> &[String] {
+        &self.resource
+    }
+
+    /// The promoted log-attribute keys (configured only).
+    #[must_use]
+    pub fn log_keys(&self) -> &[String] {
+        &self.log
+    }
+
+    /// The promoted column names in schema order: `resource.<key>`
+    /// columns first (`resource.service.name` leading), then
+    /// `attr.<key>` columns.
+    pub fn column_names(&self) -> impl Iterator<Item = String> + '_ {
+        self.resource
+            .iter()
+            .map(|k| format!("{RESOURCE_PREFIX}{k}"))
+            .chain(self.log.iter().map(|k| format!("{ATTR_PREFIX}{k}")))
+    }
+
+    /// The promoted columns as Arrow fields (RFC 0022 §3.1: `OPTIONAL`
+    /// Utf8 — Parquet `STRING` logical type over `BYTE_ARRAY`), in
+    /// [`Self::column_names`] order.
+    #[must_use]
+    pub fn fields(&self) -> Vec<Field> {
+        self.column_names()
+            .map(|name| Field::new(name, DataType::Utf8, true))
+            .collect()
+    }
+}
+
+/// Project one promoted key out of an attribute list (RFC 0022 §3.1):
+/// the value **iff** the key is present with a string `AnyValue`;
+/// `None` (a `NULL` cell) when the key is absent or its value is any
+/// other `AnyValue` variant. First occurrence wins, mirroring the
+/// first-match semantics of the query-side JSON `LIKE` arm.
+#[must_use]
+pub fn project_string_value<'a>(attrs: &'a [KeyValue], key: &str) -> Option<&'a str> {
+    attrs.iter().find(|kv| kv.key == key).and_then(string_value)
+}
+
+/// The §3.1 value projection of a single attribute: the payload **iff**
+/// the `AnyValue` is a string, `None` for any other variant (or none).
+#[must_use]
+pub fn string_value(kv: &KeyValue) -> Option<&str> {
+    match kv.value.as_ref().and_then(|v| v.value.as_ref()) {
+        Some(any_value::Value::StringValue(s)) => Some(s.as_str()),
+        _ => None,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use ourios_core::otlp::AnyValue;
+
+    use super::*;
+
+    fn kv_str(key: &str, value: &str) -> KeyValue {
+        KeyValue {
+            key: key.to_string(),
+            value: Some(AnyValue {
+                value: Some(any_value::Value::StringValue(value.to_string())),
+            }),
+            ..Default::default()
+        }
+    }
+
+    fn kv_int(key: &str, value: i64) -> KeyValue {
+        KeyValue {
+            key: key.to_string(),
+            value: Some(AnyValue {
+                value: Some(any_value::Value::IntValue(value)),
+            }),
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn service_name_is_implicit_first_and_deduplicated() {
+        let p = PromotedAttributes::new(
+            ["service.name".to_string(), "k8s.namespace.name".to_string()],
+            ["http.route".to_string(), "http.route".to_string()],
+        );
+        assert_eq!(p.resource_keys(), ["service.name", "k8s.namespace.name"]);
+        assert_eq!(p.log_keys(), ["http.route"]);
+        assert_eq!(
+            p.column_names().collect::<Vec<_>>(),
+            [
+                "resource.service.name",
+                "resource.k8s.namespace.name",
+                "attr.http.route"
+            ]
+        );
+    }
+
+    #[test]
+    fn default_set_is_service_name_only() {
+        let p = PromotedAttributes::default();
+        assert_eq!(p.resource_keys(), [SERVICE_NAME_KEY]);
+        assert!(p.log_keys().is_empty());
+    }
+
+    #[test]
+    fn fields_are_optional_utf8() {
+        for f in PromotedAttributes::default().fields() {
+            assert_eq!(*f.data_type(), DataType::Utf8);
+            assert!(f.is_nullable());
+        }
+    }
+
+    #[test]
+    fn projection_is_string_only_first_match() {
+        let attrs = [
+            kv_int("http.status_code", 500),
+            kv_str("service.name", "api"),
+            kv_str("service.name", "shadowed"),
+        ];
+        assert_eq!(project_string_value(&attrs, "service.name"), Some("api"));
+        assert_eq!(project_string_value(&attrs, "http.status_code"), None);
+        assert_eq!(project_string_value(&attrs, "absent"), None);
+    }
+}

crates/ourios-parquet/src/record_batch.rs

@@ -22,6 +22,7 @@
 //! the body into the canonical bytes (`MinedRecord.body` carries
 //! the bytes verbatim), so the writer just appends them.
 
+use std::collections::HashMap;
 use std::fmt;
 use std::sync::Arc;
 
@@ -37,7 +38,8 @@ use ourios_core::otlp::KeyValue;
 use ourios_core::record::{BodyKind, MinedRecord};
 
 use crate::partition::{TimestampOverflowError, effective_time_unix_nano};
-use crate::{columns, data_schema};
+use crate::promoted::{self, PromotedAttributes};
+use crate::{columns, data_schema, data_schema_with_promoted};
 
 /// Build an Arrow `RecordBatch` matching `data_schema()` from a
 /// slice of [`MinedRecord`]s.
@@ -60,6 +62,82 @@ pub fn mined_records_to_batch(records: &[MinedRecord]) -> Result<RecordBatch, Ba
     RecordBatch::try_new(data_schema(), arrays).map_err(BatchError::Arrow)
 }
 
+/// [`mined_records_to_batch`] plus the RFC 0022 promoted attribute
+/// columns for `promoted`, appended in [`PromotedAttributes`] column
+/// order to match [`crate::data_schema_with_promoted`]. Each promoted
+/// cell is the §3.1 string projection out of the record's
+/// resource/log attribute list (`NULL` for absent or non-string
+/// values); the canonical-JSON columns are built exactly as in the
+/// base path and remain the source of truth.
+///
+/// # Errors
+///
+/// See [`mined_records_to_batch`].
+pub fn mined_records_to_batch_with_promoted(
+    records: &[MinedRecord],
+    promoted: &PromotedAttributes,
+) -> Result<RecordBatch, BatchError> {
+    let mut b = Builders::with_capacity(records.len());
+    for r in records {
+        b.append(r)?;
+    }
+    let mut arrays = b.finish();
+    arrays.extend(project_promoted_columns(
+        records,
+        promoted.resource_keys(),
+        |r| r.resource_attributes.as_slice(),
+    ));
+    arrays.extend(project_promoted_columns(
+        records,
+        promoted.log_keys(),
+        |r| r.attributes.as_slice(),
+    ));
+    RecordBatch::try_new(data_schema_with_promoted(promoted), arrays).map_err(BatchError::Arrow)
+}
+
+/// Materialise the promoted columns of one attribute-list family
+/// (resource or log) in key order, visiting each record's attribute
+/// list once rather than once per key. The first occurrence of a
+/// promoted key decides its cell — [`promoted::string_value`] or a
+/// `NULL` for a non-string — matching
+/// [`promoted::project_string_value`] per column.
+fn project_promoted_columns(
+    records: &[MinedRecord],
+    keys: &[String],
+    attrs_of: impl Fn(&MinedRecord) -> &[KeyValue],
+) -> Vec<ArrayRef> {
+    if keys.is_empty() {
+        return Vec::new();
+    }
+    let index: HashMap<&str, usize> = keys
+        .iter()
+        .enumerate()
+        .map(|(i, k)| (k.as_str(), i))
+        .collect();
+    let mut builders: Vec<StringBuilder> = keys
+        .iter()
+        .map(|_| StringBuilder::with_capacity(records.len(), 0))
+        .collect();
+    let mut cells: Vec<Option<Option<&str>>> = vec![None; keys.len()];
+    for r in records {
+        cells.fill(None);
+        for kv in attrs_of(r) {
+            if let Some(&i) = index.get(kv.key.as_str())
+                && cells[i].is_none()
+            {
+                cells[i] = Some(promoted::string_value(kv));
+            }
+        }
+        for (b, cell) in builders.iter_mut().zip(&cells) {
+            append_option_str(b, cell.flatten());
+        }
+    }
+    builders
+        .into_iter()
+        .map(|mut b| Arc::new(b.finish()) as ArrayRef)
+        .collect()
+}
+
 /// Errors produced by [`mined_records_to_batch`].
 #[derive(Debug)]
 pub enum BatchError {