Enum ExaImport

pub enum ExaImport {
    Setup(BoxFuture<'static, Result<ExaSocket>>, usize, bool),
    Writing(ExaImportWriter),
}

An ETL IMPORT worker.

The type implements AsyncWrite and is Send and Sync so it can be (almost) freely used for any data pipeline.

The only caveat is that you MUST call futures_util::AsyncWriteExt::close on each worker to finalize the import. Otherwise, Exasol keeps on expecting data.

Atomicity

IMPORT jobs are not atomic by themselves. If an error occurs during the data ingestion, some of the data might be already sent and written in the database. However, since IMPORT is fundamentally just a query, it can be transactional. Therefore, beginning a transaction and passing that to the ImportBuilder::build method will result in the import job needing to be explicitly committed:

use std::env;

use sqlx_exasol::{etl::*, *};

let pool = ExaPool::connect(&env::var("DATABASE_URL").unwrap()).await?;
let mut con = pool.acquire().await?;
let mut tx = con.begin().await?;

let (query_fut, writers) = ImportBuilder::new("SOME_TABLE").build(&mut *tx).await?;

// concurrently use the writers and await the query future

tx.commit().await?;

IMPORTANT

In multi-node environments closing a writer without writing any data to it can, and most likely will, cause issues; Exasol does not immediately start reading data from all workers but rather seems to connect to them sequentially after each of them provides some data.

From what I could gather from the logs, providing no data (although the request is responded to gracefully) makes Exasol retry the connection. With these workers being implemented as one-shot HTTP servers, there’s nothing to connect to anymore. Even if it were, the connection would just be re-attempted over and over since we’d still be sending no data.

Since not using one or more import workers seems to be treated as an error on Exasol’s side, it’s best not to create excess writers that you don’t plan on using to avoid such issues.

See https://github.com/exasol/websocket-api/issues/33 for more details.

Variants

Setup(BoxFuture<'static, Result<ExaSocket>>, usize, bool)

👎Deprecated: will be made private

Setup state of the worker. This typically means waiting on the TLS handshake.

This approach is needed because Exasol will issue connections sequentially and thus perform TLS handshakes the same way.

Therefore we accommodate the worker state until the query gets executed and data gets sent through the workers, which happens within consumer code.

Writing(ExaImportWriter)

👎Deprecated: will be made private

The worker is fully connected and ready for I/O.

Trait Implementations

impl AsyncWrite for ExaImport

fn poll_write( self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8], ) -> Poll<Result<usize>>

Attempt to write bytes from buf into the object.

fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>>

Attempt to flush the object, ensuring that any buffered data reach their destination.

fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>>

Attempt to close the object.

fn poll_write_vectored( self: Pin<&mut Self>, cx: &mut Context<'_>, bufs: &[IoSlice<'_>], ) -> Poll<Result<usize, Error>>

Attempt to write bytes from bufs into the object using vectored IO operations.

impl Debug for ExaImport

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.