Struct arrow_odbc::OdbcWriter

pub struct OdbcWriter<S> { /* private fields */ }

Inserts batches from an [arrow::record_batch::RecordBatchReader] into a database.

Implementations

impl<S> OdbcWriter<S>where
    S: AsStatementRef,

pub fn new(
    row_capacity: usize,
    schema: &Schema,
    statement: Prepared<S>
) -> Result<Self, WriterError>

Construct a new ODBC writer using an alredy existing prepared statement. Usually you want to call a higher level constructor like Self::with_connection. Yet, this constructor is useful in two scenarios.

1. The prepared statement is already constructed and you do not want to spend the time to prepare it again.
2. You want to use the arrow arrays as arrar parameters for a statement, but that statement is not necessarily an INSERT statement with a simple 1to1 mapping of columns between table and arrow schema.

Parameters

• row_capacity: The amount of rows send to the database in each chunk. With the exception of the last chunk, which may be smaller.
• schema: Schema needs to have one column for each positional parameter of the statement and match the data which will be supplied to the instance later. Otherwise your code will panic.
• statement: A prepared statement whose SQL text representation contains one placeholder for each column. The order of the placeholers must correspond to the orders of the columns in the schema.

Examples found in repository

src/odbc_writer.rs (line 267)

    pub fn from_connection(
        connection: Connection<'env>,
        schema: &Schema,
        table_name: &str,
        row_capacity: usize,
    ) -> Result<Self, WriterError> {
        let sql = insert_statement_from_schema(schema, table_name);
        let statement = connection
            .into_prepared(&sql)
            .map_err(|source| WriterError::PreparingInsertStatement { source, sql })?;
        Self::new(row_capacity, schema, statement)
    }
}

impl<'o> OdbcWriter<StatementImpl<'o>> {
    /// A writer which borrows the connection and inserts the given schema into a table with
    /// matching column names.
    ///
    /// **Note:**
    ///
    /// If table or column names are derived from user input, be sure to sanatize the input in order
    /// to prevent SQL injection attacks.
    pub fn with_connection(
        connection: &'o Connection<'o>,
        schema: &Schema,
        table_name: &str,
        row_capacity: usize,
    ) -> Result<Self, WriterError> {
        let sql = insert_statement_from_schema(schema, table_name);
        let statement = connection
            .prepare(&sql)
            .map_err(|source| WriterError::PreparingInsertStatement { source, sql })?;
        Self::new(row_capacity, schema, statement)
    }

pub fn write_all(
    &mut self,
    reader: impl Iterator<Item = Result<RecordBatch, ArrowError>>
) -> Result<(), WriterError>

Consumes all the batches in the record batch reader and sends them chunk by chunk to the database.

Examples found in repository

src/odbc_writer.rs (line 56)

pub fn insert_into_table(
    connection: &Connection,
    batches: &mut impl RecordBatchReader,
    table_name: &str,
    batch_size: usize,
) -> Result<(), WriterError> {
    let schema = batches.schema();
    let mut inserter =
        OdbcWriter::with_connection(connection, schema.as_ref(), table_name, batch_size)?;
    inserter.write_all(batches)
}

pub fn write_batch(
    &mut self,
    record_batch: &RecordBatch
) -> Result<(), WriterError>

Consumes a single batch and sends it chunk by chunk to the database. The last batch may not be consumed until Self::flush is called.

Examples found in repository

src/odbc_writer.rs (line 196)

    pub fn write_all(
        &mut self,
        reader: impl Iterator<Item = Result<RecordBatch, ArrowError>>,
    ) -> Result<(), WriterError> {
        for result in reader {
            let record_batch = result.map_err(WriterError::ReadingRecordBatch)?;
            self.write_batch(&record_batch)?;
        }
        self.flush()?;
        Ok(())
    }

pub fn flush(&mut self) -> Result<(), WriterError>

The number of row in an individual record batch must not necessarily match the capacity of the buffers owned by this writer. Therfore sometimes records are not send to the database immediatly but rather we wait for the buffers to be filled then reading the next batch. Once we reach the last batch however, there is no “next batch” anymore. In that case we call this method in order to send the remainder of the records to the database as well.

Examples found in repository

src/odbc_writer.rs (line 198)

    pub fn write_all(
        &mut self,
        reader: impl Iterator<Item = Result<RecordBatch, ArrowError>>,
    ) -> Result<(), WriterError> {
        for result in reader {
            let record_batch = result.map_err(WriterError::ReadingRecordBatch)?;
            self.write_batch(&record_batch)?;
        }
        self.flush()?;
        Ok(())
    }

    /// Consumes a single batch and sends it chunk by chunk to the database. The last batch may not
    /// be consumed until [`Self::flush`] is called.
    pub fn write_batch(&mut self, record_batch: &RecordBatch) -> Result<(), WriterError> {
        let capacity = self.inserter.capacity();
        let mut remanining_rows = record_batch.num_rows();
        // The record batch may contain more rows than the capacity of our writer can hold. So we
        // need to be able to fill the buffers multiple times and send them to the database in
        // between.
        while remanining_rows != 0 {
            let chunk_size = min(capacity - self.inserter.num_rows(), remanining_rows);
            let param_offset = self.inserter.num_rows();
            self.inserter.set_num_rows(param_offset + chunk_size);
            let chunk = record_batch.slice(record_batch.num_rows() - remanining_rows, chunk_size);
            for (index, (array, strategy)) in chunk
                .columns()
                .iter()
                .zip(self.strategies.iter())
                .enumerate()
            {
                strategy.write_rows(param_offset, self.inserter.column_mut(index), array)?
            }

            // If we used up all capacity we send the parameters to the database and reset the
            // parameter buffers.
            if self.inserter.num_rows() == capacity {
                self.flush()?;
            }
            remanining_rows -= chunk_size;
        }

        Ok(())
    }

impl<'env> OdbcWriter<StatementConnection<'env>>

pub fn from_connection(
    connection: Connection<'env>,
    schema: &Schema,
    table_name: &str,
    row_capacity: usize
) -> Result<Self, WriterError>

A writer which takes ownership of the connection and inserts the given schema into a table with matching column names.

Note:

If table or column names are derived from user input, be sure to sanatize the input in order to prevent SQL injection attacks.

impl<'o> OdbcWriter<StatementImpl<'o>>

pub fn with_connection(
    connection: &'o Connection<'o>,
    schema: &Schema,
    table_name: &str,
    row_capacity: usize
) -> Result<Self, WriterError>

A writer which borrows the connection and inserts the given schema into a table with matching column names.

Note:

If table or column names are derived from user input, be sure to sanatize the input in order to prevent SQL injection attacks.

Examples found in repository

src/odbc_writer.rs (line 55)

pub fn insert_into_table(
    connection: &Connection,
    batches: &mut impl RecordBatchReader,
    table_name: &str,
    batch_size: usize,
) -> Result<(), WriterError> {
    let schema = batches.schema();
    let mut inserter =
        OdbcWriter::with_connection(connection, schema.as_ref(), table_name, batch_size)?;
    inserter.write_all(batches)
}