async fn produce_snapshot<W: AsyncWrite + Unpin>(
        &mut self,
        snapshot_tx: &mut SourceTransaction<'_>,
        buffer: &mut W,
    ) -> Result<(), ReplicationError> {
        let client =
            try_recoverable!(postgres_util::connect_replication(&self.connector.conn).await);

        // We're initialising this source so any previously existing slot must be removed and
        // re-created. Once we have data persistence we will be able to reuse slots across restarts
        let _ = client
            .simple_query(&format!(
                "DROP_REPLICATION_SLOT {:?}",
                &self.connector.slot_name
            ))
            .await;

        // Get all the relevant tables for this publication
        let publication_tables = try_recoverable!(
            postgres_util::publication_info(&self.connector.conn, &self.connector.publication)
                .await
        );

        // Start a transaction and immediatelly create a replication slot with the USE SNAPSHOT
        // directive. This makes the starting point of the slot and the snapshot of the transaction
        // identical.
        client
            .simple_query("BEGIN READ ONLY ISOLATION LEVEL REPEATABLE READ;")
            .await?;

        let slot_query = format!(
            r#"CREATE_REPLICATION_SLOT {:?} LOGICAL "pgoutput" USE_SNAPSHOT"#,
            &self.connector.slot_name
        );
        let slot_row = client
            .simple_query(&slot_query)
            .await?
            .into_iter()
            .next()
            .and_then(|msg| match msg {
                SimpleQueryMessage::Row(row) => Some(row),
                _ => None,
            })
            .ok_or_else(|| {
                ReplicationError::Recoverable(anyhow!(
                    "empty result after creating replication slot"
                ))
            })?;

        // Store the lsn at which we will need to start the replication stream from
        let consistent_point = try_recoverable!(slot_row
            .get("consistent_point")
            .ok_or_else(|| anyhow!("missing expected column: `consistent_point`")));
        self.lsn = try_fatal!(consistent_point
            .parse()
            .or_else(|_| Err(anyhow!("invalid lsn"))));

        for info in publication_tables {
            // TODO(petrosagg): use a COPY statement here for more efficient network transfer
            let data = client
                .simple_query(&format!(
                    "SELECT * FROM {:?}.{:?}",
                    info.namespace, info.name
                ))
                .await?;
            for msg in data {
                if let SimpleQueryMessage::Row(row) = msg {
                    let mut mz_row = Row::default();
                    let rel_id: Datum = (info.rel_id as i32).into();
                    mz_row.push(rel_id);
                    mz_row.push_list((0..row.len()).map(|n| {
                        let a: Datum = row.get(n).into();
                        a
                    }));
                    try_recoverable!(snapshot_tx.insert(mz_row.clone()).await);
                    try_fatal!(buffer.write(&try_fatal!(bincode::serialize(&mz_row))).await);
                }
            }
        }
        client.simple_query("COMMIT;").await?;
        Ok(())
    }
/// Creates the replication slot and produces the initial snapshot of the data
///
/// After the initial snapshot has been produced it returns the name of the created slot and
/// the LSN at which we should start the replication stream at.
Creates the replication slot and produces the initial snapshot of the data
After the initial snapshot has been produced it returns the name of the created slot and
the LSN at which we should start the replication stream at.
Creates the replication slot and produces the initial snapshot of the data After the initial snapshot has been produced it returns the name of the created slot and the LSN at which we should start the replication stream at .
async fn produce_snapshot<W: AsyncWrite + Unpin>(
        &mut self,
        snapshot_tx: &mut SourceTransaction<'_>,
        buffer: &mut W,
    ) -> Result<(), ReplicationError> {
        let client =
            try_recoverable!(postgres_util::connect_replication(&self.connector.conn).await);
        // We're initialising this source so any previously existing slot must be removed and
        // re-created. Once we have data persistence we will be able to reuse slots across restarts
        let _ = client
            .simple_query(&format!(
                "DROP_REPLICATION_SLOT {:?}",
                &self.connector.slot_name
            ))
            .await;
        // Get all the relevant tables for this publication
        let publication_tables = try_recoverable!(
            postgres_util::publication_info(&self.connector.conn, &self.connector.publication)
                .await
        );
        // Start a transaction and immediatelly create a replication slot with the USE SNAPSHOT
        // directive. This makes the starting point of the slot and the snapshot of the transaction
        // identical.
        client
            .simple_query("BEGIN READ ONLY ISOLATION LEVEL REPEATABLE READ;")
            .await?;
        let slot_query = format!(
            r#"CREATE_REPLICATION_SLOT {:?} LOGICAL "pgoutput" USE_SNAPSHOT"#,
            &self.connector.slot_name
        );
        let slot_row = client
            .simple_query(&slot_query)
            .await?
            .into_iter()
            .next()
            .and_then(|msg| match msg {
                SimpleQueryMessage::Row(row) => Some(row),
                _ => None,
            })
            .ok_or_else(|| {
                ReplicationError::Recoverable(anyhow!(
                    "empty result after creating replication slot"
                ))
            })?;
        // Store the lsn at which we will need to start the replication stream from
        let consistent_point = try_recoverable!(slot_row
            .get("consistent_point")
            .ok_or_else(|| anyhow!("missing expected column: `consistent_point`")));
        self.lsn = try_fatal!(consistent_point
            .parse()
            .or_else(|_| Err(anyhow!("invalid lsn"))));
        for info in publication_tables {
            // TODO(petrosagg): use a COPY statement here for more efficient network transfer
            let data = client
                .simple_query(&format!(
                    "SELECT * FROM {:?}.{:?}",
                    info.namespace, info.name
                ))
                .await?;
            for msg in data {
                if let SimpleQueryMessage::Row(row) = msg {
                    let mut mz_row = Row::default();
                    let rel_id: Datum = (info.rel_id as i32).into();
                    mz_row.push(rel_id);
                    mz_row.push_list((0..row.len()).map(|n| {
                        let a: Datum = row.get(n).into();
                        a
                    }));
                    try_recoverable!(snapshot_tx.insert(mz_row.clone()).await);
                    try_fatal!(buffer.write(&try_fatal!(bincode::serialize(&mz_row))).await);
                }
            }
        }
        client.simple_query("COMMIT;").await?;
        Ok(())
    }
