pub fn render_decode_delimited<G>(
    stream: &Stream<G, SourceOutput<Option<Vec<u8>>, Option<Vec<u8>>>>,
    key_encoding: Option<DataEncoding>,
    value_encoding: DataEncoding,
    debug_name: &str,
    envelope: &SourceEnvelope,
    metadata_items: Vec<IncludedColumnSource>,
    // Information about optional transformations that can be eagerly done.
    // If the decoding elects to perform them, it should replace this with
    // `None`.
    operators: &mut Option<LinearOperator>,
    fast_forwarded: bool,
    metrics: Metrics,
) -> (Stream<G, DecodeResult>, Option<Box<dyn Any>>)
where
    G: Scope,
{
    let op_name = format!(
        "{}{}DecodeDelimited",
        key_encoding
            .as_ref()
            .map(|key_encoding| key_encoding.op_name())
            .unwrap_or(""),
        value_encoding.op_name()
    );
    let mut key_decoder = key_encoding.map(|key_encoding| {
        get_decoder(
            key_encoding,
            debug_name,
            &SourceEnvelope::None(KeyEnvelope::None),
            operators,
            false,
            true,
            metrics.clone(),
        )
    });

    let mut value_decoder = get_decoder(
        value_encoding,
        debug_name,
        envelope,
        operators,
        fast_forwarded,
        true,
        metrics,
    );

    // The Debezium deduplication and upsert logic rely on elements for the same key going to the same worker.
    // Other decoders don't care; so we distribute things round-robin (i.e., by "position"), and
    // fall back to arbitrarily hashing by value if the upstream didn't give us a position.
    let use_key_contract = matches!(envelope, SourceEnvelope::Debezium(..));
    let results = stream.unary_frontier(
        Exchange::new(move |x: &SourceOutput<Option<Vec<u8>>, Option<Vec<u8>>>| {
            if use_key_contract {
                x.key.hashed()
            } else if let Some(position) = x.position {
                position.hashed()
            } else {
                x.value.hashed()
            }
        }),
        &op_name,
        move |_, _| {
            move |input, output| {
                let mut n_errors = 0;
                let mut n_successes = 0;
                input.for_each(|cap, data| {
                    let mut session = output.session(&cap);
                    for SourceOutput {
                        key,
                        value,
                        position,
                        upstream_time_millis: upstream_time,
                        partition,
                    } in data.iter()
                    {
                        let key = key_decoder
                            .as_mut()
                            .and_then(|decoder| try_decode(decoder, key.as_ref(), *upstream_time));

                        let value = try_decode(&mut value_decoder, value.as_ref(), *upstream_time);

                        if matches!(&key, Some(Err(_))) || matches!(&value, Some(Err(_))) {
                            n_errors += 1;
                        } else if matches!(&value, Some(Ok(_))) {
                            n_successes += 1;
                        }

                        session.give(DecodeResult {
                            key,
                            value,
                            position: *position,
                            metadata: to_metadata_row(
                                &metadata_items,
                                partition.clone(),
                                *position,
                                *upstream_time,
                            ),
                        });
                    }
                });
                // Matching historical practice, we only log metrics on the value decoder.
                if n_errors > 0 {
                    value_decoder.log_errors(n_errors);
                }
                if n_successes > 0 {
                    value_decoder.log_successes(n_successes);
                }
            }
        },
    );
    (results, None)
}
/// Decode already delimited records of data.
///
/// Precondition: each record in the stream has at most one key and at most one value.
/// This function is useful mainly for decoding data from systems like Kafka,
/// that have already separated the stream into records/messages/etc. before we
/// decode them.
///
/// Because we expect the upstream connector to have already delimited the data,
/// we return an error here if the decoder does not consume all the bytes. This
/// often lets us, for example, detect when Avro decoding has gone off the rails
/// (which is not always possible otherwise, since often gibberish strings can be interpreted as Avro,
///  so the only signal is how many bytes you managed to decode).
Decode already delimited records of data.
Precondition: each record in the stream has at most one key and at most one value.
This function is useful mainly for decoding data from systems like Kafka,
that have already separated the stream into records/messages/etc. before we
decode them.

