wp-core-connectors 0.7.0

Core connector registry and sink runtimes for WarpParse
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

//! Shared Arrow decode helpers for batch sources.
//!
//! Both `TcpBatchSource` and `FileBatchSource` decode their payloads into
//! Arrow `RecordBatch`es. [`WireFormat`] and the core decode logic are
//! re-exported from `wp_connector_utils::arrow`.

pub use wp_connector_utils::arrow::WireFormat;

use arrow::record_batch::RecordBatch;
use wf_connector_api::{SourceReason, SourceResult};
use wp_connector_api::SourceBatch;

use super::payload::payload_to_bytes;

/// Decode raw Arrow IPC Stream bytes from `SourceEvent`s into `RecordBatch`es.
///
/// Each event's payload is decoded independently via
/// [`wp_connector_utils::arrow::decode_arrow_ipc_batches`]; results are
/// concatenated.
pub fn decode_arrow_ipc_batches(events: &SourceBatch) -> SourceResult<Vec<RecordBatch>> {
    let mut batches = Vec::new();
    for event in events {
        let payload = payload_to_bytes(&event.payload);
        let decoded = wp_connector_utils::arrow::decode_arrow_ipc_batches(&payload)
            .map_err(|e| SourceReason::Decode.err_detail(e))?;
        batches.extend(decoded);
    }
    Ok(batches)
}

/// Decode wp_arrow frames into `RecordBatch`es.
///
/// Each event's payload is decoded independently via
/// [`wp_connector_utils::arrow::decode_arrow_framed_batches`]; results are
/// concatenated.
pub fn decode_arrow_framed_batches(events: &SourceBatch) -> SourceResult<Vec<RecordBatch>> {
    let mut batches = Vec::new();
    for event in events {
        let payload = payload_to_bytes(&event.payload);
        let decoded = wp_connector_utils::arrow::decode_arrow_framed_batches(&payload)
            .map_err(|e| SourceReason::Decode.err_detail(e))?;
        batches.extend(decoded);
    }
    Ok(batches)
}