async fn execute_statement_internal(
        &mut self,
        statement: Statement,
    ) -> QldbResult<StatementResults> {
        let statement_hash = QldbHash::from_bytes(ion_hash(&statement.partiql)).unwrap();
        self.commit_digest = self.commit_digest.dot(&statement_hash);

        let mut execution_stats = ExecutionStats::default();
        let execute_result = self
            .pooled_session
            .execute_statement(
                &self.pooled_session.session_token(),
                &self.id,
                statement.partiql.clone(),
            )
            .await?;
        execution_stats.accumulate(&execute_result);

        let mut values = vec![];
        let mut current = execute_result.first_page;
        loop {
            let page = match &current {
                Some(_) => current.take().unwrap(),
                None => break,
            };

            if let Some(holders) = page.values {
                for holder in holders {
                    let bytes = match (holder.ion_text, holder.ion_binary) {
                        (None, Some(bytes)) => bytes,
                        (Some(_txt), None) => unimplemented!(), // TextIonCursor::new(txt),
                        _ => Err(QldbError::UnexpectedResponse(
                            "expected only one of ion binary or text".to_string(),
                        ))?,
                    };
                    values.push(Bytes::from(bytes.into_inner()));
                }

                if let Some(next_page_token) = page.next_page_token {
                    let fetch_page_result = self
                        .pooled_session
                        .fetch_page(
                            &self.pooled_session.session_token(),
                            &self.id,
                            next_page_token,
                        )
                        .await?;

                    execution_stats.accumulate(&fetch_page_result);

                    if let Some(p) = fetch_page_result.page {
                        current.replace(p);
                    }
                }
            }
        }

        self.accumulated_execution_stats
            .accumulate(&execution_stats);
        Ok(StatementResults::new(values, execution_stats))
    }
// FIXME: don't buffer all results
don't buffer all results
don ' t buffer all results
async fn execute_statement_internal(
        &mut self,
        statement: Statement,
    ) -> QldbResult<StatementResults> {
        let statement_hash = QldbHash::from_bytes(ion_hash(&statement.partiql)).unwrap();
        self.commit_digest = self.commit_digest.dot(&statement_hash);
        let mut execution_stats = ExecutionStats::default();
        let execute_result = self
            .pooled_session
            .execute_statement(
                &self.pooled_session.session_token(),
                &self.id,
                statement.partiql.clone(),
            )
            .await?;
        execution_stats.accumulate(&execute_result);
        let mut values = vec![];
        let mut current = execute_result.first_page;
        loop {
            let page = match &current {
                Some(_) => current.take().unwrap(),
                None => break,
            };
            if let Some(holders) = page.values {
                for holder in holders {
                    let bytes = match (holder.ion_text, holder.ion_binary) {
                        (None, Some(bytes)) => bytes,
                        (Some(_txt), None) => unimplemented!(), // TextIonCursor::new(txt),
                        _ => Err(QldbError::UnexpectedResponse(
                            "expected only one of ion binary or text".to_string(),
                        ))?,
                    };
                    values.push(Bytes::from(bytes.into_inner()));
                }
                if let Some(next_page_token) = page.next_page_token {
                    let fetch_page_result = self
                        .pooled_session
                        .fetch_page(
                            &self.pooled_session.session_token(),
                            &self.id,
                            next_page_token,
                        )
                        .await?;
                    execution_stats.accumulate(&fetch_page_result);
                    if let Some(p) = fetch_page_result.page {
                        current.replace(p);
                    }
                }
            }
        }
        self.accumulated_execution_stats
            .accumulate(&execution_stats);
        Ok(StatementResults::new(values, execution_stats))
    }
async fn execute_statement_internal ( & mut self , statement : Statement , ) -> QldbResult < StatementResults > { let statement_hash = QldbHash :: from_bytes ( ion_hash ( & statement . partiql ) ) . unwrap ( ) ; self . commit_digest = self . commit_digest . dot ( & statement_hash ) ; let mut execution_stats = ExecutionStats :: default ( ) ; let execute_result = self . pooled_session . execute_statement ( & self . pooled_session . session_token ( ) , & self . id , statement . partiql . clone ( ) , ) . await ? ; execution_stats . accumulate ( & execute_result ) ; let mut values = vec ! [ ] ; let mut current = execute_result . first_page ; loop { let page = match & current { Some ( _ ) => current . take ( ) . unwrap ( ) , None => break , } ; if let Some ( holders ) = page . values { for holder in holders { let bytes = match ( holder . ion_text , holder . ion_binary ) { ( None , Some ( bytes ) ) => bytes , ( Some ( _txt ) , None ) => unimplemented ! ( ) , // TextIonCursor::new(txt), _ => Err ( QldbError :: UnexpectedResponse ( " " . to_string ( ) , ) ) ? , } ; values . push ( Bytes :: from ( bytes . into_inner ( ) ) ) ; } if let Some ( next_page_token ) = page . next_page_token { let fetch_page_result = self . pooled_session . fetch_page ( & self . pooled_session . session_token ( ) , & self . id , next_page_token , ) . await ? ; execution_stats . accumulate ( & fetch_page_result ) ; if let Some ( p ) = fetch_page_result . page { current . replace ( p ) ; } } } } self . accumulated_execution_stats . accumulate ( & execution_stats ) ; Ok ( StatementResults :: new ( values , execution_stats ) ) }
FIXME: don't buffer all results
FIXME : don ' t buffer all results
&mut self
statement
Statement
&mut self
statement
Statement