fn evaluate(&self, ctx: &ExecutionContext, record: &RecordBatch) -> Result<ArrayRef, Error> {
        let source_schema = self.query.schema(Arc::new(SchemaContextWithSchema {
            previous: ctx.variable_context.clone(),
            schema: record.schema(),
        }))?;
        let output_type = source_schema.field(0).data_type().clone();

        let single_value_byte_size: usize = match output_type {
            DataType::Int64 => 8,
            _ => unimplemented!(),
        };

        let builder = ArrayDataBuilder::new(output_type);
        let mut null_bitmap_builder = BooleanBufferBuilder::new(record.num_rows() * single_value_byte_size);
        let mut buffer = MutableBuffer::new(record.num_rows() * single_value_byte_size);

        for i in 0..record.num_rows() {
            let mut row = Vec::with_capacity(record.num_columns());
            for _i in 0..record.num_columns() {
                row.push(ScalarValue::Null);
            }

            create_row(record.columns(), i, &mut row)?;

            let ctx = ExecutionContext {
                variable_context: Arc::new(VariableContext {
                    previous: Some(ctx.variable_context.clone()),
                    schema: record.schema().clone(),
                    variables: row,
                })
            };

            let mut batches = vec![];

            self.query.run(
                &ctx,
                &mut |_produce_ctx, batch| {
                    batches.push(batch);
                    Ok(())
                },
                &mut noop_meta_send,
            )?;

            if batches.len() == 0 || batches[0].num_rows() == 0 {
                null_bitmap_builder.append(false)?;
                buffer.resize(buffer.len() + single_value_byte_size)?;
                continue;
            }

            null_bitmap_builder.append(true)?;

            if batches.len() != 1 {
                unimplemented!()
            }

            if batches[0].num_rows() != 1 {
                unimplemented!()
            }

            let cur_data = batches[0].column(0).data();
            let cur_buffer = &cur_data.buffers()[0];

            if cur_buffer.len() != single_value_byte_size {
                unimplemented!();
            }

            buffer.reserve(buffer.len() + single_value_byte_size).unwrap();
            buffer.write_bytes(cur_buffer.data(), 0).unwrap();
        }

        let builder = builder.add_buffer(buffer.freeze());
        let builder = builder.null_bit_buffer(null_bitmap_builder.finish());
        let builder = builder.len(record.num_rows());

        let output_array = make_array(builder.build());

        Ok(output_array)
    }
// TODO: Would probably be more elegant to gather vectors of record batches, and then do a type switch later, creating the final array in a typesafe way.
Would probably be more elegant to gather vectors of record batches, and then do a type switch later, creating the final array in a typesafe way.
Would probably be more elegant to gather vectors of record batches and then do a type switch later creating the final array in a typesafe way .
fn evaluate(&self, ctx: &ExecutionContext, record: &RecordBatch) -> Result<ArrayRef, Error> {
        let source_schema = self.query.schema(Arc::new(SchemaContextWithSchema {
            previous: ctx.variable_context.clone(),
            schema: record.schema(),
        }))?;
        let output_type = source_schema.field(0).data_type().clone();
        let single_value_byte_size: usize = match output_type {
            DataType::Int64 => 8,
            _ => unimplemented!(),
        };
        let builder = ArrayDataBuilder::new(output_type);
        let mut null_bitmap_builder = BooleanBufferBuilder::new(record.num_rows() * single_value_byte_size);
        let mut buffer = MutableBuffer::new(record.num_rows() * single_value_byte_size);
        for i in 0..record.num_rows() {
            let mut row = Vec::with_capacity(record.num_columns());
            for _i in 0..record.num_columns() {
                row.push(ScalarValue::Null);
            }
            create_row(record.columns(), i, &mut row)?;
            let ctx = ExecutionContext {
                variable_context: Arc::new(VariableContext {
                    previous: Some(ctx.variable_context.clone()),
                    schema: record.schema().clone(),
                    variables: row,
                })
            };
            let mut batches = vec![];
            self.query.run(
                &ctx,
                &mut |_produce_ctx, batch| {
                    batches.push(batch);
                    Ok(())
                },
                &mut noop_meta_send,
            )?;
            if batches.len() == 0 || batches[0].num_rows() == 0 {
                null_bitmap_builder.append(false)?;
                buffer.resize(buffer.len() + single_value_byte_size)?;
                continue;
            }
            null_bitmap_builder.append(true)?;
            if batches.len() != 1 {
                unimplemented!()
            }
            if batches[0].num_rows() != 1 {
                unimplemented!()
            }
            let cur_data = batches[0].column(0).data();
            let cur_buffer = &cur_data.buffers()[0];
            if cur_buffer.len() != single_value_byte_size {
                unimplemented!();
            }
            buffer.reserve(buffer.len() + single_value_byte_size).unwrap();
            buffer.write_bytes(cur_buffer.data(), 0).unwrap();
        }
        let builder = builder.add_buffer(buffer.freeze());
        let builder = builder.null_bit_buffer(null_bitmap_builder.finish());
        let builder = builder.len(record.num_rows());
        let output_array = make_array(builder.build());
        Ok(output_array)
    }
