async fn schema_pivot(
    mut input_stream: SendableRecordBatchStream,
    input_schema: SchemaRef,
    output_schema: SchemaRef,
    tx: mpsc::Sender<ArrowResult<RecordBatch>>,
    baseline_metrics: BaselineMetrics,
) -> ArrowResult<()> {
    let input_fields = input_schema.fields();
    let num_fields = input_fields.len();
    let mut field_indexes_with_seen_values = vec![false; num_fields];
    let mut num_fields_seen_with_values = 0;

    // use a loop so that we release the mutex once we have read each input_batch
    let mut keep_searching = true;
    while keep_searching {
        let input_batch = input_stream.next().await.transpose()?;
        let timer = baseline_metrics.elapsed_compute().timer();

        keep_searching = match input_batch {
            Some(input_batch) => {
                let num_rows = input_batch.num_rows();

                for (i, seen_value) in field_indexes_with_seen_values.iter_mut().enumerate() {
                    // only check fields we haven't seen values for
                    if !*seen_value {
                        let column = input_batch.column(i);

                        let field_has_values = !column.is_empty() && column.null_count() < num_rows;

                        if field_has_values {
                            *seen_value = true;
                            num_fields_seen_with_values += 1;
                        }
                    }
                }
                // need to keep searching if there are still some
                // fields without values
                num_fields_seen_with_values < num_fields
            }
            // no more input
            None => false,
        };
        timer.done();
    }

    // now, output a string for each column in the input schema
    // that we saw values for
    let column_names: StringArray = field_indexes_with_seen_values
        .iter()
        .enumerate()
        .filter_map(|(field_index, has_values)| {
            if *has_values {
                Some(input_fields[field_index].name())
            } else {
                None
            }
        })
        .map(Some)
        .collect();

    let batch = RecordBatch::try_new(output_schema, vec![Arc::new(column_names)])
        .record_output(&baseline_metrics)?;

    // and send the result back
    tx.send(Ok(batch))
        .await
        .map_err(|e| ArrowError::from_external_error(Box::new(e)))?;
    Ok(())
}
// Algorithm: for each column we haven't seen a value for yet,
// check each input row;
//
// Performance Optimizations: Don't continue scaning columns
// if we have already seen a non-null value, and stop early we
// have seen values for all columns.
for each column we haven't seen a value for yet,
check each input row.

Performance Optimizations: Don't continue scaning columns
if we have already seen a non-null value, and stop early we
have seen values for all columns.
for each column we haven ' t seen a value for yet check each input row . Performance Optimizations : Don ' t continue scaning columns if we have already seen a non - null value and stop early we have seen values for all columns .
async fn schema_pivot(
    mut input_stream: SendableRecordBatchStream,
    input_schema: SchemaRef,
    output_schema: SchemaRef,
    tx: mpsc::Sender<ArrowResult<RecordBatch>>,
    baseline_metrics: BaselineMetrics,
) -> ArrowResult<()> {
    let input_fields = input_schema.fields();
    let num_fields = input_fields.len();
    let mut field_indexes_with_seen_values = vec![false; num_fields];
    let mut num_fields_seen_with_values = 0;
    // use a loop so that we release the mutex once we have read each input_batch
    let mut keep_searching = true;
    while keep_searching {
        let input_batch = input_stream.next().await.transpose()?;
        let timer = baseline_metrics.elapsed_compute().timer();
        keep_searching = match input_batch {
            Some(input_batch) => {
                let num_rows = input_batch.num_rows();
                for (i, seen_value) in field_indexes_with_seen_values.iter_mut().enumerate() {
                    // only check fields we haven't seen values for
                    if !*seen_value {
                        let column = input_batch.column(i);
                        let field_has_values = !column.is_empty() && column.null_count() < num_rows;
                        if field_has_values {
                            *seen_value = true;
                            num_fields_seen_with_values += 1;
                        }
                    }
                }
                // need to keep searching if there are still some
                // fields without values
                num_fields_seen_with_values < num_fields
            }
            // no more input
            None => false,
        };
        timer.done();
    }
    // now, output a string for each column in the input schema
    // that we saw values for
    let column_names: StringArray = field_indexes_with_seen_values
        .iter()
        .enumerate()
        .filter_map(|(field_index, has_values)| {
            if *has_values {
                Some(input_fields[field_index].name())
            } else {
                None
            }
        })
        .map(Some)
        .collect();
    let batch = RecordBatch::try_new(output_schema, vec![Arc::new(column_names)])
        .record_output(&baseline_metrics)?;
    // and send the result back
    tx.send(Ok(batch))
        .await
        .map_err(|e| ArrowError::from_external_error(Box::new(e)))?;
    Ok(())
}
async fn schema_pivot ( mut input_stream : SendableRecordBatchStream , input_schema : SchemaRef , output_schema : SchemaRef , tx : mpsc :: Sender < ArrowResult < RecordBatch > > , baseline_metrics : BaselineMetrics , ) -> ArrowResult < ( ) > { let input_fields = input_schema . fields ( ) ; let num_fields = input_fields . len ( ) ; let mut field_indexes_with_seen_values = vec ! [ false num_fields ] ; let mut num_fields_seen_with_values = 0 ; // use a loop so that we release the mutex once we have read each input_batch let mut keep_searching = true ; while keep_searching { let input_batch = input_stream . next ( ) . await . transpose ( ) ? ; let timer = baseline_metrics . elapsed_compute ( ) . timer ( ) ; keep_searching = match input_batch { Some ( input_batch ) => { let num_rows = input_batch . num_rows ( ) ; for ( i , seen_value ) in field_indexes_with_seen_values . iter_mut ( ) . enumerate ( ) { // only check fields we haven't seen values for if ! * seen_value { let column = input_batch . column ( i ) ; let field_has_values = ! column . is_empty ( ) && column . null_count ( ) < num_rows ; if field_has_values { * seen_value = true ; num_fields_seen_with_values += 1 ; } } } // need to keep searching if there are still some // fields without values num_fields_seen_with_values < num_fields } // no more input None => false , } ; timer . done ( ) ; } // now, output a string for each column in the input schema // that we saw values for let column_names : StringArray = field_indexes_with_seen_values . iter ( ) . enumerate ( ) . filter_map ( | ( field_index , has_values ) | { if * has_values { Some ( input_fields [ field_index ] . name ( ) ) } else { None } } ) . map ( Some ) . collect ( ) ; let batch = RecordBatch :: try_new ( output_schema , vec ! [ Arc new ( column_names ) ] ) . record_output ( & baseline_metrics ) ? ; // and send the result back tx . send ( Ok ( batch ) ) . await . map_err ( | e | ArrowError :: from_external_error ( Box :: new ( e ) ) ) ? ; Ok ( ( ) ) }
Algorithm: for each column we haven't seen a value for yet,
check each input row;
Algorithm : for each column we haven ' t seen a value for yet check each input row ;
input_stream
SendableRecordBatchStream
input_schema
SchemaRef
output_schema
SchemaRef
tx
mpsc::Sender<ArrowResult<RecordBatch>>
baseline_metrics
BaselineMetrics
input_stream
SendableRecordBatchStream
input_schema
SchemaRef
output_schema
SchemaRef
tx
mpsc::Sender<ArrowResult<RecordBatch>>
baseline_metrics
BaselineMetrics