micromegas_analytics/lakehouse/
block_object_decoder.rs

1//! Format → decoder registry for `parse_block`: turns a block payload of any
2//! registered `streams.format` into generic `(type_name, jsonb)` objects, one
3//! per row, mirroring the existing `BlockProcessorMap` pattern
4//! (`block_partition_spec.rs`).
5
6use crate::lakehouse::otel::block_decoders::{
7    OtelLogsBlockDecoder, OtelMetricsBlockDecoder, OtelTracesBlockDecoder,
8};
9use crate::lakehouse::parse_block_table_function::transit_value_to_jsonb;
10use crate::metadata::StreamMetadata;
11use crate::payload::parse_block;
12use anyhow::Result;
13use micromegas_ingestion::web_ingestion_service::{
14    FORMAT_OTLP_LOGS, FORMAT_OTLP_METRICS, FORMAT_OTLP_TRACES, FORMAT_TRANSIT,
15};
16use micromegas_telemetry::block_wire_format::BlockPayload;
17use micromegas_tracing::prelude::*;
18use micromegas_transit::value::Value as TransitValue;
19use std::collections::HashMap;
20use std::fmt::Debug;
21use std::sync::Arc;
22
23/// Receives objects decoded from a block payload, in payload order.
24pub trait ObjectVisitor {
25    /// `value` is the JSONB encoding of one object. Returns false to stop iteration.
26    fn visit(&mut self, type_name: &str, value: &[u8]) -> Result<bool>;
27
28    /// Consumes an ordinal without emitting a row (a payload entry the decoder
29    /// recognizes but cannot represent). Keeps `object_index` aligned with the
30    /// block's true ordinals. Returns false to stop iteration.
31    fn skip(&mut self) -> Result<bool>;
32}
33
34/// Decodes one block payload of a given `streams.format` into generic objects.
35pub trait BlockObjectDecoder: Send + Sync + Debug {
36    fn decode(
37        &self,
38        stream: &StreamMetadata,
39        payload: &BlockPayload,
40        visitor: &mut dyn ObjectVisitor,
41    ) -> Result<()>;
42}
43
44/// Map from `streams.format` to the decoder that turns that wire format into
45/// generic `(type_name, jsonb)` objects for `parse_block`.
46pub type BlockObjectDecoderMap = HashMap<&'static str, Arc<dyn BlockObjectDecoder>>;
47
48/// Decodes `micromegas-transit` block payloads: the native CBOR/transit wire
49/// format used by every in-tree SDK. A straight lift of the per-object callback
50/// `parse_block_table_function.rs` used to drive directly.
51#[derive(Debug)]
52pub struct TransitBlockDecoder;
53
54impl BlockObjectDecoder for TransitBlockDecoder {
55    fn decode(
56        &self,
57        stream: &StreamMetadata,
58        payload: &BlockPayload,
59        visitor: &mut dyn ObjectVisitor,
60    ) -> Result<()> {
61        parse_block(stream, payload, |value| {
62            if let TransitValue::Object(obj) = value {
63                let jsonb_val = transit_value_to_jsonb(value);
64                let mut buf = Vec::new();
65                jsonb_val.write_to_vec(&mut buf);
66                visitor.visit(obj.type_name, &buf)
67            } else {
68                // Defensive branch: no in-tree writer emits a non-Object top-level
69                // value (see Testing Strategy in the design plan). `skip()` keeps
70                // `object_index` aligned with the block's true ordinals.
71                warn!("parse_block: skipping non-Object value in transit block");
72                visitor.skip()
73            }
74        })?;
75        Ok(())
76    }
77}
78
79/// Registry covering every `streams.format` shipped in-tree.
80pub fn default_block_object_decoders() -> Arc<BlockObjectDecoderMap> {
81    let mut m: BlockObjectDecoderMap = HashMap::new();
82    m.insert(
83        FORMAT_TRANSIT,
84        Arc::new(TransitBlockDecoder) as Arc<dyn BlockObjectDecoder>,
85    );
86    m.insert(
87        FORMAT_OTLP_LOGS,
88        Arc::new(OtelLogsBlockDecoder) as Arc<dyn BlockObjectDecoder>,
89    );
90    m.insert(
91        FORMAT_OTLP_METRICS,
92        Arc::new(OtelMetricsBlockDecoder) as Arc<dyn BlockObjectDecoder>,
93    );
94    m.insert(
95        FORMAT_OTLP_TRACES,
96        Arc::new(OtelTracesBlockDecoder) as Arc<dyn BlockObjectDecoder>,
97    );
98    Arc::new(m)
99}
micromegas_analytics/lakehouse/block_object_decoder.rs

micromegas_analytics/lakehouse/
block_object_decoder.rs