fn evaluate ( & self , ctx : & ExecutionContext , record : & RecordBatch ) -> Result < ArrayRef , Error > { let source_schema = self . query . schema ( Arc :: new ( SchemaContextWithSchema { previous : ctx . variable_context . clone ( ) , schema : record . schema ( ) , } ) ) ? ; let output_type = source_schema . field ( 0 ) . data_type ( ) . clone ( ) ; let single_value_byte_size : usize = match output_type { DataType :: Int64 => 8 , _ => unimplemented ! ( ) , } ; let builder = ArrayDataBuilder :: new ( output_type ) ; let mut null_bitmap_builder = BooleanBufferBuilder :: new ( record . num_rows ( ) * single_value_byte_size ) ; let mut buffer = MutableBuffer :: new ( record . num_rows ( ) * single_value_byte_size ) ; for i in 0 .. record . num_rows ( ) { let mut row = Vec :: with_capacity ( record . num_columns ( ) ) ; for _i in 0 .. record . num_columns ( ) { row . push ( ScalarValue :: Null ) ; } create_row ( record . columns ( ) , i , & mut row ) ? ; let ctx = ExecutionContext { variable_context : Arc :: new ( VariableContext { previous : Some ( ctx . variable_context . clone ( ) ) , schema : record . schema ( ) . clone ( ) , variables : row , } ) } ; let mut batches = vec ! [ ] ; self . query . run ( & ctx , & mut | _produce_ctx , batch | { batches . push ( batch ) ; Ok ( ( ) ) } , & mut noop_meta_send , ) ? ; if batches . len ( ) == 0 || batches [ 0 ] . num_rows ( ) == 0 { null_bitmap_builder . append ( false ) ? ; buffer . resize ( buffer . len ( ) + single_value_byte_size ) ? ; continue ; } null_bitmap_builder . append ( true ) ? ; if batches . len ( ) != 1 { unimplemented ! ( ) } if batches [ 0 ] . num_rows ( ) != 1 { unimplemented ! ( ) } let cur_data = batches [ 0 ] . column ( 0 ) . data ( ) ; let cur_buffer = & cur_data . buffers ( ) [ 0 ] ; if cur_buffer . len ( ) != single_value_byte_size { unimplemented ! ( ) ; } buffer . reserve ( buffer . len ( ) + single_value_byte_size ) . unwrap ( ) ; buffer . write_bytes ( cur_buffer . data ( ) , 0 ) . unwrap ( ) ; } let builder = builder . add_buffer ( buffer . freeze ( ) ) ; let builder = builder . null_bit_buffer ( null_bitmap_builder . finish ( ) ) ; let builder = builder . len ( record . num_rows ( ) ) ; let output_array = make_array ( builder . build ( ) ) ; Ok ( output_array ) }
TODO: Would probably be more elegant to gather vectors of record batches, and then do a type switch later, creating the final array in a typesafe way.
TODO : Would probably be more elegant to gather vectors of record batches and then do a type switch later creating the final array in a typesafe way .
&self
ctx
&ExecutionContext
record
&RecordBatch
&self
ctx
&ExecutionContext
record
&RecordBatch