async fn produce_snapshot < W : AsyncWrite + Unpin > ( & mut self , snapshot_tx : & mut SourceTransaction < ' _ > , buffer : & mut W , ) -> Result < ( ) , ReplicationError > { let client = try_recoverable ! ( postgres_util connect_replication ( self connector conn ) await ) ; // We're initialising this source so any previously existing slot must be removed and // re-created. Once we have data persistence we will be able to reuse slots across restarts let _ = client . simple_query ( & format ! ( " " self connector slot_name ) ) . await ; // Get all the relevant tables for this publication let publication_tables = try_recoverable ! ( postgres_util publication_info ( self connector conn self connector publication ) await ) ; // Start a transaction and immediatelly create a replication slot with the USE SNAPSHOT // directive. This makes the starting point of the slot and the snapshot of the transaction // identical. client . simple_query ( " " ) . await ? ; let slot_query = format ! ( r#"CREATE_REPLICATION_SLOT {:?} LOGICAL "pgoutput" USE_SNAPSHOT"# self connector slot_name ) ; let slot_row = client . simple_query ( & slot_query ) . await ? . into_iter ( ) . next ( ) . and_then ( | msg | match msg { SimpleQueryMessage :: Row ( row ) => Some ( row ) , _ => None , } ) . ok_or_else ( | | { ReplicationError :: Recoverable ( anyhow ! ( " " ) ) } ) ? ; // Store the lsn at which we will need to start the replication stream from let consistent_point = try_recoverable ! ( slot_row get ( " " ) ok_or_else ( anyhow ( " " ) ) ) ; self . lsn = try_fatal ! ( consistent_point parse ( ) or_else ( Err ( anyhow ( " " ) ) ) ) ; for info in publication_tables { // TODO(petrosagg): use a COPY statement here for more efficient network transfer let data = client . simple_query ( & format ! ( " " info namespace info name ) ) . await ? ; for msg in data { if let SimpleQueryMessage :: Row ( row ) = msg { let mut mz_row = Row :: default ( ) ; let rel_id : Datum = ( info . rel_id as i32 ) . into ( ) ; mz_row . push ( rel_id ) ; mz_row . push_list ( ( 0 .. row . len ( ) ) . map ( | n | { let a : Datum = row . get ( n ) . into ( ) ; a } ) ) ; try_recoverable ! ( snapshot_tx insert ( mz_row clone ( ) ) await ) ; try_fatal ! ( buffer write ( try_fatal ( bincode serialize ( mz_row ) ) ) await ) ; } } } client . simple_query ( " " ) . await ? ; Ok ( ( ) ) }
Creates the replication slot and produces the initial snapshot of the data
After the initial snapshot has been produced it returns the name of the created slot and
the LSN at which we should start the replication stream at.
Creates the replication slot and produces the initial snapshot of the data After the initial snapshot has been produced it returns the name of the created slot and the LSN at which we should start the replication stream at .
&mut self
snapshot_tx
&mut SourceTransaction<'_>
buffer
&mut W
&mut self
snapshot_tx
&mut SourceTransaction<'_>
buffer
&mut W