pub fn rewrite_for_upsert(
    val: Result<Row, DecodeError>,
    metadata: Row,
    keys: &mut HashMap<Row, Row>,
    key: Row,
    metrics: &UIntGauge,
) -> Result<Row, DecodeError> {
    if let Ok(row) = val {
        // often off by one, but is only tracked every N seconds so it will always be off
        metrics.set(keys.len() as u64);

        let entry = keys.entry(key);

        // Upsert-shaped data must be of shape Row[[before_row_as_list], [after_row_as_list]]
        let mut rowiter = row.iter();
        let before = rowiter.next().expect("must have a before list");
        let after = rowiter.next().expect("must have an after list");

        let mut arena = Row::default();
        let after_md = concat_datum_list(&mut arena, after, metadata);

        assert!(
            matches!(before, Datum::List { .. } | Datum::Null),
            "[customer-data] Debezium logic should be a List or absent, got {:?}",
            before
        );

        match entry {
            Entry::Vacant(vacant) => {
                // if the key is new, then we know that we always need to ignore the "before" part,
                // so zero it out
                vacant.insert(Row::pack_slice(&[after_md]));

                Ok(Row::pack_slice(&[Datum::Null, after_md]))
            }
            Entry::Occupied(mut occupied) => {
                let previous_insert = if after.is_null() {
                    // We are trying to retract something that doesn't exist, so just assume
                    // that the key is supposed to be empty at this point
                    let (_k, v) = occupied.remove_entry();
                    v
                } else {
                    occupied.insert(Row::pack_slice(&[after_md]))
                };

                Ok(Row::pack_slice(&[previous_insert.unpack_first(), after_md]))
            }
        }
    } else {
        val
    }
}
/// Update row to blank out retractions of rows that we have never seen
Update row to blank out retractions of rows that we have never seen
Update row to blank out retractions of rows that we have never seen
pub fn rewrite_for_upsert(
    val: Result<Row, DecodeError>,
    metadata: Row,
    keys: &mut HashMap<Row, Row>,
    key: Row,
    metrics: &UIntGauge,
) -> Result<Row, DecodeError> {
    if let Ok(row) = val {
        // often off by one, but is only tracked every N seconds so it will always be off
        metrics.set(keys.len() as u64);
        let entry = keys.entry(key);
        // Upsert-shaped data must be of shape Row[[before_row_as_list], [after_row_as_list]]
        let mut rowiter = row.iter();
        let before = rowiter.next().expect("must have a before list");
        let after = rowiter.next().expect("must have an after list");
        let mut arena = Row::default();
        let after_md = concat_datum_list(&mut arena, after, metadata);
        assert!(
            matches!(before, Datum::List { .. } | Datum::Null),
            "[customer-data] Debezium logic should be a List or absent, got {:?}",
            before
        );
        match entry {
            Entry::Vacant(vacant) => {
                // if the key is new, then we know that we always need to ignore the "before" part,
                // so zero it out
                vacant.insert(Row::pack_slice(&[after_md]));
                Ok(Row::pack_slice(&[Datum::Null, after_md]))
            }
            Entry::Occupied(mut occupied) => {
                let previous_insert = if after.is_null() {
                    // We are trying to retract something that doesn't exist, so just assume
                    // that the key is supposed to be empty at this point
                    let (_k, v) = occupied.remove_entry();
                    v
                } else {
                    occupied.insert(Row::pack_slice(&[after_md]))
                };
                Ok(Row::pack_slice(&[previous_insert.unpack_first(), after_md]))
            }
        }
    } else {
        val
    }
}
pub fn rewrite_for_upsert ( val : Result < Row , DecodeError > , metadata : Row , keys : & mut HashMap < Row , Row > , key : Row , metrics : & UIntGauge , ) -> Result < Row , DecodeError > { if let Ok ( row ) = val { // often off by one, but is only tracked every N seconds so it will always be off metrics . set ( keys . len ( ) as u64 ) ; let entry = keys . entry ( key ) ; // Upsert-shaped data must be of shape Row[[before_row_as_list], [after_row_as_list]] let mut rowiter = row . iter ( ) ; let before = rowiter . next ( ) . expect ( " " ) ; let after = rowiter . next ( ) . expect ( " " ) ; let mut arena = Row :: default ( ) ; let after_md = concat_datum_list ( & mut arena , after , metadata ) ; assert ! ( matches ( before Datum List { } Datum Null ) " " before ) ; match entry { Entry :: Vacant ( vacant ) => { // if the key is new, then we know that we always need to ignore the "before" part, // so zero it out vacant . insert ( Row :: pack_slice ( & [ after_md ] ) ) ; Ok ( Row :: pack_slice ( & [ Datum :: Null , after_md ] ) ) } Entry :: Occupied ( mut occupied ) => { let previous_insert = if after . is_null ( ) { // We are trying to retract something that doesn't exist, so just assume // that the key is supposed to be empty at this point let ( _k , v ) = occupied . remove_entry ( ) ; v } else { occupied . insert ( Row :: pack_slice ( & [ after_md ] ) ) } ; Ok ( Row :: pack_slice ( & [ previous_insert . unpack_first ( ) , after_md ] ) ) } } } else { val } }
Update row to blank out retractions of rows that we have never seen
Update row to blank out retractions of rows that we have never seen
val
Result<Row, DecodeError>
metadata
Row
keys
&mut HashMap<Row, Row>
key
Row
metrics
&UIntGauge
val
Result<Row, DecodeError>
metadata
Row
keys
&mut HashMap<Row, Row>
key
Row
metrics
&UIntGauge