pub fn desc(
        &self,
        key_desc: Option<RelationDesc>,
        value_desc: RelationDesc,
        metadata_desc: RelationDesc,
    ) -> anyhow::Result<RelationDesc> {
        Ok(match self {
            SourceEnvelope::None(key_envelope) | SourceEnvelope::Upsert(key_envelope) => {
                let key_desc = match key_desc {
                    Some(desc) => desc,
                    None => return Ok(value_desc.concat(metadata_desc)),
                };

                let keyed = match key_envelope {
                    KeyEnvelope::None => value_desc,
                    KeyEnvelope::Flattened => {
                        // Add the key columns as a key.
                        let key_indices = (0..key_desc.arity()).collect();
                        let key_desc = key_desc.with_key(key_indices);
                        key_desc.concat(value_desc)
                    }
                    KeyEnvelope::LegacyUpsert => {
                        let key_indices = (0..key_desc.arity()).collect();
                        let key_desc = key_desc.with_key(key_indices);
                        let names = (0..key_desc.arity()).map(|i| format!("key{}", i));
                        // Rename key columns to "keyN"
                        key_desc.with_names(names).concat(value_desc)
                    }
                    KeyEnvelope::Named(key_name) => {
                        let key_desc = {
                            // if the key has multiple objects, nest them as a record inside of a single name
                            if key_desc.arity() > 1 {
                                let key_type = key_desc.typ();
                                let key_as_record = RelationType::new(vec![ColumnType {
                                    nullable: false,
                                    scalar_type: ScalarType::Record {
                                        fields: key_desc
                                            .iter_names()
                                            .zip(key_type.column_types.iter())
                                            .map(|(name, ty)| (name.clone(), ty.clone()))
                                            .collect(),
                                        custom_oid: None,
                                        custom_name: None,
                                    },
                                }]);

                                RelationDesc::new(key_as_record, [key_name.to_string()])
                            } else {
                                key_desc.with_names([key_name.to_string()])
                            }
                        };
                        // In all cases the first column is the key
                        key_desc.with_key(vec![0]).concat(value_desc)
                    }
                };
                keyed.concat(metadata_desc)
            }
            SourceEnvelope::Debezium(..) => {
                // TODO [btv] - Get rid of this, when we can. Right now source processing is still
                // not fully orthogonal, so we have some special logic in the debezium processor
                // that only passes on the first two columns ("before" and "after"), and uses the
                // information in the other ones itself
                let mut diter = value_desc.into_iter();
                let mut key = diter
                    .next()
                    .ok_or_else(|| anyhow::anyhow!("Spec does not contain key"))?;
                match &mut key.1.scalar_type {
                    ScalarType::Record { fields, .. } => {
                        fields.extend(metadata_desc.iter().map(|(k, v)| (k.clone(), v.clone())))
                    }
                    ty => bail!(
                        "Incorrect type for Debezium value, expected Record, got {:?}",
                        ty
                    ),
                }
                let mut value = diter
                    .next()
                    .ok_or_else(|| anyhow::anyhow!("Spec does not contain value"))?;
                match &mut value.1.scalar_type {
                    ScalarType::Record { fields, .. } => fields.extend(metadata_desc),
                    ty => bail!(
                        "Incorrect type for Debezium value, expected Record, got {:?}",
                        ty
                    ),
                }
                RelationDesc::from_names_and_types([key, value].into_iter())
            }
            SourceEnvelope::CdcV2 => {
                // TODO: Validate that the value schema has an `updates` and `progress` column of
                // the correct types

                // CdcV2 row data are in a record in a record in a list
                match &value_desc.typ().column_types[0].scalar_type {
                    ScalarType::List { element_type, .. } => match &**element_type {
                        ScalarType::Record { fields, .. } => {
                            // TODO maybe check this by name
                            match &fields[0].1.scalar_type {
                                ScalarType::Record { fields, .. } => {
                                    RelationDesc::from_names_and_types(fields.clone())
                                }
                                ty => bail!("Unepxected type for MATERIALIZE envelope: {:?}", ty),
                            }
                        }
                        ty => bail!("Unepxected type for MATERIALIZE envelope: {:?}", ty),
                    },
                    ty => bail!("Unepxected type for MATERIALIZE envelope: {:?}", ty),
                }
            }
        })
    }
