async fn simulate_write_routing(
        &mut self,
        lp_lines: &[&str],
        partition_key: &str,
    ) -> (DmlOperation, Vec<PartitionId>) {
        // detect table names and schemas from LP lines
        let mut converter = LinesConverter::new(0);
        for lp_line in lp_lines {
            converter.write_lp(lp_line).unwrap();
        }
        let (mutable_batches, _stats) = converter.finish().unwrap();

        // set up catalog
        let tables = {
            // sort names so that IDs are deterministic
            let mut table_names: Vec<_> = mutable_batches.keys().cloned().collect();
            table_names.sort();

            let mut tables = vec![];
            for table_name in table_names {
                let table = self.ns.create_table(&table_name).await;
                tables.push(table);
            }
            tables
        };
        let mut partition_ids = vec![];
        for table in &tables {
            let partition = table
                .with_sequencer(&self.sequencer)
                .create_partition(partition_key)
                .await;
            partition_ids.push(partition.partition.id);
        }
        for table in tables {
            let schema = mutable_batches
                .get(&table.table.name)
                .unwrap()
                .schema(Selection::All)
                .unwrap();

            for (t, field) in schema.iter() {
                let t = t.unwrap();
                table.create_column(field.name(), t.into()).await;
            }
        }

        let sequence_number = self.next_sequence_number();
        self.partition_keys.insert(
            SequenceNumber::new(sequence_number as i64),
            partition_key.to_string(),
        );
        let meta = DmlMeta::sequenced(
            Sequence::new(self.sequencer.sequencer.id.get() as u32, sequence_number),
            self.catalog.time_provider().now(),
            None,
            0,
        );
        let op = DmlOperation::Write(DmlWrite::new(
            self.ns.namespace.name.clone(),
            mutable_batches,
            meta,
        ));
        (op, partition_ids)
    }
/// Simulate what the router would do when it encounters the given set of line protocol lines.
///
/// In contrast to a real router we have tight control over the partition key here.
Simulate what the router would do when it encounters the given set of line protocol lines.
In contrast to a real router we have tight control over the partition key here.
Simulate what the router would do when it encounters the given set of line protocol lines . In contrast to a real router we have tight control over the partition key here .
async fn simulate_write_routing(
        &mut self,
        lp_lines: &[&str],
        partition_key: &str,
    ) -> (DmlOperation, Vec<PartitionId>) {
        // detect table names and schemas from LP lines
        let mut converter = LinesConverter::new(0);
        for lp_line in lp_lines {
            converter.write_lp(lp_line).unwrap();
        }
        let (mutable_batches, _stats) = converter.finish().unwrap();
        // set up catalog
        let tables = {
            // sort names so that IDs are deterministic
            let mut table_names: Vec<_> = mutable_batches.keys().cloned().collect();
            table_names.sort();
            let mut tables = vec![];
            for table_name in table_names {
                let table = self.ns.create_table(&table_name).await;
                tables.push(table);
            }
            tables
        };
        let mut partition_ids = vec![];
        for table in &tables {
            let partition = table
                .with_sequencer(&self.sequencer)
                .create_partition(partition_key)
                .await;
            partition_ids.push(partition.partition.id);
        }
        for table in tables {
            let schema = mutable_batches
                .get(&table.table.name)
                .unwrap()
                .schema(Selection::All)
                .unwrap();
            for (t, field) in schema.iter() {
                let t = t.unwrap();
                table.create_column(field.name(), t.into()).await;
            }
        }
        let sequence_number = self.next_sequence_number();
        self.partition_keys.insert(
            SequenceNumber::new(sequence_number as i64),
            partition_key.to_string(),
        );
        let meta = DmlMeta::sequenced(
            Sequence::new(self.sequencer.sequencer.id.get() as u32, sequence_number),
            self.catalog.time_provider().now(),
            None,
            0,
        );
        let op = DmlOperation::Write(DmlWrite::new(
            self.ns.namespace.name.clone(),
            mutable_batches,
            meta,
        ));
        (op, partition_ids)
    }
async fn simulate_write_routing ( & mut self , lp_lines : & [ & str ] , partition_key : & str , ) -> ( DmlOperation , Vec < PartitionId > ) { // detect table names and schemas from LP lines let mut converter = LinesConverter :: new ( 0 ) ; for lp_line in lp_lines { converter . write_lp ( lp_line ) . unwrap ( ) ; } let ( mutable_batches , _stats ) = converter . finish ( ) . unwrap ( ) ; // set up catalog let tables = { // sort names so that IDs are deterministic let mut table_names : Vec < _ > = mutable_batches . keys ( ) . cloned ( ) . collect ( ) ; table_names . sort ( ) ; let mut tables = vec ! [ ] ; for table_name in table_names { let table = self . ns . create_table ( & table_name ) . await ; tables . push ( table ) ; } tables } ; let mut partition_ids = vec ! [ ] ; for table in & tables { let partition = table . with_sequencer ( & self . sequencer ) . create_partition ( partition_key ) . await ; partition_ids . push ( partition . partition . id ) ; } for table in tables { let schema = mutable_batches . get ( & table . table . name ) . unwrap ( ) . schema ( Selection :: All ) . unwrap ( ) ; for ( t , field ) in schema . iter ( ) { let t = t . unwrap ( ) ; table . create_column ( field . name ( ) , t . into ( ) ) . await ; } } let sequence_number = self . next_sequence_number ( ) ; self . partition_keys . insert ( SequenceNumber :: new ( sequence_number as i64 ) , partition_key . to_string ( ) , ) ; let meta = DmlMeta :: sequenced ( Sequence :: new ( self . sequencer . sequencer . id . get ( ) as u32 , sequence_number ) , self . catalog . time_provider ( ) . now ( ) , None , 0 , ) ; let op = DmlOperation :: Write ( DmlWrite :: new ( self . ns . namespace . name . clone ( ) , mutable_batches , meta , ) ) ; ( op , partition_ids ) }
Simulate what the router would do when it encounters the given set of line protocol lines.
Simulate what the router would do when it encounters the given set of line protocol lines .
&mut self
lp_lines
&[&str]
partition_key
&str
&mut self
lp_lines
&[&str]
partition_key
&str