/// Write given table of a given chunk to object store.
/// The writing only happen if that chunk already in read buffer
Write given table of a given chunk to object store.
The writing only happen if that chunk already in read buffer
Write given table of a given chunk to object store . The writing only happen if that chunk already in read buffer
pub async fn write_chunk_to_object_store(
        &self,
        partition_key: &str,
        table_name: &str,
        chunk_id: u32,
        tracker: &TaskRegistration,
    ) -> Result<Arc<DbChunk>> {
        // Get the chunk from the catalog
        let chunk = {
            let partition = self
                .preserved_catalog
                .state()
                .valid_partition(partition_key)
                .context(LoadingChunkToParquet {
                    partition_key,
                    table_name,
                    chunk_id,
                })?;
            let partition = partition.read();
            partition
                .chunk(table_name, chunk_id)
                .context(LoadingChunkToParquet {
                    partition_key,
                    table_name,
                    chunk_id,
                })?
        };
        // update the catalog to say we are processing this chunk and
        // then drop the lock while we do the work
        let (rb_chunk, table_summary) = {
            let mut chunk = chunk.write();
            let rb_chunk =
                chunk
                    .set_writing_to_object_store(tracker)
                    .context(LoadingChunkToParquet {
                        partition_key,
                        table_name,
                        chunk_id,
                    })?;
            (rb_chunk, chunk.table_summary())
        };
        debug!(%partition_key, %table_name, %chunk_id, "chunk marked WRITING , loading tables into object store");
        // Create a storage to save data of this chunk
        let storage = Storage::new(
            Arc::clone(&self.store),
            self.server_id,
            self.rules.read().name.to_string(),
        );
        let table_name = table_summary.name.as_str();
        debug!(%partition_key, %table_name, %chunk_id, table=table_name, "loading table to object store");
        let predicate = read_buffer::Predicate::default();
        // Get RecordBatchStream of data from the read buffer chunk
        let read_results = rb_chunk
            .read_filter(table_name, predicate, Selection::All)
            .context(ReadBufferChunkError {
                table_name,
                chunk_id,
            })?;
        let arrow_schema: ArrowSchemaRef = rb_chunk
            .read_filter_table_schema(table_name, Selection::All)
            .context(ReadBufferChunkSchemaError {
                table_name,
                chunk_id,
            })?
            .into();
        let stream: SendableRecordBatchStream = Box::pin(streams::ReadFilterResultsStream::new(
            read_results,
            Arc::clone(&arrow_schema),
        ));
        // catalog-level transaction for preseveration layer
        {
            let mut transaction = self.preserved_catalog.open_transaction().await;
            // Write this table data into the object store
            let metadata = IoxMetadata {
                transaction_revision_counter: transaction.revision_counter(),
                transaction_uuid: transaction.uuid(),
            };
            let (path, parquet_metadata) = storage
                .write_to_object_store(
                    partition_key.to_string(),
                    chunk_id,
                    table_name.to_string(),
                    stream,
                    metadata,
                )
                .await
                .context(WritingToObjectStore)?;
            transaction
                .add_parquet(&path.into(), &parquet_metadata)
                .context(TransactionError)?;
            transaction.commit().await.context(TransactionError)?;
        }
        // We know this chunk is ParquetFile type
        let chunk = chunk.read();
        Ok(DbChunk::parquet_file_snapshot(&chunk))
    }
pub async fn write_chunk_to_object_store ( & self , partition_key : & str , table_name : & str , chunk_id : u32 , tracker : & TaskRegistration , ) -> Result < Arc < DbChunk > > { // Get the chunk from the catalog let chunk = { let partition = self . preserved_catalog . state ( ) . valid_partition ( partition_key ) . context ( LoadingChunkToParquet { partition_key , table_name , chunk_id , } ) ? ; let partition = partition . read ( ) ; partition . chunk ( table_name , chunk_id ) . context ( LoadingChunkToParquet { partition_key , table_name , chunk_id , } ) ? } ; // update the catalog to say we are processing this chunk and // then drop the lock while we do the work let ( rb_chunk , table_summary ) = { let mut chunk = chunk . write ( ) ; let rb_chunk = chunk . set_writing_to_object_store ( tracker ) . context ( LoadingChunkToParquet { partition_key , table_name , chunk_id , } ) ? ; ( rb_chunk , chunk . table_summary ( ) ) } ; debug ! ( partition_key table_name chunk_id " " ) ; // Create a storage to save data of this chunk let storage = Storage :: new ( Arc :: clone ( & self . store ) , self . server_id , self . rules . read ( ) . name . to_string ( ) , ) ; let table_name = table_summary . name . as_str ( ) ; debug ! ( partition_key table_name chunk_id table table_name " " ) ; let predicate = read_buffer :: Predicate :: default ( ) ; // Get RecordBatchStream of data from the read buffer chunk let read_results = rb_chunk . read_filter ( table_name , predicate , Selection :: All ) . context ( ReadBufferChunkError { table_name , chunk_id , } ) ? ; let arrow_schema : ArrowSchemaRef = rb_chunk . read_filter_table_schema ( table_name , Selection :: All ) . context ( ReadBufferChunkSchemaError { table_name , chunk_id , } ) ? . into ( ) ; let stream : SendableRecordBatchStream = Box :: pin ( streams :: ReadFilterResultsStream :: new ( read_results , Arc :: clone ( & arrow_schema ) , ) ) ; // catalog-level transaction for preseveration layer { let mut transaction = self . preserved_catalog . open_transaction ( ) . await ; // Write this table data into the object store let metadata = IoxMetadata { transaction_revision_counter : transaction . revision_counter ( ) , transaction_uuid : transaction . uuid ( ) , } ; let ( path , parquet_metadata ) = storage . write_to_object_store ( partition_key . to_string ( ) , chunk_id , table_name . to_string ( ) , stream , metadata , ) . await . context ( WritingToObjectStore ) ? ; transaction . add_parquet ( & path . into ( ) , & parquet_metadata ) . context ( TransactionError ) ? ; transaction . commit ( ) . await . context ( TransactionError ) ? ; } // We know this chunk is ParquetFile type let chunk = chunk . read ( ) ; Ok ( DbChunk :: parquet_file_snapshot ( & chunk ) ) }
Write given table of a given chunk to object store.
Write given table of a given chunk to object store .
&self
partition_key
&str
table_name
&str
chunk_id
u32
tracker
&TaskRegistration