test: Re-enable running compactor once in step tests
Putting back the test that I removed in <https://github.com/influxdata/influxdb_iox/pull/6850> now that they can use `run compactor2 --compaction-process-once --compaction-process-all-partitions` thanks to <https://github.com/influxdata/influxdb_iox/pull/6886>!pull/24376/head
Carol (Nichols || Goulding)
parent
c0a53ae5a9
commit
528d83e989
|
|
@ -179,6 +179,147 @@ async fn parquet_to_lp() {
|
|||
.await
|
||||
}
|
||||
|
||||
/// Write, compact, then use the remote partition command
|
||||
#[tokio::test]
|
||||
async fn compact_and_get_remote_partition() {
|
||||
test_helpers::maybe_start_logging();
|
||||
let database_url = maybe_skip_integration!();
|
||||
|
||||
// The test below assumes a specific partition id, so use a
|
||||
// non-shared one here so concurrent tests don't interfere with
|
||||
// each other
|
||||
let mut cluster = MiniCluster::create_non_shared2(database_url).await;
|
||||
|
||||
StepTest::new(
|
||||
&mut cluster,
|
||||
vec![
|
||||
Step::RecordNumParquetFiles,
|
||||
Step::WriteLineProtocol(String::from(
|
||||
"my_awesome_table,tag1=A,tag2=B val=42i 123456",
|
||||
)),
|
||||
// wait for partitions to be persisted
|
||||
Step::WaitForPersisted2 {
|
||||
expected_increase: 1,
|
||||
},
|
||||
// Run the compactor
|
||||
Step::Compact,
|
||||
// Run the 'remote partition' command
|
||||
Step::Custom(Box::new(|state: &mut StepTestState| {
|
||||
async {
|
||||
let router_addr = state.cluster().router().router_grpc_base().to_string();
|
||||
let namespace = state.cluster().namespace().to_string();
|
||||
|
||||
// Validate the output of the remote partition CLI command
|
||||
//
|
||||
// Looks something like:
|
||||
// {
|
||||
// "id": "1",
|
||||
// "namespaceId": 1,
|
||||
// "tableId": 1,
|
||||
// "partitionId": "1",
|
||||
// "objectStoreId": "fa6cdcd1-cbc2-4fb7-8b51-4773079124dd",
|
||||
// "minTime": "123456",
|
||||
// "maxTime": "123456",
|
||||
// "fileSizeBytes": "2029",
|
||||
// "rowCount": "1",
|
||||
// "compactionLevel": "1",
|
||||
// "createdAt": "1650019674289347000"
|
||||
// }
|
||||
let out = Command::cargo_bin("influxdb_iox")
|
||||
.unwrap()
|
||||
.arg("-h")
|
||||
.arg(&router_addr)
|
||||
.arg("remote")
|
||||
.arg("partition")
|
||||
.arg("show")
|
||||
.arg("1")
|
||||
.assert()
|
||||
.success()
|
||||
.stdout(
|
||||
// Important parts are the expected partition ID
|
||||
predicate::str::contains(r#""partitionId": "1","#)
|
||||
// and compaction level
|
||||
.and(predicate::str::contains(r#""compactionLevel": 1"#)),
|
||||
)
|
||||
.get_output()
|
||||
.stdout
|
||||
.clone();
|
||||
|
||||
let object_store_id = get_object_store_id(&out);
|
||||
let dir = tempdir().unwrap();
|
||||
let f = dir.path().join("tmp.parquet");
|
||||
let filename = f.as_os_str().to_str().unwrap();
|
||||
|
||||
Command::cargo_bin("influxdb_iox")
|
||||
.unwrap()
|
||||
.arg("-h")
|
||||
.arg(&router_addr)
|
||||
.arg("remote")
|
||||
.arg("store")
|
||||
.arg("get")
|
||||
.arg(&object_store_id)
|
||||
.arg(filename)
|
||||
.assert()
|
||||
.success()
|
||||
.stdout(
|
||||
predicate::str::contains("wrote")
|
||||
.and(predicate::str::contains(filename)),
|
||||
);
|
||||
|
||||
// Ensure a warning is emitted when specifying (or
|
||||
// defaulting to) in-memory file storage.
|
||||
Command::cargo_bin("influxdb_iox")
|
||||
.unwrap()
|
||||
.arg("-h")
|
||||
.arg(&router_addr)
|
||||
.arg("remote")
|
||||
.arg("partition")
|
||||
.arg("pull")
|
||||
.arg("--catalog")
|
||||
.arg("memory")
|
||||
.arg("--object-store")
|
||||
.arg("memory")
|
||||
.arg(&namespace)
|
||||
.arg("my_awesome_table")
|
||||
.arg("1970-01-01")
|
||||
.assert()
|
||||
.failure()
|
||||
.stderr(predicate::str::contains("try passing --object-store=file"));
|
||||
|
||||
// Ensure files are actually wrote to the filesystem
|
||||
let dir = tempfile::tempdir().expect("could not get temporary directory");
|
||||
|
||||
Command::cargo_bin("influxdb_iox")
|
||||
.unwrap()
|
||||
.arg("-h")
|
||||
.arg(&router_addr)
|
||||
.arg("remote")
|
||||
.arg("partition")
|
||||
.arg("pull")
|
||||
.arg("--catalog")
|
||||
.arg("memory")
|
||||
.arg("--object-store")
|
||||
.arg("file")
|
||||
.arg("--data-dir")
|
||||
.arg(dir.path().to_str().unwrap())
|
||||
.arg(&namespace)
|
||||
.arg("my_awesome_table")
|
||||
.arg("1970-01-01")
|
||||
.assert()
|
||||
.success()
|
||||
.stdout(
|
||||
predicate::str::contains("wrote file")
|
||||
.and(predicate::str::contains(object_store_id)),
|
||||
);
|
||||
}
|
||||
.boxed()
|
||||
})),
|
||||
],
|
||||
)
|
||||
.run()
|
||||
.await
|
||||
}
|
||||
|
||||
/// Test the schema cli command
|
||||
#[tokio::test]
|
||||
async fn schema_cli() {
|
||||
|
|
|
|||
|
|
@ -121,6 +121,7 @@ impl TestConfig {
|
|||
other.dsn().to_owned(),
|
||||
other.catalog_schema_name(),
|
||||
)
|
||||
.with_env("INFLUXDB_IOX_RPC_MODE", "2")
|
||||
.with_existing_object_store(other)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -473,9 +473,12 @@ impl MiniCluster {
|
|||
|
||||
let mut command = Command::cargo_bin("influxdb_iox").unwrap();
|
||||
let command = command
|
||||
.arg("compactor")
|
||||
.arg("run-once")
|
||||
.arg("run")
|
||||
.arg("compactor2")
|
||||
.arg("--compaction-process-once")
|
||||
.arg("--compaction-process-all-partitions")
|
||||
.env("LOG_FILTER", log_filter)
|
||||
.env("INFLUXDB_IOX_GRPC_BIND_ADDR", "8999")
|
||||
.env(
|
||||
"INFLUXDB_IOX_CATALOG_DSN",
|
||||
self.compactor_config()
|
||||
|
|
@ -494,7 +497,10 @@ impl MiniCluster {
|
|||
|
||||
log_command(command);
|
||||
|
||||
command.ok().unwrap();
|
||||
if let Err(e) = command.ok() {
|
||||
dump_log_to_stdout("compactor run-once", &log_path);
|
||||
panic!("Command failed: {:?}", e);
|
||||
}
|
||||
dump_log_to_stdout("compactor run-once", &log_path);
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue