/// Takes a `RecordBatch` and the column names that make up the primary key of the schema. Returns
/// a map of column names to the set of the distinct string values, for the specified columns. Used
/// to compute cardinality across multiple `RecordBatch`es.
Takes a `RecordBatch` and the column names that make up the primary key of the schema. Returns
a map of column names to the set of the distinct string values, for the specified columns. Used
to compute cardinality across multiple `RecordBatch`es.
Takes a ` RecordBatch ` and the column names that make up the primary key of the schema . Returns a map of column names to the set of the distinct string values for the specified columns . Used to compute cardinality across multiple ` RecordBatch ` es .
fn distinct_values(batch: &RecordBatch, primary_key: &[&str]) -> HashMap<String, HashSet<String>> {
    let schema = batch.schema();
    batch
        .columns()
        .iter()
        .zip(schema.fields())
        .filter(|(_col, field)| primary_key.contains(&field.name().as_str()))
        .flat_map(|(col, field)| match field.data_type() {
            // Dictionaries of I32 => Utf8 are supported as tags in
            // `schema::InfluxColumnType::valid_arrow_type`
            DataType::Dictionary(key, value)
                if key.as_ref() == &DataType::Int32 && value.as_ref() == &DataType::Utf8 =>
            {
                let col = col
                    .as_any()
                    .downcast_ref::<DictionaryArray<Int32Type>>()
                    .expect("unexpected datatype");
                let values = col.values();
                let values = values
                    .as_any()
                    .downcast_ref::<StringArray>()
                    .expect("unexpected datatype");
                Some((
                    field.name().into(),
                    values.iter().flatten().map(ToString::to_string).collect(),
                ))
            }
            // Utf8 types are supported as tags
            DataType::Utf8 => {
                let values = col
                    .as_any()
                    .downcast_ref::<StringArray>()
                    .expect("unexpected datatype");
                Some((
                    field.name().into(),
                    values.iter().flatten().map(ToString::to_string).collect(),
                ))
            }
            // No other data types are supported as tags; don't compute distinct values for them
            _ => None,
        })
        .collect()
}
fn distinct_values ( batch : & RecordBatch , primary_key : & [ & str ] ) -> HashMap < String , HashSet < String > > { let schema = batch . schema ( ) ; batch . columns ( ) . iter ( ) . zip ( schema . fields ( ) ) . filter ( | ( _col , field ) | primary_key . contains ( & field . name ( ) . as_str ( ) ) ) . flat_map ( | ( col , field ) | match field . data_type ( ) { // Dictionaries of I32 => Utf8 are supported as tags in // `schema::InfluxColumnType::valid_arrow_type` DataType :: Dictionary ( key , value ) if key . as_ref ( ) == & DataType :: Int32 && value . as_ref ( ) == & DataType :: Utf8 => { let col = col . as_any ( ) . downcast_ref :: < DictionaryArray < Int32Type > > ( ) . expect ( " " ) ; let values = col . values ( ) ; let values = values . as_any ( ) . downcast_ref :: < StringArray > ( ) . expect ( " " ) ; Some ( ( field . name ( ) . into ( ) , values . iter ( ) . flatten ( ) . map ( ToString :: to_string ) . collect ( ) , ) ) } // Utf8 types are supported as tags DataType :: Utf8 => { let values = col . as_any ( ) . downcast_ref :: < StringArray > ( ) . expect ( " " ) ; Some ( ( field . name ( ) . into ( ) , values . iter ( ) . flatten ( ) . map ( ToString :: to_string ) . collect ( ) , ) ) } // No other data types are supported as tags; don't compute distinct values for them _ => None , } ) . collect ( ) }
Takes a `RecordBatch` and the column names that make up the primary key of the schema.
Takes a ` RecordBatch ` and the column names that make up the primary key of the schema .
batch
&RecordBatch
primary_key
&[&str]