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

//! Arrow IPC serialization helpers for sink `protocol: arrow`.
//!
//! - `schema` — Schema inference from `DataRecord` fields
//! - `batch`  — `DataRecord` → `RecordBatch` conversion with typed column builders

pub use wp_connector_utils::arrow::{
    data_record_to_batch, data_records_to_batch, infer_arrow_schema, infer_schema_from_record,
};

use arrow::record_batch::RecordBatch;
use orion_error::conversion::ToStructError;
use wp_connector_api::{SinkReason, SinkResult};

// ---------------------------------------------------------------------------
// Shared error helper
// ---------------------------------------------------------------------------

pub(crate) fn sink_err<E>(msg: &'static str, err: E) -> wp_connector_api::SinkError
where
    E: std::fmt::Display,
{
    SinkReason::Sink
        .to_err()
        .with_detail(format!("{msg}: {err}"))
}

// ---------------------------------------------------------------------------
// Shared Arrow IPC encoding helpers — delegate to wp_connector_utils
// ---------------------------------------------------------------------------

/// Encode a single `RecordBatch` as Arrow IPC Stream bytes.
pub(crate) fn encode_batch_ipc_stream(batch: &RecordBatch) -> SinkResult<Vec<u8>> {
    wp_connector_utils::arrow::encode_batch_ipc_stream(batch)
}

/// Encode a `RecordBatch` as a wp_arrow IPC frame: `[4B tag_len BE][tag][Arrow IPC stream]`.
pub(crate) fn encode_ipc_frame(tag: &str, batch: &RecordBatch) -> SinkResult<Vec<u8>> {
    wp_connector_utils::arrow::encode_ipc_frame(tag, batch)
}

/// Encode multiple `RecordBatch`es as a single wp_arrow frame.
pub(crate) fn encode_ipc_frame_multi(tag: &str, batches: &[RecordBatch]) -> SinkResult<Vec<u8>> {
    wp_connector_utils::arrow::encode_ipc_frame_multi(tag, batches)
}