/// Computes the output relation of this envelope when applied on top of the decoded key and
/// value relation desc
Computes the output relation of this envelope when applied on top of the decoded key and
value relation desc
Computes the output relation of this envelope when applied on top of the decoded key and value relation desc
pub fn desc(
        &self,
        key_desc: Option<RelationDesc>,
        value_desc: RelationDesc,
        metadata_desc: RelationDesc,
    ) -> anyhow::Result<RelationDesc> {
        Ok(match self {
            SourceEnvelope::None(key_envelope) | SourceEnvelope::Upsert(key_envelope) => {
                let key_desc = match key_desc {
                    Some(desc) => desc,
                    None => return Ok(value_desc.concat(metadata_desc)),
                };
                let keyed = match key_envelope {
                    KeyEnvelope::None => value_desc,
                    KeyEnvelope::Flattened => {
                        // Add the key columns as a key.
                        let key_indices = (0..key_desc.arity()).collect();
                        let key_desc = key_desc.with_key(key_indices);
                        key_desc.concat(value_desc)
                    }
                    KeyEnvelope::LegacyUpsert => {
                        let key_indices = (0..key_desc.arity()).collect();
                        let key_desc = key_desc.with_key(key_indices);
                        let names = (0..key_desc.arity()).map(|i| format!("key{}", i));
                        // Rename key columns to "keyN"
                        key_desc.with_names(names).concat(value_desc)
                    }
                    KeyEnvelope::Named(key_name) => {
                        let key_desc = {
                            // if the key has multiple objects, nest them as a record inside of a single name
                            if key_desc.arity() > 1 {
                                let key_type = key_desc.typ();
                                let key_as_record = RelationType::new(vec![ColumnType {
                                    nullable: false,
                                    scalar_type: ScalarType::Record {
                                        fields: key_desc
                                            .iter_names()
                                            .zip(key_type.column_types.iter())
                                            .map(|(name, ty)| (name.clone(), ty.clone()))
                                            .collect(),
                                        custom_oid: None,
                                        custom_name: None,
                                    },
                                }]);
                                RelationDesc::new(key_as_record, [key_name.to_string()])
                            } else {
                                key_desc.with_names([key_name.to_string()])
                            }
                        };
                        // In all cases the first column is the key
                        key_desc.with_key(vec![0]).concat(value_desc)
                    }
                };
                keyed.concat(metadata_desc)
            }
            SourceEnvelope::Debezium(..) => {
                // TODO [btv] - Get rid of this, when we can. Right now source processing is still
                // not fully orthogonal, so we have some special logic in the debezium processor
                // that only passes on the first two columns ("before" and "after"), and uses the
                // information in the other ones itself
                let mut diter = value_desc.into_iter();
                let mut key = diter
                    .next()
                    .ok_or_else(|| anyhow::anyhow!("Spec does not contain key"))?;
                match &mut key.1.scalar_type {
                    ScalarType::Record { fields, .. } => {
                        fields.extend(metadata_desc.iter().map(|(k, v)| (k.clone(), v.clone())))
                    }
                    ty => bail!(
                        "Incorrect type for Debezium value, expected Record, got {:?}",
                        ty
                    ),
                }
                let mut value = diter
                    .next()
                    .ok_or_else(|| anyhow::anyhow!("Spec does not contain value"))?;
                match &mut value.1.scalar_type {
                    ScalarType::Record { fields, .. } => fields.extend(metadata_desc),
                    ty => bail!(
                        "Incorrect type for Debezium value, expected Record, got {:?}",
                        ty
                    ),
                }
                RelationDesc::from_names_and_types([key, value].into_iter())
            }
            SourceEnvelope::CdcV2 => {
                // TODO: Validate that the value schema has an `updates` and `progress` column of
                // the correct types
                // CdcV2 row data are in a record in a record in a list
                match &value_desc.typ().column_types[0].scalar_type {
                    ScalarType::List { element_type, .. } => match &**element_type {
                        ScalarType::Record { fields, .. } => {
                            // TODO maybe check this by name
                            match &fields[0].1.scalar_type {
                                ScalarType::Record { fields, .. } => {
                                    RelationDesc::from_names_and_types(fields.clone())
                                }
                                ty => bail!("Unepxected type for MATERIALIZE envelope: {:?}", ty),
                            }
                        }
                        ty => bail!("Unepxected type for MATERIALIZE envelope: {:?}", ty),
                    },
                    ty => bail!("Unepxected type for MATERIALIZE envelope: {:?}", ty),
                }
            }
        })
    }