Because we expect the upstream connector to have already delimited the data,
we return an error here if the decoder does not consume all the bytes. This
often lets us, for example, detect when Avro decoding has gone off the rails
(which is not always possible otherwise, since often gibberish strings can be interpreted as Avro,
so the only signal is how many bytes you managed to decode).
Decode already delimited records of data . Precondition : each record in the stream has at most one key and at most one value . This function is useful mainly for decoding data from systems like Kafka that have already separated the stream into records / messages / etc . before we decode them . Because we expect the upstream connector to have already delimited the data we return an error here if the decoder does not consume all the bytes . This often lets us for example detect when Avro decoding has gone off the rails ( which is not always possible otherwise since often gibberish strings can be interpreted as Avro so the only signal is how many bytes you managed to decode ) .
pub fn render_decode_delimited<G>(
    stream: &Stream<G, SourceOutput<Option<Vec<u8>>, Option<Vec<u8>>>>,
    key_encoding: Option<DataEncoding>,
    value_encoding: DataEncoding,
    debug_name: &str,
    envelope: &SourceEnvelope,
    metadata_items: Vec<IncludedColumnSource>,
    // Information about optional transformations that can be eagerly done.
    // If the decoding elects to perform them, it should replace this with
    // `None`.
    operators: &mut Option<LinearOperator>,
    fast_forwarded: bool,
    metrics: Metrics,
) -> (Stream<G, DecodeResult>, Option<Box<dyn Any>>)
where
    G: Scope,
{
    let op_name = format!(
        "{}{}DecodeDelimited",
        key_encoding
            .as_ref()
            .map(|key_encoding| key_encoding.op_name())
            .unwrap_or(""),
        value_encoding.op_name()
    );
    let mut key_decoder = key_encoding.map(|key_encoding| {
        get_decoder(
            key_encoding,
            debug_name,
            &SourceEnvelope::None(KeyEnvelope::None),
            operators,
            false,
            true,
            metrics.clone(),
        )
    });
    let mut value_decoder = get_decoder(
        value_encoding,
        debug_name,
        envelope,
        operators,
        fast_forwarded,
        true,
        metrics,
    );
    // The Debezium deduplication and upsert logic rely on elements for the same key going to the same worker.
    // Other decoders don't care; so we distribute things round-robin (i.e., by "position"), and
    // fall back to arbitrarily hashing by value if the upstream didn't give us a position.
    let use_key_contract = matches!(envelope, SourceEnvelope::Debezium(..));
    let results = stream.unary_frontier(
        Exchange::new(move |x: &SourceOutput<Option<Vec<u8>>, Option<Vec<u8>>>| {
            if use_key_contract {
                x.key.hashed()
            } else if let Some(position) = x.position {
                position.hashed()
            } else {
                x.value.hashed()
            }
        }),
        &op_name,
        move |_, _| {
            move |input, output| {
                let mut n_errors = 0;
                let mut n_successes = 0;
                input.for_each(|cap, data| {
                    let mut session = output.session(&cap);
                    for SourceOutput {
                        key,
                        value,
                        position,
                        upstream_time_millis: upstream_time,
                        partition,
                    } in data.iter()
                    {
                        let key = key_decoder
                            .as_mut()
                            .and_then(|decoder| try_decode(decoder, key.as_ref(), *upstream_time));
                        let value = try_decode(&mut value_decoder, value.as_ref(), *upstream_time);
                        if matches!(&key, Some(Err(_))) || matches!(&value, Some(Err(_))) {
                            n_errors += 1;
                        } else if matches!(&value, Some(Ok(_))) {
                            n_successes += 1;
                        }
                        session.give(DecodeResult {
                            key,
                            value,
                            position: *position,
                            metadata: to_metadata_row(
                                &metadata_items,
                                partition.clone(),
                                *position,
                                *upstream_time,
                            ),
                        });
                    }
                });
                // Matching historical practice, we only log metrics on the value decoder.
                if n_errors > 0 {
                    value_decoder.log_errors(n_errors);
                }
                if n_successes > 0 {
                    value_decoder.log_successes(n_successes);
                }
            }
        },
    );
    (results, None)
}
pub fn render_decode_delimited < G > ( stream : & Stream < G , SourceOutput < Option < Vec < u8 > > , Option < Vec < u8 > > > > , key_encoding : Option < DataEncoding > , value_encoding : DataEncoding , debug_name : & str , envelope : & SourceEnvelope , metadata_items : Vec < IncludedColumnSource > , // Information about optional transformations that can be eagerly done. // If the decoding elects to perform them, it should replace this with // `None`. operators : & mut Option < LinearOperator > , fast_forwarded : bool , metrics : Metrics , ) -> ( Stream < G , DecodeResult > , Option < Box < dyn Any > > ) where G : Scope , { let op_name = format ! ( " " key_encoding as_ref ( ) map ( key_encoding key_encoding op_name ( ) ) unwrap_or ( " " ) value_encoding op_name ( ) ) ; let mut key_decoder = key_encoding . map ( | key_encoding | { get_decoder ( key_encoding , debug_name , & SourceEnvelope :: None ( KeyEnvelope :: None ) , operators , false , true , metrics . clone ( ) , ) } ) ; let mut value_decoder = get_decoder ( value_encoding , debug_name , envelope , operators , fast_forwarded , true , metrics , ) ; // The Debezium deduplication and upsert logic rely on elements for the same key going to the same worker. // Other decoders don't care; so we distribute things round-robin (i.e., by "position"), and // fall back to arbitrarily hashing by value if the upstream didn't give us a position. let use_key_contract = matches ! ( envelope SourceEnvelope Debezium ( ) ) ; let results = stream . unary_frontier ( Exchange :: new ( move | x : & SourceOutput < Option < Vec < u8 > > , Option < Vec < u8 > > > | { if use_key_contract { x . key . hashed ( ) } else if let Some ( position ) = x . position { position . hashed ( ) } else { x . value . hashed ( ) } } ) , & op_name , move | _ , _ | { move | input , output | { let mut n_errors = 0 ; let mut n_successes = 0 ; input . for_each ( | cap , data | { let mut session = output . session ( & cap ) ; for SourceOutput { key , value , position , upstream_time_millis : upstream_time , partition , } in data . iter ( ) { let key = key_decoder . as_mut ( ) . and_then ( | decoder | try_decode ( decoder , key . as_ref ( ) , * upstream_time ) ) ; let value = try_decode ( & mut value_decoder , value . as_ref ( ) , * upstream_time ) ; if matches ! ( key Some ( Err ( ) ) ) || matches ! ( value Some ( Err ( ) ) ) { n_errors += 1 ; } else if matches ! ( value Some ( Ok ( ) ) ) { n_successes += 1 ; } session . give ( DecodeResult { key , value , position : * position , metadata : to_metadata_row ( & metadata_items , partition . clone ( ) , * position , * upstream_time , ) , } ) ; } } ) ; // Matching historical practice, we only log metrics on the value decoder. if n_errors > 0 { value_decoder . log_errors ( n_errors ) ; } if n_successes > 0 { value_decoder . log_successes ( n_successes ) ; } } } , ) ; ( results , None ) }
Decode already delimited records of data.
Decode already delimited records of data .
stream
&Stream<G, SourceOutput<Option<Vec<u8>>, Option<Vec<u8>>>>
key_encoding
Option<DataEncoding>
value_encoding
DataEncoding
debug_name
&str
envelope
&SourceEnvelope
metadata_items
Vec<IncludedColumnSource>
operators
&mut Option<LinearOperator>
fast_forwarded
bool
metrics
Metrics
stream
&Stream<G, SourceOutput<Option<Vec<u8>>, Option<Vec<u8>>>>
key_encoding
Option<DataEncoding>
value_encoding
DataEncoding
debug_name
&str
envelope
&SourceEnvelope
metadata_items
Vec<IncludedColumnSource>
operators
&mut Option<LinearOperator>
fast_forwarded
bool
metrics
Metrics