Skip to content

fix: clear the target chunk when the copy goes through the hypertable #207

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
alejandrodnm wants to merge 1 commit into from
Open

fix: clear the target chunk when the copy goes through the hypertable #207

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 22 additions & 7 deletions src/execute.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,7 @@ pub async fn copy_chunk(
target_tx,
&task.source_chunk,
&target_chunk,
&target_chunk,
&task.filter,
CopyMode::UncompressedAndCompressed,
)
Expand Down Expand Up @@ -147,6 +148,7 @@ pub async fn copy_chunk(
source_tx,
target_tx,
&task.source_chunk,
&target_chunk,
&target_chunk.hypertable,
&task.filter,
UncompressedOnly,
Expand All @@ -158,6 +160,7 @@ pub async fn copy_chunk(
target_tx,
&task.source_chunk,
&target_chunk,
&target_chunk,
&task.filter,
UncompressedOnly,
)
Expand Down Expand Up @@ -233,6 +236,7 @@ async fn copy_chunk_data_and_recompress(
target_tx,
&task.source_chunk,
target_chunk,
target_chunk,
&task.filter,
CopyMode::UncompressedAndCompressed,
)
Expand Down Expand Up @@ -372,48 +376,59 @@ enum CopyMode {
UncompressedAndCompressed,
}

/// Copies a source chunk's uncompressed rows into the target.
///
/// `target_chunk` is the chunk the rows belong to. Clearing the range and
/// suspending the cagg invalidation trigger both act on it, because it is the
/// relation that ends up holding the rows. `copy_into` is where the rows are
/// written, which is that same chunk except when the caller has to route them
/// through the hypertable to let TimescaleDB set the chunk's status. Deleting
/// from the hypertable instead would clear nothing: `DELETE FROM ONLY
/// <hypertable>` reaches only the hypertable's own heap, and a hypertable keeps
/// every row in its chunks.
async fn copy_chunk_data<S: QuotedName, T: QuotedName>(
source_tx: &Transaction<'_>,
target_tx: &Transaction<'_>,
source_table: &S,
target_table: &T,
target_chunk: &TargetChunk,
copy_into: &T,
filter: &Option<String>,
mode: CopyMode,
) -> Result<CopyResult> {
debug!("Copying uncompressed chunk {}", source_table.quoted_name());

let trigger_dropped = if features::cagg_invalidation_trigger() {
drop_invalidation_trigger(target_tx, &target_table.quoted_name()).await?
drop_invalidation_trigger(target_tx, &target_chunk.quoted_name()).await?
} else {
false
};

if let Some(filter) = filter {
delete_data_using_filter(target_tx, target_table, filter).await?;
delete_data_using_filter(target_tx, target_chunk, filter).await?;
} else {
delete_all_rows_from_chunk(target_tx, &target_table.quoted_name()).await?;
delete_all_rows_from_chunk(target_tx, &target_chunk.quoted_name()).await?;
}

let copy_result = copy_chunk_from_source_to_target(
source_tx,
target_tx,
&source_table.quoted_name(),
&target_table.quoted_name(),
&copy_into.quoted_name(),
filter,
mode == UncompressedOnly,
)
.await?;

if trigger_dropped {
create_invalidation_trigger(target_tx, &target_table.quoted_name()).await?;
create_invalidation_trigger(target_tx, &target_chunk.quoted_name()).await?;
}
debug!(
"Finished copying uncompressed chunk {}. Starting analysis.",
source_table.quoted_name()
);

target_tx
.execute(&format!("analyze {}", target_table.quoted_name()), &[])
.execute(&format!("analyze {}", copy_into.quoted_name()), &[])
.await?;

debug!(
Expand Down
125 changes: 103 additions & 22 deletions tests/integration/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1632,20 +1632,6 @@ fn skips_unless_ts228(test: &str) -> bool {
true
}

/// Reports a version-gated skip for a test that can only observe its bug below
/// TimescaleDB 2.23. From 2.23 on (timescaledb#8704) the `DELETE` that clears a
/// chunk's uncompressed rows also decompresses and removes its compressed
/// batches, so the assertion holds there for an unrelated reason and would
/// report coverage the run doesn't have. 2.22 is the only version in CI below
/// the cutoff, so dropping it from the matrix takes this test with it.
fn skips_unless_ts_lt_223(test: &str) -> bool {
if ts_version() < TS223 {
return false;
}
eprintln!("SKIPPED {test}: requires TimescaleDB < 2.23");
true
}

/// A chunk compressed under a pre-2.28 format carries `minmax` sparse-index
/// metadata only. When such a chunk is backfilled into a target whose
/// hypertable was configured under TS 2.28, `create_compressed_chunk` copies
Expand Down Expand Up @@ -1781,16 +1767,13 @@ fn filtered_copy_still_recompresses_on_unbacked_sparse_index_metadata() -> Resul
/// the uncompressed chunk that carries the hypertable's indexes and
/// constraints.
///
/// Only TimescaleDB < 2.23 is affected, so that is where this runs. From 2.23
/// on (timescaledb#8704) the `DELETE FROM ONLY <chunk>` that clears the
/// uncompressed rows decompresses the chunk's batches and removes them too,
/// which hides the missing delete.
/// Only TimescaleDB < 2.23 could ever observe the duplication. From 2.23 on
/// (timescaledb#8704) the `DELETE FROM ONLY <chunk>` that clears the
/// uncompressed rows decompresses the chunk's batches and removes them too, so
/// the assertion holds there for an unrelated reason. The invariant is the same
/// on every version, so the test runs on all of them.
#[test]
fn copying_a_compressed_chunk_twice_replaces_its_rows() -> Result<()> {
if skips_unless_ts_lt_223("copying_a_compressed_chunk_twice_replaces_its_rows") {
return Ok(());
}

let _ = pretty_env_logger::try_init();

let docker = Cli::default();
Expand Down Expand Up @@ -1852,6 +1835,104 @@ fn copying_a_compressed_chunk_twice_replaces_its_rows() -> Result<()> {
Ok(())
}

static INSERT_SECOND_SERIES_FOR_MAY: &str = r"
INSERT INTO metrics (time, device_id, val)
SELECT time, 2, random()
FROM generate_series('2023-05-01T00:30:00Z'::timestamptz, '2023-05-31T23:30:00Z'::timestamptz, '1 hour'::interval) time;
";

static RECOMPRESS_ALL_CHUNKS: &str = r"
SELECT decompress_chunk(format('%I.%I', chunk_schema, chunk_name)) FROM timescaledb_information.chunks WHERE is_compressed = true;
SELECT compress_chunk(format('%I.%I', chunk_schema, chunk_name)) FROM timescaledb_information.chunks WHERE is_compressed = false;
";

/// When the source chunk is partial but the target chunk isn't, `copy_chunk`
/// writes the uncompressed rows through the hypertable rather than into the
/// chunk, because setting a chunk's status to partial needs a permission
/// `tsdbadmin` doesn't have. The delete that precedes the copy has to keep
/// pointing at the chunk: `DELETE FROM ONLY <hypertable>` reaches only the
/// hypertable's own heap, which is always empty, so it clears nothing and the
/// copy adds to whatever the chunk already held.
///
/// The target chunk stops being partial once something folds its loose rows
/// back into compressed form, which is what a compression policy does. That is
/// what puts a second copy back on this branch with data already in place.
#[test]
fn copying_a_partial_chunk_through_the_hypertable_replaces_its_rows() -> Result<()> {
let _ = pretty_env_logger::try_init();

let docker = Cli::default();

let source_container = docker.run(timescaledb(pg_version(), ts_version()));
let target_container = docker.run(timescaledb(pg_version(), ts_version()));

psql(&source_container, PsqlInput::Sql(SETUP_HYPERTABLE))?;
psql(&source_container, PsqlInput::Sql(INSERT_DATA_FOR_MAY))?;
psql(
&source_container,
PsqlInput::Sql(ENABLE_HYPERTABLE_COMPRESSION),
)?;
psql(&source_container, PsqlInput::Sql(COMPRESS_ALL_CHUNKS))?;
// Inserted after compression, so these rows stay uncompressed in the
// already-compressed chunks and make every source chunk partial.
psql(
&source_container,
PsqlInput::Sql(INSERT_SECOND_SERIES_FOR_MAY),
)?;

psql(&target_container, PsqlInput::Sql(SETUP_HYPERTABLE))?;
psql(
&target_container,
PsqlInput::Sql(ENABLE_HYPERTABLE_COMPRESSION),
)?;

let stage =
|| TestConfigStage::new(&source_container, &target_container, "2023-06-01T00:00:00");

run_backfill(stage())?
.assert()
.success()
.stdout(contains("Staged 5 chunks to copy"));
run_backfill(TestConfigCopy::new(&source_container, &target_container))?
.assert()
.success()
.stdout(contains("is partial in source, but not in target"));

// Stand in for a compression policy catching up with the target: the loose
// rows the copy just wrote are folded back into compressed form, which
// clears the chunks' partial status and puts the second copy back on the
// hypertable branch.
psql(&target_container, PsqlInput::Sql(RECOMPRESS_ALL_CHUNKS))?;

run_backfill(TestConfigClean::new(&target_container))?
.assert()
.success();

run_backfill(stage())?.assert().success();
run_backfill(TestConfigCopy::new(&source_container, &target_container))?
.assert()
.success()
.stdout(contains("is partial in source, but not in target"));

let mut source_dbassert =
DbAssert::new(&source_container.connection_string())?.with_name("source");
let mut target_dbassert =
DbAssert::new(&target_container.connection_string())?.with_name("target");

for dbassert in [&mut source_dbassert, &mut target_dbassert] {
dbassert
.has_table_count("public", "metrics", 1488)
.has_chunk_count("public", "metrics", 5);
}

run_backfill(TestConfigVerify::new(&source_container, &target_container))?
.assert()
.success()
.stdout(contains("Chunk verification failed").not());

Ok(())
}

#[test]
fn ctrl_c_stops_gracefully() -> Result<()> {
let _ = pretty_env_logger::try_init();
Expand Down