pub fn render_decode<G>(
    stream: &Stream<G, SourceOutput<(), MessagePayload>>,
    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<Timestamp = Timestamp>,
{
    let op_name = format!("{}Decode", value_encoding.op_name());

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

    let mut value_buf = vec![];

    // The `position` value from `SourceOutput` is meaningless here -- it's just the index of a chunk.
    // We therefore ignore it, and keep track ourselves of how many records we've seen (for filling in `mz_line_no`, etc).
    // Historically, non-delimited sources have their offset start at 1
    let mut n_seen = 1..;
    let results = stream.unary_frontier(Pipeline, &op_name, move |_, _| {
        let metadata_items = metadata_items;
        move |input, output| {
            let mut n_errors = 0;
            let mut n_successes = 0;
            input.for_each(|cap, data| {
                // Currently Kafka is the only kind of source that can have metadata, and it is
                // always delimited, so we will never have metadata in `render_decode`
                let mut session = output.session(&cap);
                for SourceOutput {
                    key: _,
                    value,
                    position: _,
                    upstream_time_millis,
                    partition,
                } in data.iter()
                {
                    let value = match value {
                        MessagePayload::Data(data) => data,
                        MessagePayload::EOF => {
                            let data = &mut &value_buf[..];
                            let mut result = value_decoder.eof(data);
                            if result.is_ok() && !data.is_empty() {
                                result = Err(DecodeError::Text(format!(
                                    "Saw unexpected EOF with bytes remaining in buffer: {:?}",
                                    data
                                )));
                            }
                            value_buf.clear();

                            match result.transpose() {
                                None => continue,
                                Some(value) => {
                                    if value.is_err() {
                                        n_errors += 1;
                                    } else if matches!(&value, Ok(_)) {
                                        n_successes += 1;
                                    }
                                    let position = n_seen.next();
                                    let metadata = to_metadata_row(
                                        &metadata_items,
                                        partition.clone(),
                                        position,
                                        *upstream_time_millis,
                                    );

                                    session.give(DecodeResult {
                                        key: None,
                                        value: Some(value),
                                        position,
                                        metadata,
                                    });
                                }
                            }
                            continue;
                        }
                    };

                    // Check whether we have a partial message from last time.
                    // If so, we need to prepend it to the bytes we got from _this_ message.
                    let value = if value_buf.is_empty() {
                        value
                    } else {
                        value_buf.extend_from_slice(&*value);
                        &value_buf
                    };

                    if value.is_empty() {
                        let metadata = to_metadata_row(
                            &metadata_items,
                            partition.clone(),
                            None,
                            *upstream_time_millis,
                        );
                        session.give(DecodeResult {
                            key: None,
                            value: None,
                            position: None,
                            metadata,
                        });
                    } else {
                        let value_bytes_remaining = &mut value.as_slice();
                        // The intent is that the below loop runs as long as there are more bytes to decode.
                        //
                        // We'd like to be able to write `while !value_cursor.empty()`
                        // here, but that runs into borrow checker issues, so we use `loop`
                        // and break manually.
                        loop {
                            let old_value_cursor = *value_bytes_remaining;
                            let value = match value_decoder
                                .next(value_bytes_remaining, *upstream_time_millis)
                            {
                                Err(e) => Err(e),
                                Ok(None) => {
                                    let leftover = value_bytes_remaining.to_vec();
                                    value_buf = leftover;
                                    break;
                                }
                                Ok(Some(value)) => Ok(value),
                            };

                            // If the decoders decoded a message, they need to have made progress consuming the bytes.
                            // Otherwise, we risk going into an infinite loop.
                            assert!(old_value_cursor != *value_bytes_remaining || value.is_err());

                            let is_err = value.is_err();
                            if is_err {
                                n_errors += 1;
                            } else if matches!(&value, Ok(_)) {
                                n_successes += 1;
                            }
                            let position = n_seen.next();
                            let metadata = to_metadata_row(
                                &metadata_items,
                                partition.clone(),
                                position,
                                *upstream_time_millis,
                            );

                            if value_bytes_remaining.is_empty() {
                                session.give(DecodeResult {
                                    key: None,
                                    value: Some(value),
                                    position,
                                    metadata,
                                });
                                value_buf = vec![];
                                break;
                            } else {
                                session.give(DecodeResult {
                                    key: None,
                                    value: Some(value),
                                    position,
                                    metadata,
                                });
                            }
                            if is_err {
                                // If decoding has gone off the rails, we can no longer be sure that the delimiters are correct, so it
                                // makes no sense to keep going.
                                break;
                            }
                        }
                    }
                }
            });
            // 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_errors);
            }
        }
    });
    (results, None)
}
/// Decode arbitrary chunks of bytes into rows.
///
/// This decode API is used for upstream connectors
/// that don't discover delimiters themselves; i.e., those
/// (like CSV files) that need help from the decoding stage to discover where
/// one record ends and another begins.
///
/// As such, the connectors simply present arbitrary chunks of bytes about which
/// we can't assume any alignment properties. The `DataDecoder` API accepts these,
/// and returns `None` if it needs more bytes to discover the boundary between messages.
/// In that case, this function remembers the already-seen bytes and waits for new ones
/// before calling into the decoder again.
///
/// If the decoder does find a message, we verify (by asserting) that it consumed some bytes, to avoid
/// the possibility of infinite loops.
Decode arbitrary chunks of bytes into rows.
This decode API is used for upstream connectors
that don't discover delimiters themselves; i.e., those
(like CSV files) that need help from the decoding stage to discover where
one record ends and another begins.

