Influxdata
/
influxdb
mirror of https://github.com/influxdata/influxdb.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

fix: chunk dropping over lifecycle policy should also respect the preserved catalog

pull/24376/head
Marco Neumann 2021-07-13 13:27:29 +02:00
parent 71cb15f017
commit 77a9191a11
3 changed files with 26 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
lifecycle/src/lib.rs
View File

@ -104,9 +104,9 @@ pub trait LockablePartition: Sized + std::fmt::Display {
/// Drops a chunk from the partition
fn drop_chunk(
s: LifecycleWriteGuard<'_, Self::Partition, Self>,
chunk_id: u32,
) -> Result<(), Self::Error>;
partition: LifecycleWriteGuard<'_, Self::Partition, Self>,
chunk: LifecycleWriteGuard<'_, <Self::Chunk as LockableChunk>::Chunk, Self::Chunk>,
) -> Result<TaskTracker<<Self::Chunk as LockableChunk>::Job>, Self::Error>;
}
/// A `LockableChunk` is a wrapper around a `LifecycleChunk` that allows for

24
lifecycle/src/policy.rs
View File

@ -134,11 +134,13 @@ where
FreeAction::Drop => match chunk.storage() {
ChunkStorage::ReadBuffer
| ChunkStorage::ClosedMutableBuffer => {
LockablePartition::drop_chunk(
let tracker = LockablePartition::drop_chunk(
partition.upgrade(),
candidate.chunk_id,
chunk.upgrade(),
)
.expect("failed to drop")
.with_metadata(ChunkLifecycleAction::Dropping);
self.trackers.push(tracker);
}
storage => warn!(
%db_name,
@ -858,12 +860,18 @@ mod tests {
}
fn drop_chunk(
mut s: LifecycleWriteGuard<'_, Self::Partition, Self>,
chunk_id: u32,
) -> Result<(), Self::Error> {
s.chunks.remove(&chunk_id);
s.data().db.events.write().push(MoverEvents::Drop(chunk_id));
Ok(())
mut partition: LifecycleWriteGuard<'_, Self::Partition, Self>,
chunk: LifecycleWriteGuard<'_, TestChunk, Self::Chunk>,
) -> Result<TaskTracker<()>, Self::Error> {
let chunk_id = chunk.addr().chunk_id;
partition.chunks.remove(&chunk_id);
partition
.data()
.db
.events
.write()
.push(MoverEvents::Drop(chunk_id));
Ok(TaskTracker::complete(()))
}
}

12
server/src/db/lifecycle.rs
View File

@ -232,11 +232,13 @@ impl LockablePartition for LockableCatalogPartition {
}
fn drop_chunk(
mut s: LifecycleWriteGuard<'_, Self::Partition, Self>,
chunk_id: u32,
) -> Result<(), Self::Error> {
s.drop_chunk(chunk_id)?;
Ok(())
partition: LifecycleWriteGuard<'_, Self::Partition, Self>,
chunk: LifecycleWriteGuard<'_, CatalogChunk, Self::Chunk>,
) -> Result<TaskTracker<Job>, Self::Error> {
info!(table=%partition.table_name(), partition=%partition.partition_key(), chunk_id=chunk.addr().chunk_id, "drop chunk");
let (tracker, fut) = drop::drop_chunk(partition, chunk)?;
let _ = tokio::spawn(async move { fut.await.log_if_error("drop chunk") });
Ok(tracker)
}
}