pub(super) fn write_chunk_to_object_store(
    partition: LifecycleWriteGuard<'_, Partition, LockableCatalogPartition>,
    mut chunk: LifecycleWriteGuard<'_, CatalogChunk, LockableCatalogChunk>,
    flush_handle: FlushHandle,
) -> Result<(
    TaskTracker<Job>,
    TrackedFuture<impl Future<Output = Result<Arc<DbChunk>>> + Send>,
)> {
    let db = Arc::clone(&chunk.data().db);
    let addr = chunk.addr().clone();
    let table_name = Arc::clone(&addr.table_name);
    let partition_key = Arc::clone(&addr.partition_key);

    let (tracker, registration) = db.jobs.register(Job::WriteChunk {
        chunk: addr.clone(),
    });

    // update the catalog to say we are processing this chunk and
    chunk.set_writing_to_object_store(&registration)?;
    let db_chunk = DbChunk::snapshot(&*chunk);

    let time_of_first_write = db_chunk.time_of_first_write();
    let time_of_last_write = db_chunk.time_of_last_write();

    debug!(chunk=%chunk.addr(), "chunk marked WRITING , loading tables into object store");

    // Drop locks
    let chunk = chunk.into_data().chunk;
    let partition = partition.into_data().partition;

    // Create a storage to save data of this chunk
    let storage = Storage::new(Arc::clone(&db.store), db.server_id);

    let catalog_transactions_until_checkpoint = db
        .rules
        .read()
        .lifecycle_rules
        .catalog_transactions_until_checkpoint
        .get();

    let fut = async move {
        debug!(chunk=%addr, "loading table to object store");

        let (partition_checkpoint, database_checkpoint) =
            collect_checkpoints(flush_handle.checkpoint(), &db.catalog);

        // Get RecordBatchStream of data from the read buffer chunk
        let stream = db_chunk
            .read_filter(&Default::default(), Selection::All)
            .expect("read filter should be infallible");

        // check that the upcoming state change will very likely succeed
        {
            // re-lock
            let guard = chunk.read();
            if matches!(guard.stage(), &ChunkStage::Persisted { .. })
                || !guard.is_in_lifecycle(ChunkLifecycleAction::Persisting)
            {
                return Err(Error::CannotWriteChunk {
                    addr: guard.addr().clone(),
                });
            }
        }

        // catalog-level transaction for preservation layer
        {
            // fetch shared (= read) guard preventing the cleanup job from deleting our files
            let _guard = db.cleanup_lock.read().await;

            // Write this table data into the object store
            //
            // IMPORTANT: Writing must take place while holding the cleanup lock, otherwise the file might be deleted
            //            between creation and the transaction commit.
            let metadata = IoxMetadata {
                creation_timestamp: Utc::now(),
                table_name: Arc::clone(&table_name),
                partition_key: Arc::clone(&partition_key),
                chunk_id: addr.chunk_id,
                partition_checkpoint,
                database_checkpoint,
                time_of_first_write,
                time_of_last_write,
            };
            let (path, file_size_bytes, parquet_metadata) = storage
                .write_to_object_store(addr, stream, metadata)
                .await
                .context(WritingToObjectStore)?;
            let parquet_metadata = Arc::new(parquet_metadata);

            let metrics = db
                .catalog
                .metrics_registry
                .register_domain_with_labels("parquet", db.catalog.metric_labels.clone());
            let metrics = ParquetChunkMetrics::new(&metrics);
            let parquet_chunk = Arc::new(
                ParquetChunk::new(
                    path.clone(),
                    Arc::clone(&db.store),
                    file_size_bytes,
                    Arc::clone(&parquet_metadata),
                    Arc::clone(&table_name),
                    Arc::clone(&partition_key),
                    metrics,
                )
                .context(ParquetChunkError)?,
            );

            let path: DirsAndFileName = path.into();

            // IMPORTANT: Start transaction AFTER writing the actual parquet file so we do not hold the
            //            transaction lock (that is part of the PreservedCatalog) for too long. By using the
            //            cleanup lock (see above) it is ensured that the file that we have written is not deleted
            //            in between.
            let mut transaction = db.preserved_catalog.open_transaction().await;
            let info = CatalogParquetInfo {
                path,
                file_size_bytes,
                metadata: parquet_metadata,
            };
            transaction.add_parquet(&info).context(TransactionError)?;

            // preserved commit
            let ckpt_handle = transaction.commit().await.context(CommitError)?;

            // in-mem commit
            {
                let mut guard = chunk.write();
                if let Err(e) = guard.set_written_to_object_store(parquet_chunk) {
                    panic!("Chunk written but cannot mark as written {}", e);
                }
            }

            let create_checkpoint =
                ckpt_handle.revision_counter() % catalog_transactions_until_checkpoint == 0;
            if create_checkpoint {
                // Commit is already done, so we can just scan the catalog for the state.
                //
                // NOTE: There can only be a single transaction in this section because the checkpoint handle holds
                //       transaction lock. Therefore we don't need to worry about concurrent modifications of
                //       preserved chunks.
                if let Err(e) = ckpt_handle
                    .create_checkpoint(checkpoint_data_from_catalog(&db.catalog))
                    .await
                {
                    warn!(%e, "cannot create catalog checkpoint");

                    // That's somewhat OK. Don't fail the entire task, because the actual preservation was completed
                    // (both in-mem and within the preserved catalog).
                }
            }
        }

        {
            // Flush persisted data from persistence windows
            let mut partition = partition.write();
            partition
                .persistence_windows_mut()
                .expect("persistence windows removed")
                .flush(flush_handle);
        }

        // We know this chunk is ParquetFile type
        let chunk = chunk.read();
        Ok(DbChunk::parquet_file_snapshot(&chunk))
    };

    Ok((tracker, fut.track(registration)))
}
/// The implementation for writing a chunk to the object store
///
/// Returns a future registered with the tracker registry, and the corresponding tracker
/// The caller can either spawn this future to tokio, or block directly on it
The implementation for writing a chunk to the object store
Returns a future registered with the tracker registry, and the corresponding tracker
The caller can either spawn this future to tokio, or block directly on it
The implementation for writing a chunk to the object store Returns a future registered with the tracker registry and the corresponding tracker The caller can either spawn this future to tokio or block directly on it
pub(super) fn write_chunk_to_object_store(
    partition: LifecycleWriteGuard<'_, Partition, LockableCatalogPartition>,
    mut chunk: LifecycleWriteGuard<'_, CatalogChunk, LockableCatalogChunk>,
    flush_handle: FlushHandle,
) -> Result<(
    TaskTracker<Job>,
    TrackedFuture<impl Future<Output = Result<Arc<DbChunk>>> + Send>,
)> {
    let db = Arc::clone(&chunk.data().db);
    let addr = chunk.addr().clone();
    let table_name = Arc::clone(&addr.table_name);
    let partition_key = Arc::clone(&addr.partition_key);
    let (tracker, registration) = db.jobs.register(Job::WriteChunk {
        chunk: addr.clone(),
    });
    // update the catalog to say we are processing this chunk and
    chunk.set_writing_to_object_store(&registration)?;
    let db_chunk = DbChunk::snapshot(&*chunk);
    let time_of_first_write = db_chunk.time_of_first_write();
    let time_of_last_write = db_chunk.time_of_last_write();
    debug!(chunk=%chunk.addr(), "chunk marked WRITING , loading tables into object store");
    // Drop locks
    let chunk = chunk.into_data().chunk;
    let partition = partition.into_data().partition;
    // Create a storage to save data of this chunk
    let storage = Storage::new(Arc::clone(&db.store), db.server_id);
    let catalog_transactions_until_checkpoint = db
        .rules
        .read()
        .lifecycle_rules
        .catalog_transactions_until_checkpoint
        .get();
    let fut = async move {
        debug!(chunk=%addr, "loading table to object store");
        let (partition_checkpoint, database_checkpoint) =
            collect_checkpoints(flush_handle.checkpoint(), &db.catalog);
        // Get RecordBatchStream of data from the read buffer chunk
        let stream = db_chunk
            .read_filter(&Default::default(), Selection::All)
            .expect("read filter should be infallible");
        // check that the upcoming state change will very likely succeed
        {
            // re-lock
            let guard = chunk.read();
            if matches!(guard.stage(), &ChunkStage::Persisted { .. })
                || !guard.is_in_lifecycle(ChunkLifecycleAction::Persisting)
            {
                return Err(Error::CannotWriteChunk {
                    addr: guard.addr().clone(),
                });
            }
        }
        // catalog-level transaction for preservation layer
        {
            // fetch shared (= read) guard preventing the cleanup job from deleting our files
            let _guard = db.cleanup_lock.read().await;
            // Write this table data into the object store
            //
            // IMPORTANT: Writing must take place while holding the cleanup lock, otherwise the file might be deleted
            //            between creation and the transaction commit.
            let metadata = IoxMetadata {
                creation_timestamp: Utc::now(),
                table_name: Arc::clone(&table_name),
                partition_key: Arc::clone(&partition_key),
                chunk_id: addr.chunk_id,
                partition_checkpoint,
                database_checkpoint,
                time_of_first_write,
                time_of_last_write,
            };
            let (path, file_size_bytes, parquet_metadata) = storage
                .write_to_object_store(addr, stream, metadata)
                .await
                .context(WritingToObjectStore)?;
            let parquet_metadata = Arc::new(parquet_metadata);
            let metrics = db
                .catalog
                .metrics_registry
                .register_domain_with_labels("parquet", db.catalog.metric_labels.clone());
            let metrics = ParquetChunkMetrics::new(&metrics);
            let parquet_chunk = Arc::new(
                ParquetChunk::new(
                    path.clone(),
                    Arc::clone(&db.store),
                    file_size_bytes,
                    Arc::clone(&parquet_metadata),
                    Arc::clone(&table_name),
                    Arc::clone(&partition_key),
                    metrics,
                )
                .context(ParquetChunkError)?,
            );
            let path: DirsAndFileName = path.into();
            // IMPORTANT: Start transaction AFTER writing the actual parquet file so we do not hold the
            //            transaction lock (that is part of the PreservedCatalog) for too long. By using the
            //            cleanup lock (see above) it is ensured that the file that we have written is not deleted
            //            in between.
            let mut transaction = db.preserved_catalog.open_transaction().await;
            let info = CatalogParquetInfo {
                path,
                file_size_bytes,
                metadata: parquet_metadata,
            };
            transaction.add_parquet(&info).context(TransactionError)?;
            // preserved commit
            let ckpt_handle = transaction.commit().await.context(CommitError)?;
            // in-mem commit
            {
                let mut guard = chunk.write();
                if let Err(e) = guard.set_written_to_object_store(parquet_chunk) {
                    panic!("Chunk written but cannot mark as written {}", e);
                }
            }
            let create_checkpoint =
                ckpt_handle.revision_counter() % catalog_transactions_until_checkpoint == 0;
            if create_checkpoint {
                // Commit is already done, so we can just scan the catalog for the state.
                //
                // NOTE: There can only be a single transaction in this section because the checkpoint handle holds
                //       transaction lock. Therefore we don't need to worry about concurrent modifications of
                //       preserved chunks.
                if let Err(e) = ckpt_handle
                    .create_checkpoint(checkpoint_data_from_catalog(&db.catalog))
                    .await
                {
                    warn!(%e, "cannot create catalog checkpoint");
                    // That's somewhat OK. Don't fail the entire task, because the actual preservation was completed
                    // (both in-mem and within the preserved catalog).
                }
            }
        }
        {
            // Flush persisted data from persistence windows
            let mut partition = partition.write();
            partition
                .persistence_windows_mut()
                .expect("persistence windows removed")
                .flush(flush_handle);
        }
        // We know this chunk is ParquetFile type
        let chunk = chunk.read();
        Ok(DbChunk::parquet_file_snapshot(&chunk))
    };
    Ok((tracker, fut.track(registration)))
}
pub ( super ) fn write_chunk_to_object_store ( partition : LifecycleWriteGuard < ' _ , Partition , LockableCatalogPartition > , mut chunk : LifecycleWriteGuard < ' _ , CatalogChunk , LockableCatalogChunk > , flush_handle : FlushHandle , ) -> Result < ( TaskTracker < Job > , TrackedFuture < impl Future < Output = Result < Arc < DbChunk > > > + Send > , ) > { let db = Arc :: clone ( & chunk . data ( ) . db ) ; let addr = chunk . addr ( ) . clone ( ) ; let table_name = Arc :: clone ( & addr . table_name ) ; let partition_key = Arc :: clone ( & addr . partition_key ) ; let ( tracker , registration ) = db . jobs . register ( Job :: WriteChunk { chunk : addr . clone ( ) , } ) ; // update the catalog to say we are processing this chunk and chunk . set_writing_to_object_store ( & registration ) ? ; let db_chunk = DbChunk :: snapshot ( & * chunk ) ; let time_of_first_write = db_chunk . time_of_first_write ( ) ; let time_of_last_write = db_chunk . time_of_last_write ( ) ; debug ! ( chunk chunk addr ( ) " " ) ; // Drop locks let chunk = chunk . into_data ( ) . chunk ; let partition = partition . into_data ( ) . partition ; // Create a storage to save data of this chunk let storage = Storage :: new ( Arc :: clone ( & db . store ) , db . server_id ) ; let catalog_transactions_until_checkpoint = db . rules . read ( ) . lifecycle_rules . catalog_transactions_until_checkpoint . get ( ) ; let fut = async move { debug ! ( chunk addr " " ) ; let ( partition_checkpoint , database_checkpoint ) = collect_checkpoints ( flush_handle . checkpoint ( ) , & db . catalog ) ; // Get RecordBatchStream of data from the read buffer chunk let stream = db_chunk . read_filter ( & Default :: default ( ) , Selection :: All ) . expect ( " " ) ; // check that the upcoming state change will very likely succeed { // re-lock let guard = chunk . read ( ) ; if matches ! ( guard stage ( ) ChunkStage Persisted { } ) || ! guard . is_in_lifecycle ( ChunkLifecycleAction :: Persisting ) { return Err ( Error :: CannotWriteChunk { addr : guard . addr ( ) . clone ( ) , } ) ; } } // catalog-level transaction for preservation layer { // fetch shared (= read) guard preventing the cleanup job from deleting our files let _guard = db . cleanup_lock . read ( ) . await ; // Write this table data into the object store // // IMPORTANT: Writing must take place while holding the cleanup lock, otherwise the file might be deleted //            between creation and the transaction commit. let metadata = IoxMetadata { creation_timestamp : Utc :: now ( ) , table_name : Arc :: clone ( & table_name ) , partition_key : Arc :: clone ( & partition_key ) , chunk_id : addr . chunk_id , partition_checkpoint , database_checkpoint , time_of_first_write , time_of_last_write , } ; let ( path , file_size_bytes , parquet_metadata ) = storage . write_to_object_store ( addr , stream , metadata ) . await . context ( WritingToObjectStore ) ? ; let parquet_metadata = Arc :: new ( parquet_metadata ) ; let metrics = db . catalog . metrics_registry . register_domain_with_labels ( " " , db . catalog . metric_labels . clone ( ) ) ; let metrics = ParquetChunkMetrics :: new ( & metrics ) ; let parquet_chunk = Arc :: new ( ParquetChunk :: new ( path . clone ( ) , Arc :: clone ( & db . store ) , file_size_bytes , Arc :: clone ( & parquet_metadata ) , Arc :: clone ( & table_name ) , Arc :: clone ( & partition_key ) , metrics , ) . context ( ParquetChunkError ) ? , ) ; let path : DirsAndFileName = path . into ( ) ; // IMPORTANT: Start transaction AFTER writing the actual parquet file so we do not hold the //            transaction lock (that is part of the PreservedCatalog) for too long. By using the //            cleanup lock (see above) it is ensured that the file that we have written is not deleted //            in between. let mut transaction = db . preserved_catalog . open_transaction ( ) . await ; let info = CatalogParquetInfo { path , file_size_bytes , metadata : parquet_metadata , } ; transaction . add_parquet ( & info ) . context ( TransactionError ) ? ; // preserved commit let ckpt_handle = transaction . commit ( ) . await . context ( CommitError ) ? ; // in-mem commit { let mut guard = chunk . write ( ) ; if let Err ( e ) = guard . set_written_to_object_store ( parquet_chunk ) { panic ! ( " " e ) ; } } let create_checkpoint = ckpt_handle . revision_counter ( ) % catalog_transactions_until_checkpoint == 0 ; if create_checkpoint { // Commit is already done, so we can just scan the catalog for the state. // // NOTE: There can only be a single transaction in this section because the checkpoint handle holds //       transaction lock. Therefore we don't need to worry about concurrent modifications of //       preserved chunks. if let Err ( e ) = ckpt_handle . create_checkpoint ( checkpoint_data_from_catalog ( & db . catalog ) ) . await { warn ! ( e " " ) ; // That's somewhat OK. Don't fail the entire task, because the actual preservation was completed // (both in-mem and within the preserved catalog). } } } { // Flush persisted data from persistence windows let mut partition = partition . write ( ) ; partition . persistence_windows_mut ( ) . expect ( " " ) . flush ( flush_handle ) ; } // We know this chunk is ParquetFile type let chunk = chunk . read ( ) ; Ok ( DbChunk :: parquet_file_snapshot ( & chunk ) ) } ; Ok ( ( tracker , fut . track ( registration ) ) ) }
The implementation for writing a chunk to the object store
Returns a future registered with the tracker registry, and the corresponding tracker
The caller can either spawn this future to tokio, or block directly on it
The implementation for writing a chunk to the object store Returns a future registered with the tracker registry and the corresponding tracker The caller can either spawn this future to tokio or block directly on it
partition
LifecycleWriteGuard<'_, Partition, LockableCatalogPartition>
chunk
LifecycleWriteGuard<'_, CatalogChunk, LockableCatalogChunk>
flush_handle
FlushHandle
partition
LifecycleWriteGuard<'_, Partition, LockableCatalogPartition>
chunk
LifecycleWriteGuard<'_, CatalogChunk, LockableCatalogChunk>
flush_handle
FlushHandle