As such, the connectors simply present arbitrary chunks of bytes about which
we can't assume any alignment properties. The `DataDecoder` API accepts these,
and returns `None` if it needs more bytes to discover the boundary between messages.
In that case, this function remembers the already-seen bytes and waits for new ones
before calling into the decoder again.

If the decoder does find a message, we verify (by asserting) that it consumed some bytes, to avoid
the possibility of infinite loops.
Decode arbitrary chunks of bytes into rows . This decode API is used for upstream connectors that don ' t discover delimiters themselves ; i . e . those ( like CSV files ) that need help from the decoding stage to discover where one record ends and another begins . As such the connectors simply present arbitrary chunks of bytes about which we can ' t assume any alignment properties . The ` DataDecoder ` API accepts these and returns ` None ` if it needs more bytes to discover the boundary between messages . In that case this function remembers the already - seen bytes and waits for new ones before calling into the decoder again . If the decoder does find a message we verify ( by asserting ) that it consumed some bytes to avoid the possibility of infinite loops .
pub fn render_decode<G>(
    stream: &Stream<G, SourceOutput<(), MessagePayload>>,
    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<Timestamp = Timestamp>,
{
    let op_name = format!("{}Decode", value_encoding.op_name());
    let mut value_decoder = get_decoder(
        value_encoding,
        debug_name,
        envelope,
        operators,
        fast_forwarded,
        false,
        metrics,
    );
    let mut value_buf = vec![];
    // The `position` value from `SourceOutput` is meaningless here -- it's just the index of a chunk.
    // We therefore ignore it, and keep track ourselves of how many records we've seen (for filling in `mz_line_no`, etc).
    // Historically, non-delimited sources have their offset start at 1
    let mut n_seen = 1..;
    let results = stream.unary_frontier(Pipeline, &op_name, move |_, _| {
        let metadata_items = metadata_items;
        move |input, output| {
            let mut n_errors = 0;
            let mut n_successes = 0;
            input.for_each(|cap, data| {
                // Currently Kafka is the only kind of source that can have metadata, and it is
                // always delimited, so we will never have metadata in `render_decode`
                let mut session = output.session(&cap);
                for SourceOutput {
                    key: _,
                    value,
                    position: _,
                    upstream_time_millis,
                    partition,
                } in data.iter()
                {
                    let value = match value {
                        MessagePayload::Data(data) => data,
                        MessagePayload::EOF => {
                            let data = &mut &value_buf[..];
                            let mut result = value_decoder.eof(data);
                            if result.is_ok() && !data.is_empty() {
                                result = Err(DecodeError::Text(format!(
                                    "Saw unexpected EOF with bytes remaining in buffer: {:?}",
                                    data
                                )));
                            }
                            value_buf.clear();
                            match result.transpose() {
                                None => continue,
                                Some(value) => {
                                    if value.is_err() {
                                        n_errors += 1;
                                    } else if matches!(&value, Ok(_)) {
                                        n_successes += 1;
                                    }
                                    let position = n_seen.next();
                                    let metadata = to_metadata_row(
                                        &metadata_items,
                                        partition.clone(),
                                        position,
                                        *upstream_time_millis,
                                    );
                                    session.give(DecodeResult {
                                        key: None,
                                        value: Some(value),
                                        position,
                                        metadata,
                                    });
                                }
                            }
                            continue;
                        }
                    };
                    // Check whether we have a partial message from last time.
                    // If so, we need to prepend it to the bytes we got from _this_ message.
                    let value = if value_buf.is_empty() {
                        value
                    } else {
                        value_buf.extend_from_slice(&*value);
                        &value_buf
                    };
                    if value.is_empty() {
                        let metadata = to_metadata_row(
                            &metadata_items,
                            partition.clone(),
                            None,
                            *upstream_time_millis,
                        );
                        session.give(DecodeResult {
                            key: None,
                            value: None,
                            position: None,
                            metadata,
                        });
                    } else {
                        let value_bytes_remaining = &mut value.as_slice();
                        // The intent is that the below loop runs as long as there are more bytes to decode.
                        //
                        // We'd like to be able to write `while !value_cursor.empty()`
                        // here, but that runs into borrow checker issues, so we use `loop`
                        // and break manually.
                        loop {
                            let old_value_cursor = *value_bytes_remaining;
                            let value = match value_decoder
                                .next(value_bytes_remaining, *upstream_time_millis)
                            {
                                Err(e) => Err(e),
                                Ok(None) => {
                                    let leftover = value_bytes_remaining.to_vec();
                                    value_buf = leftover;
                                    break;
                                }
                                Ok(Some(value)) => Ok(value),
                            };
                            // If the decoders decoded a message, they need to have made progress consuming the bytes.
                            // Otherwise, we risk going into an infinite loop.
                            assert!(old_value_cursor != *value_bytes_remaining || value.is_err());
                            let is_err = value.is_err();
                            if is_err {
                                n_errors += 1;
                            } else if matches!(&value, Ok(_)) {
                                n_successes += 1;
                            }
                            let position = n_seen.next();
                            let metadata = to_metadata_row(
                                &metadata_items,
                                partition.clone(),
                                position,
                                *upstream_time_millis,
                            );
                            if value_bytes_remaining.is_empty() {
                                session.give(DecodeResult {
                                    key: None,
                                    value: Some(value),
                                    position,
                                    metadata,
                                });
                                value_buf = vec![];
                                break;
                            } else {
                                session.give(DecodeResult {
                                    key: None,
                                    value: Some(value),
                                    position,
                                    metadata,
                                });
                            }
                            if is_err {
                                // If decoding has gone off the rails, we can no longer be sure that the delimiters are correct, so it
                                // makes no sense to keep going.
                                break;
                            }
                        }
                    }
                }
            });
            // 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_errors);
            }
        }
    });
    (results, None)
}
pub fn render_decode < G > ( stream : & Stream < G , SourceOutput < ( ) , MessagePayload > > , 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 < Timestamp = Timestamp > , { let op_name = format ! ( " " value_encoding op_name ( ) ) ; let mut value_decoder = get_decoder ( value_encoding , debug_name , envelope , operators , fast_forwarded , false , metrics , ) ; let mut value_buf = vec ! [ ] ; // The `position` value from `SourceOutput` is meaningless here -- it's just the index of a chunk. // We therefore ignore it, and keep track ourselves of how many records we've seen (for filling in `mz_line_no`, etc). // Historically, non-delimited sources have their offset start at 1 let mut n_seen = 1 .. ; let results = stream . unary_frontier ( Pipeline , & op_name , move | _ , _ | { let metadata_items = metadata_items ; move | input , output | { let mut n_errors = 0 ; let mut n_successes = 0 ; input . for_each ( | cap , data | { // Currently Kafka is the only kind of source that can have metadata, and it is // always delimited, so we will never have metadata in `render_decode` let mut session = output . session ( & cap ) ; for SourceOutput { key : _ , value , position : _ , upstream_time_millis , partition , } in data . iter ( ) { let value = match value { MessagePayload :: Data ( data ) => data , MessagePayload :: EOF => { let data = & mut & value_buf [ .. ] ; let mut result = value_decoder . eof ( data ) ; if result . is_ok ( ) && ! data . is_empty ( ) { result = Err ( DecodeError :: Text ( format ! ( " " data ) ) ) ; } value_buf . clear ( ) ; match result . transpose ( ) { None => continue , Some ( value ) => { if value . is_err ( ) { n_errors += 1 ; } else if matches ! ( value Ok ( ) ) { n_successes += 1 ; } let position = n_seen . next ( ) ; let metadata = to_metadata_row ( & metadata_items , partition . clone ( ) , position , * upstream_time_millis , ) ; session . give ( DecodeResult { key : None , value : Some ( value ) , position , metadata , } ) ; } } continue ; } } ; // Check whether we have a partial message from last time. // If so, we need to prepend it to the bytes we got from _this_ message. let value = if value_buf . is_empty ( ) { value } else { value_buf . extend_from_slice ( & * value ) ; & value_buf } ; if value . is_empty ( ) { let metadata = to_metadata_row ( & metadata_items , partition . clone ( ) , None , * upstream_time_millis , ) ; session . give ( DecodeResult { key : None , value : None , position : None , metadata , } ) ; } else { let value_bytes_remaining = & mut value . as_slice ( ) ; // The intent is that the below loop runs as long as there are more bytes to decode. // // We'd like to be able to write `while !value_cursor.empty()` // here, but that runs into borrow checker issues, so we use `loop` // and break manually. loop { let old_value_cursor = * value_bytes_remaining ; let value = match value_decoder . next ( value_bytes_remaining , * upstream_time_millis ) { Err ( e ) => Err ( e ) , Ok ( None ) => { let leftover = value_bytes_remaining . to_vec ( ) ; value_buf = leftover ; break ; } Ok ( Some ( value ) ) => Ok ( value ) , } ; // If the decoders decoded a message, they need to have made progress consuming the bytes. // Otherwise, we risk going into an infinite loop. assert ! ( old_value_cursor value_bytes_remaining value is_err ( ) ) ; let is_err = value . is_err ( ) ; if is_err { n_errors += 1 ; } else if matches ! ( value Ok ( ) ) { n_successes += 1 ; } let position = n_seen . next ( ) ; let metadata = to_metadata_row ( & metadata_items , partition . clone ( ) , position , * upstream_time_millis , ) ; if value_bytes_remaining . is_empty ( ) { session . give ( DecodeResult { key : None , value : Some ( value ) , position , metadata , } ) ; value_buf = vec ! [ ] ; break ; } else { session . give ( DecodeResult { key : None , value : Some ( value ) , position , metadata , } ) ; } if is_err { // If decoding has gone off the rails, we can no longer be sure that the delimiters are correct, so it // makes no sense to keep going. break ; } } } } } ) ; // 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_errors ) ; } } } ) ; ( results , None ) }
Decode arbitrary chunks of bytes into rows.
Decode arbitrary chunks of bytes into rows .
stream
&Stream<G, SourceOutput<(), MessagePayload>>
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<(), MessagePayload>>
value_encoding
DataEncoding
debug_name
&str
envelope
&SourceEnvelope
metadata_items
Vec<IncludedColumnSource>
operators
&mut Option<LinearOperator>
fast_forwarded
bool
metrics
Metrics