pub fn desc ( & self , key_desc : Option < RelationDesc > , value_desc : RelationDesc , metadata_desc : RelationDesc , ) -> anyhow :: Result < RelationDesc > { Ok ( match self { SourceEnvelope :: None ( key_envelope ) | SourceEnvelope :: Upsert ( key_envelope ) => { let key_desc = match key_desc { Some ( desc ) => desc , None => return Ok ( value_desc . concat ( metadata_desc ) ) , } ; let keyed = match key_envelope { KeyEnvelope :: None => value_desc , KeyEnvelope :: Flattened => { // Add the key columns as a key. let key_indices = ( 0 .. key_desc . arity ( ) ) . collect ( ) ; let key_desc = key_desc . with_key ( key_indices ) ; key_desc . concat ( value_desc ) } KeyEnvelope :: LegacyUpsert => { let key_indices = ( 0 .. key_desc . arity ( ) ) . collect ( ) ; let key_desc = key_desc . with_key ( key_indices ) ; let names = ( 0 .. key_desc . arity ( ) ) . map ( | i | format ! ( " " i ) ) ; // Rename key columns to "keyN" key_desc . with_names ( names ) . concat ( value_desc ) } KeyEnvelope :: Named ( key_name ) => { let key_desc = { // if the key has multiple objects, nest them as a record inside of a single name if key_desc . arity ( ) > 1 { let key_type = key_desc . typ ( ) ; let key_as_record = RelationType :: new ( vec ! [ ColumnType { nullable false scalar_type ScalarType Record { fields key_desc iter_names ( ) zip ( key_type column_types iter ( ) ) map ( ( name ty ) ( name clone ( ) ty clone ( ) ) ) collect ( ) custom_oid None custom_name None } } ] ) ; RelationDesc :: new ( key_as_record , [ key_name . to_string ( ) ] ) } else { key_desc . with_names ( [ key_name . to_string ( ) ] ) } } ; // In all cases the first column is the key key_desc . with_key ( vec ! [ 0 ] ) . concat ( value_desc ) } } ; keyed . concat ( metadata_desc ) } SourceEnvelope :: Debezium ( .. ) => { // TODO [btv] - Get rid of this, when we can. Right now source processing is still // not fully orthogonal, so we have some special logic in the debezium processor // that only passes on the first two columns ("before" and "after"), and uses the // information in the other ones itself let mut diter = value_desc . into_iter ( ) ; let mut key = diter . next ( ) . ok_or_else ( | | anyhow :: anyhow ! ( " " ) ) ? ; match & mut key . 1 . scalar_type { ScalarType :: Record { fields , .. } => { fields . extend ( metadata_desc . iter ( ) . map ( | ( k , v ) | ( k . clone ( ) , v . clone ( ) ) ) ) } ty => bail ! ( " " ty ) , } let mut value = diter . next ( ) . ok_or_else ( | | anyhow :: anyhow ! ( " " ) ) ? ; match & mut value . 1 . scalar_type { ScalarType :: Record { fields , .. } => fields . extend ( metadata_desc ) , ty => bail ! ( " " ty ) , } RelationDesc :: from_names_and_types ( [ key , value ] . into_iter ( ) ) } SourceEnvelope :: CdcV2 => { // TODO: Validate that the value schema has an `updates` and `progress` column of // the correct types // CdcV2 row data are in a record in a record in a list match & value_desc . typ ( ) . column_types [ 0 ] . scalar_type { ScalarType :: List { element_type , .. } => match & * * element_type { ScalarType :: Record { fields , .. } => { // TODO maybe check this by name match & fields [ 0 ] . 1 . scalar_type { ScalarType :: Record { fields , .. } => { RelationDesc :: from_names_and_types ( fields . clone ( ) ) } ty => bail ! ( " " ty ) , } } ty => bail ! ( " " ty ) , } , ty => bail ! ( " " ty ) , } } } ) }
Computes the output relation of this envelope when applied on top of the decoded key and
value relation desc
Computes the output relation of this envelope when applied on top of the decoded key and value relation desc
&self
key_desc
Option<RelationDesc>
value_desc
RelationDesc
metadata_desc
RelationDesc
&self
key_desc
Option<RelationDesc>
value_desc
RelationDesc
metadata_desc
RelationDesc