diff --git a/Cargo.lock b/Cargo.lock index 71685e5c6ed..346c9c1dd6c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -444,7 +444,7 @@ dependencies = [ [[package]] name = "arrow_util" -version = "3.10.4" +version = "3.10.5" dependencies = [ "ahash 0.8.12", "arrow", @@ -541,7 +541,7 @@ checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" [[package]] name = "authz" -version = "3.10.4" +version = "3.10.5" dependencies = [ "assert_matches", "async-trait", @@ -655,7 +655,7 @@ dependencies = [ [[package]] name = "backoff" -version = "3.10.4" +version = "3.10.5" dependencies = [ "rand 0.9.4", "snafu 0.9.0", @@ -1078,7 +1078,7 @@ checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5" [[package]] name = "catalog_cache" -version = "3.10.4" +version = "3.10.5" dependencies = [ "bytes", "criterion 0.8.2", @@ -1222,14 +1222,14 @@ checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" [[package]] name = "cli_types" -version = "3.10.4" +version = "3.10.5" dependencies = [ "tokio", ] [[package]] name = "client_util" -version = "3.10.4" +version = "3.10.5" dependencies = [ "http 1.4.0", "iox_http_util", @@ -1832,7 +1832,7 @@ dependencies = [ [[package]] name = "data_types" -version = "3.10.4" +version = "3.10.5" dependencies = [ "arrow", "arrow-buffer", @@ -2566,7 +2566,7 @@ dependencies = [ [[package]] name = "datafusion_util" -version = "3.10.4" +version = "3.10.5" dependencies = [ "assert_matches", "async-trait", @@ -2661,7 +2661,7 @@ dependencies = [ [[package]] name = "dml" -version = "3.10.4" +version = "3.10.5" dependencies = [ "arrow_util", "data_types", @@ -2773,7 +2773,7 @@ dependencies = [ [[package]] name = "error_reporting" -version = "3.10.4" +version = "3.10.5" dependencies = [ "datafusion", "snafu 0.9.0", @@ -2803,7 +2803,7 @@ dependencies = [ [[package]] name = "executor" -version = "3.10.4" +version = "3.10.5" dependencies = [ "futures", "metric", @@ -2884,7 +2884,7 @@ dependencies = [ [[package]] name = "flightsql" -version = "3.10.4" +version = "3.10.5" dependencies = [ "arrow", "arrow-flight", @@ -3091,7 +3091,7 @@ dependencies = [ [[package]] name = "futures_test_utils" -version = "3.10.4" +version = "3.10.5" dependencies = [ "clap", "futures", @@ -3102,7 +3102,7 @@ dependencies = [ [[package]] name = "generated_types" -version = "3.10.4" +version = "3.10.5" dependencies = [ "bytes", "pbjson", @@ -3756,7 +3756,7 @@ dependencies = [ [[package]] name = "influxdb2_client" -version = "3.10.4" +version = "3.10.5" dependencies = [ "bytes", "futures", @@ -3775,7 +3775,7 @@ dependencies = [ [[package]] name = "influxdb3" -version = "3.10.4" +version = "3.10.5" dependencies = [ "anyhow", "arrow", @@ -3867,7 +3867,7 @@ dependencies = [ [[package]] name = "influxdb3_authz" -version = "3.10.4" +version = "3.10.5" dependencies = [ "async-trait", "authz", @@ -3885,7 +3885,7 @@ dependencies = [ [[package]] name = "influxdb3_cache" -version = "3.10.4" +version = "3.10.5" dependencies = [ "anyhow", "arrow", @@ -3920,7 +3920,7 @@ dependencies = [ [[package]] name = "influxdb3_catalog" -version = "3.10.4" +version = "3.10.5" dependencies = [ "ahash 0.8.12", "anyhow", @@ -3977,7 +3977,7 @@ dependencies = [ [[package]] name = "influxdb3_clap_blocks" -version = "3.10.4" +version = "3.10.5" dependencies = [ "async-trait", "bytes", @@ -4014,7 +4014,7 @@ dependencies = [ [[package]] name = "influxdb3_client" -version = "3.10.4" +version = "3.10.5" dependencies = [ "bytes", "flate2", @@ -4035,7 +4035,7 @@ dependencies = [ [[package]] name = "influxdb3_commands" -version = "3.10.4" +version = "3.10.5" dependencies = [ "anyhow", "chrono", @@ -4062,7 +4062,7 @@ dependencies = [ [[package]] name = "influxdb3_id" -version = "3.10.4" +version = "3.10.5" dependencies = [ "criterion 0.5.1", "indexmap 2.14.0", @@ -4074,7 +4074,7 @@ dependencies = [ [[package]] name = "influxdb3_internal_api" -version = "3.10.4" +version = "3.10.5" dependencies = [ "anyhow", "async-trait", @@ -4095,7 +4095,7 @@ dependencies = [ [[package]] name = "influxdb3_load_generator" -version = "3.10.4" +version = "3.10.5" dependencies = [ "anyhow", "bytes", @@ -4125,7 +4125,7 @@ dependencies = [ [[package]] name = "influxdb3_process" -version = "3.10.4" +version = "3.10.5" dependencies = [ "iox_time", "uuid", @@ -4133,7 +4133,7 @@ dependencies = [ [[package]] name = "influxdb3_processing_engine" -version = "3.10.4" +version = "3.10.5" dependencies = [ "anyhow", "async-trait", @@ -4177,7 +4177,7 @@ dependencies = [ [[package]] name = "influxdb3_py_api" -version = "3.10.4" +version = "3.10.5" dependencies = [ "anyhow", "arrow-array", @@ -4205,7 +4205,7 @@ dependencies = [ [[package]] name = "influxdb3_query_executor" -version = "3.10.4" +version = "3.10.5" dependencies = [ "anyhow", "arrow", @@ -4256,7 +4256,7 @@ dependencies = [ [[package]] name = "influxdb3_server" -version = "3.10.4" +version = "3.10.5" dependencies = [ "anyhow", "arrow", @@ -4343,7 +4343,7 @@ dependencies = [ [[package]] name = "influxdb3_shutdown" -version = "3.10.4" +version = "3.10.5" dependencies = [ "futures", "futures-util", @@ -4358,14 +4358,14 @@ dependencies = [ [[package]] name = "influxdb3_startup" -version = "3.10.4" +version = "3.10.5" dependencies = [ "chrono", ] [[package]] name = "influxdb3_sys_events" -version = "3.10.4" +version = "3.10.5" dependencies = [ "arrow", "arrow-array", @@ -4384,7 +4384,7 @@ dependencies = [ [[package]] name = "influxdb3_system_tables" -version = "3.10.4" +version = "3.10.5" dependencies = [ "arrow", "arrow-array", @@ -4416,7 +4416,7 @@ dependencies = [ [[package]] name = "influxdb3_telemetry" -version = "3.10.4" +version = "3.10.5" dependencies = [ "futures", "futures-util", @@ -4437,7 +4437,7 @@ dependencies = [ [[package]] name = "influxdb3_test_helpers" -version = "3.10.4" +version = "3.10.5" dependencies = [ "async-trait", "bytes", @@ -4450,7 +4450,7 @@ dependencies = [ [[package]] name = "influxdb3_types" -version = "3.10.4" +version = "3.10.5" dependencies = [ "anyhow", "chrono", @@ -4470,7 +4470,7 @@ dependencies = [ [[package]] name = "influxdb3_wal" -version = "3.10.4" +version = "3.10.5" dependencies = [ "async-trait", "bitcode", @@ -4499,7 +4499,7 @@ dependencies = [ [[package]] name = "influxdb3_write" -version = "3.10.4" +version = "3.10.5" dependencies = [ "anyhow", "arrow", @@ -4569,7 +4569,7 @@ dependencies = [ [[package]] name = "influxdb_influxql_parser" -version = "3.10.4" +version = "3.10.5" dependencies = [ "assert_matches", "chrono", @@ -4586,7 +4586,7 @@ dependencies = [ [[package]] name = "influxdb_iox_client" -version = "3.10.4" +version = "3.10.5" dependencies = [ "arrow", "arrow-flight", @@ -4615,7 +4615,7 @@ dependencies = [ [[package]] name = "ingester_query_grpc" -version = "3.10.4" +version = "3.10.5" dependencies = [ "base64 0.22.1", "data_types", @@ -4681,7 +4681,7 @@ checksum = "06432fb54d3be7964ecd3649233cddf80db2832f47fec34c01f65b3d9d774983" [[package]] name = "iox_http" -version = "3.10.4" +version = "3.10.5" dependencies = [ "assert_matches", "async-trait", @@ -4701,7 +4701,7 @@ dependencies = [ [[package]] name = "iox_http_util" -version = "3.10.4" +version = "3.10.5" dependencies = [ "futures", "http 1.4.0", @@ -4712,7 +4712,7 @@ dependencies = [ [[package]] name = "iox_query" -version = "3.10.4" +version = "3.10.5" dependencies = [ "arrow", "arrow_util", @@ -4761,7 +4761,7 @@ dependencies = [ [[package]] name = "iox_query_influxql" -version = "3.10.4" +version = "3.10.5" dependencies = [ "arrow", "assert_matches", @@ -4788,7 +4788,7 @@ dependencies = [ [[package]] name = "iox_query_influxql_rewrite" -version = "3.10.4" +version = "3.10.5" dependencies = [ "influxdb_influxql_parser", "thiserror 2.0.18", @@ -4796,7 +4796,7 @@ dependencies = [ [[package]] name = "iox_query_params" -version = "3.10.4" +version = "3.10.5" dependencies = [ "arrow", "assert_matches", @@ -4810,7 +4810,7 @@ dependencies = [ [[package]] name = "iox_query_udf" -version = "3.10.4" +version = "3.10.5" dependencies = [ "datafusion-udf-wasm-host", "datafusion-udf-wasm-query", @@ -4821,7 +4821,7 @@ dependencies = [ [[package]] name = "iox_system_tables" -version = "3.10.4" +version = "3.10.5" dependencies = [ "arrow", "async-trait", @@ -4831,7 +4831,7 @@ dependencies = [ [[package]] name = "iox_time" -version = "3.10.4" +version = "3.10.5" dependencies = [ "chrono", "parking_lot", @@ -4840,7 +4840,7 @@ dependencies = [ [[package]] name = "iox_v1_query_api" -version = "3.10.4" +version = "3.10.5" dependencies = [ "anyhow", "arrow", @@ -4954,7 +4954,7 @@ checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" [[package]] name = "jemalloc_pprof_http" -version = "3.10.4" +version = "3.10.5" dependencies = [ "http-body-util", "hyper 1.9.0", @@ -4967,7 +4967,7 @@ dependencies = [ [[package]] name = "jemalloc_stats" -version = "3.10.4" +version = "3.10.5" dependencies = [ "futures", "snafu 0.9.0", @@ -5160,7 +5160,7 @@ dependencies = [ [[package]] name = "linear_buffer" -version = "3.10.4" +version = "3.10.5" [[package]] name = "linux-raw-sys" @@ -5204,7 +5204,7 @@ checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" [[package]] name = "logfmt" -version = "3.10.4" +version = "3.10.5" dependencies = [ "humantime", "parking_lot", @@ -5303,14 +5303,14 @@ dependencies = [ [[package]] name = "metric" -version = "3.10.4" +version = "3.10.5" dependencies = [ "parking_lot", ] [[package]] name = "metric_exporters" -version = "3.10.4" +version = "3.10.5" dependencies = [ "metric", "mockito", @@ -5435,7 +5435,7 @@ checksum = "9252111cf132ba0929b6f8e030cac2a24b507f3a4d6db6fb2896f27b354c714b" [[package]] name = "mutable_batch" -version = "3.10.4" +version = "3.10.5" dependencies = [ "arrow", "arrow_util", @@ -5453,7 +5453,7 @@ dependencies = [ [[package]] name = "mutable_batch_lp" -version = "3.10.4" +version = "3.10.5" dependencies = [ "arrow_util", "assert_matches", @@ -5469,7 +5469,7 @@ dependencies = [ [[package]] name = "mutable_batch_pb" -version = "3.10.4" +version = "3.10.5" dependencies = [ "arrow_util", "criterion 0.8.2", @@ -5707,7 +5707,7 @@ dependencies = [ [[package]] name = "object_store_limit" -version = "3.10.4" +version = "3.10.5" dependencies = [ "async-trait", "bytes", @@ -5719,7 +5719,7 @@ dependencies = [ [[package]] name = "object_store_mem_cache" -version = "3.10.4" +version = "3.10.5" dependencies = [ "async-trait", "bincode", @@ -5747,7 +5747,7 @@ dependencies = [ [[package]] name = "object_store_metrics" -version = "3.10.4" +version = "3.10.5" dependencies = [ "async-trait", "bloom2", @@ -5772,7 +5772,7 @@ dependencies = [ [[package]] name = "object_store_mock" -version = "3.10.4" +version = "3.10.5" dependencies = [ "async-trait", "bytes", @@ -5786,7 +5786,7 @@ dependencies = [ [[package]] name = "object_store_size_hinting" -version = "3.10.4" +version = "3.10.5" dependencies = [ "chrono", "http 1.4.0", @@ -5796,7 +5796,7 @@ dependencies = [ [[package]] name = "object_store_utils" -version = "3.10.4" +version = "3.10.5" dependencies = [ "async-trait", "backon", @@ -5810,7 +5810,7 @@ dependencies = [ [[package]] name = "observability_deps" -version = "3.10.4" +version = "3.10.5" dependencies = [ "tracing", ] @@ -5875,7 +5875,7 @@ dependencies = [ [[package]] name = "panic_logging" -version = "3.10.4" +version = "3.10.5" dependencies = [ "metric", "test_helpers", @@ -5950,7 +5950,7 @@ dependencies = [ [[package]] name = "parquet_file" -version = "3.10.4" +version = "3.10.5" dependencies = [ "arrow", "arrow_util", @@ -5981,7 +5981,7 @@ dependencies = [ [[package]] name = "partition" -version = "3.10.4" +version = "3.10.5" dependencies = [ "arrow", "assert_matches", @@ -6278,7 +6278,7 @@ dependencies = [ [[package]] name = "predicate" -version = "3.10.4" +version = "3.10.5" dependencies = [ "arrow", "async-trait", @@ -6622,7 +6622,7 @@ dependencies = [ [[package]] name = "query_functions" -version = "3.10.4" +version = "3.10.5" dependencies = [ "arrow", "chrono", @@ -7332,7 +7332,7 @@ dependencies = [ [[package]] name = "schema" -version = "3.10.4" +version = "3.10.5" dependencies = [ "arrow", "base64 0.22.1", @@ -7515,7 +7515,7 @@ dependencies = [ [[package]] name = "service_common" -version = "3.10.4" +version = "3.10.5" dependencies = [ "arrow", "arrow-flight", @@ -7529,7 +7529,7 @@ dependencies = [ [[package]] name = "service_grpc_flight" -version = "3.10.4" +version = "3.10.5" dependencies = [ "arrow", "arrow-flight", @@ -7605,7 +7605,7 @@ dependencies = [ [[package]] name = "sharder" -version = "3.10.4" +version = "3.10.5" dependencies = [ "criterion 0.8.2", "data_types", @@ -8127,7 +8127,7 @@ dependencies = [ [[package]] name = "table_batch" -version = "3.10.4" +version = "3.10.5" dependencies = [ "arrow_util", "bitvec", @@ -8223,7 +8223,7 @@ dependencies = [ [[package]] name = "test_helpers" -version = "3.10.4" +version = "3.10.5" dependencies = [ "async-trait", "dotenvy", @@ -8242,7 +8242,7 @@ dependencies = [ [[package]] name = "test_helpers_authz" -version = "3.10.4" +version = "3.10.5" dependencies = [ "futures", "generated_types", @@ -8515,7 +8515,7 @@ dependencies = [ [[package]] name = "tokio_metrics_bridge" -version = "3.10.4" +version = "3.10.5" dependencies = [ "metric", "parking_lot", @@ -8524,7 +8524,7 @@ dependencies = [ [[package]] name = "tokio_watchdog" -version = "3.10.4" +version = "3.10.5" dependencies = [ "metric", "test_helpers", @@ -8747,7 +8747,7 @@ checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" [[package]] name = "tower_trailer" -version = "3.10.4" +version = "3.10.5" dependencies = [ "futures", "http 1.4.0", @@ -8759,7 +8759,7 @@ dependencies = [ [[package]] name = "trace" -version = "3.10.4" +version = "3.10.5" dependencies = [ "chrono", "parking_lot", @@ -8769,7 +8769,7 @@ dependencies = [ [[package]] name = "trace_exporters" -version = "3.10.4" +version = "3.10.5" dependencies = [ "async-trait", "chrono", @@ -8786,7 +8786,7 @@ dependencies = [ [[package]] name = "trace_http" -version = "3.10.4" +version = "3.10.5" dependencies = [ "bytes", "futures", @@ -8881,7 +8881,7 @@ dependencies = [ [[package]] name = "tracker" -version = "3.10.4" +version = "3.10.5" dependencies = [ "futures", "hashbrown 0.14.5", @@ -8901,7 +8901,7 @@ dependencies = [ [[package]] name = "trogging" -version = "3.10.4" +version = "3.10.5" dependencies = [ "clap", "logfmt", diff --git a/Cargo.toml b/Cargo.toml index 30c40e47851..396e30743e0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -127,7 +127,7 @@ exclude = [ # # For porting to the actual OSS repo (influxdb), this would need to change to `-nightly`, i.e., by # stripping the `-oss`. -version = "3.10.4" +version = "3.10.5" authors = ["InfluxData OSS Developers"] edition = "2024" license = "MIT OR Apache-2.0" diff --git a/influxdb3_write/src/paths.rs b/influxdb3_write/src/paths.rs index 05648f02818..eaba8e5c0bf 100644 --- a/influxdb3_write/src/paths.rs +++ b/influxdb3_write/src/paths.rs @@ -50,10 +50,46 @@ impl ParquetFilePath { table_id: u32, chunk_time: i64, wal_file_sequence_number: WalFileSequenceNumber, + ) -> Self { + Self::new_with_chunk_ordinal( + host_prefix, + db_id, + table_id, + chunk_time, + wal_file_sequence_number, + 0, + ) + } + + /// Like [`Self::new`], but for the `chunk_ordinal`-th buffer chunk persisted for the same + /// `(table, chunk_time)` in one snapshot. + /// + /// A gen1 chunk splits into multiple buffer chunks when a string/tag column would exceed + /// the Arrow varchar limit (~2 GiB, see `table_buffer::var_col_max_bytes`). Each split + /// chunk must persist to a distinct object path: with a shared path, the persist jobs + /// race, the last PUT silently overwrites the others, and the snapshot records every + /// job's size for one object — leaving stale size records that fail reads with + /// "Invalid Parquet file. Corrupt footer". + /// + /// Ordinal 0 produces the historical filename exactly, so single-chunk persists (the + /// overwhelmingly common case) and all files written before this change are unaffected. + /// Ordinal n >= 1 appends `-{n}` before the extension. + pub fn new_with_chunk_ordinal( + host_prefix: &str, + db_id: u32, + table_id: u32, + chunk_time: i64, + wal_file_sequence_number: WalFileSequenceNumber, + chunk_ordinal: u32, ) -> Self { let date_time = DateTime::::from_timestamp_nanos(chunk_time); + let ordinal_suffix = if chunk_ordinal == 0 { + String::new() + } else { + format!("-{chunk_ordinal}") + }; let path = ObjPath::from(format!( - "{host_prefix}/dbs/{db_id}/{table_id}/{date_string}/{wal_seq:010}.{ext}", + "{host_prefix}/dbs/{db_id}/{table_id}/{date_string}/{wal_seq:010}{ordinal_suffix}.{ext}", date_string = date_time.format("%Y-%m-%d/%H-%M"), wal_seq = wal_file_sequence_number.as_u64(), ext = PARQUET_FILE_EXTENSION @@ -545,6 +581,41 @@ fn parquet_file_path_new() { ); } +#[test] +fn parquet_file_path_with_chunk_ordinal() { + let ts = Utc + .with_ymd_and_hms(2038, 1, 19, 3, 14, 7) + .unwrap() + .timestamp_nanos_opt() + .unwrap(); + // Ordinal 0 must produce the historical filename exactly — files written before + // this change and the common single-chunk case keep their names. + assert_eq!( + ParquetFilePath::new_with_chunk_ordinal( + "my_host", + 0, + 0, + ts, + WalFileSequenceNumber::new(1337), + 0 + ), + ParquetFilePath::new("my_host", 0, 0, ts, WalFileSequenceNumber::new(1337)), + ); + // Ordinal n >= 1 appends "-{n}" before the extension, keeping the wal sequence + // number as the sortable filename prefix. + assert_eq!( + *ParquetFilePath::new_with_chunk_ordinal( + "my_host", + 0, + 0, + ts, + WalFileSequenceNumber::new(1337), + 2 + ), + ObjPath::from("my_host/dbs/0/0/2038-01-19/03-14/0000001337-2.parquet") + ); +} + #[test] fn parquet_file_percent_encoded() { assert_eq!( diff --git a/influxdb3_write/src/write_buffer/queryable_buffer.rs b/influxdb3_write/src/write_buffer/queryable_buffer.rs index 6ed07c17a98..2c18d940c31 100644 --- a/influxdb3_write/src/write_buffer/queryable_buffer.rs +++ b/influxdb3_write/src/write_buffer/queryable_buffer.rs @@ -200,7 +200,17 @@ impl QueryableBuffer { let snapshot_chunks = table_buffer.snapshot(table_def, snapshot_details.end_time_marker); + // A chunk_time can yield multiple buffer chunks when a string/tag + // column crossed the Arrow varchar limit and the chunk split + // (table_buffer::buffer_chunk). Each chunk needs a distinct path: + // with a shared path the persist jobs race, the last PUT wins the + // object, and every job's size is recorded — stale records that + // fail reads with "Corrupt footer". + let mut chunk_ordinals: HashMap = HashMap::new(); for chunk in snapshot_chunks { + let ordinal_ref = chunk_ordinals.entry(chunk.chunk_time).or_insert(0); + let chunk_ordinal = *ordinal_ref; + *ordinal_ref += 1; let table_name = db_schema.table_id_to_name(table_id).expect("table exists"); let persist_job = PersistJob { @@ -208,12 +218,13 @@ impl QueryableBuffer { table_id: *table_id, table_name: Arc::clone(&table_name), chunk_time: chunk.chunk_time, - path: ParquetFilePath::new( + path: ParquetFilePath::new_with_chunk_ordinal( self.persister.node_identifier_prefix(), database_id.get(), table_id.get(), chunk.chunk_time, snapshot_details.last_wal_sequence_number, + chunk_ordinal, ), batch: chunk.record_batch, schema: chunk.schema, diff --git a/influxdb3_write/src/write_buffer/queryable_buffer/tests.rs b/influxdb3_write/src/write_buffer/queryable_buffer/tests.rs index 08b4ef7c4bb..aee4374496e 100644 --- a/influxdb3_write/src/write_buffer/queryable_buffer/tests.rs +++ b/influxdb3_write/src/write_buffer/queryable_buffer/tests.rs @@ -336,3 +336,155 @@ async fn snapshot_skips_deleted_table() { "Soft deleted table should not have persisted files" ); } + +#[tokio::test] +async fn split_chunks_persist_to_distinct_paths() { + // EAR6985 regression: when a chunk_time splits into multiple buffer chunks (a + // string/tag column crossing the Arrow varchar limit), each chunk must persist to + // its own parquet path. With a shared path the concurrent persist jobs overwrite + // one object while every job's size is recorded, leaving stale size records that + // fail reads with "Invalid Parquet file. Corrupt footer". + use crate::write_buffer::table_buffer::VarColMaxGuard; + + // 50-byte strings with a 99-byte limit: the second write forces a second chunk + // for the same chunk_time. + let _guard = VarColMaxGuard::new(99); + + let object_store: Arc = Arc::new(InMemory::new()); + let metrics = Arc::new(metric::Registry::default()); + let parquet_store = + ParquetStorage::new(Arc::clone(&object_store), StorageId::from("influxdb3")); + let exec = Arc::new(Executor::new_with_config_and_executor( + ExecutorConfig { + target_query_partitions: NonZeroUsize::new(1).unwrap(), + object_stores: [&parquet_store] + .into_iter() + .map(|store| (store.id(), Arc::clone(store.object_store()))) + .collect(), + metric_registry: Arc::clone(&metrics), + mem_pool_size: 1024 * 1024 * 1024, + per_query_mem_pool_config: PerQueryMemoryPoolConfig::Disabled, + heap_memory_limit: None, + }, + DedicatedExecutor::new_testing(), + )); + let runtime_env = exec.new_context().inner().runtime_env(); + register_iox_object_store(runtime_env, parquet_store.id(), Arc::clone(&object_store)); + register_current_runtime_for_io(); + + let time_provider = Arc::new(MockProvider::new(Time::from_timestamp_nanos(0))); + let catalog = Arc::new( + Catalog::new( + "hosta", + Arc::clone(&object_store), + Arc::clone(&time_provider) as _, + Default::default(), + ) + .await + .unwrap(), + ); + let persister = Arc::new(Persister::new( + Arc::clone(&object_store), + "hosta", + Arc::clone(&time_provider) as _, + None, + )); + let time_provider: Arc = time_provider; + + let queryable_buffer = QueryableBuffer::new(QueryableBufferArgs { + executor: Arc::clone(&exec), + catalog: Arc::clone(&catalog), + persister: Arc::clone(&persister), + last_cache_provider: LastCacheProvider::new_from_catalog(Arc::clone(&catalog)) + .await + .unwrap(), + distinct_cache_provider: DistinctCacheProvider::new_from_catalog( + Arc::clone(&time_provider), + Arc::clone(&catalog), + ) + .await + .unwrap(), + persisted_files: Arc::new(PersistedFiles::new(None)), + parquet_cache: None, + parquet_snapshot_concurrency_limit: NonZeroUsize::new(10).unwrap(), + }); + + let db = DatabaseName::new("testdb").unwrap(); + + // Two writes into the SAME gen1 block, each with a 50-byte string: the second + // buffer_chunk call crosses the 99-byte limit and starts a second chunk for the + // same chunk_time. Two ops => two buffer_chunk calls. + let mut ops = vec![]; + for (tag, ts) in [("a", 1_i64), ("b", 2_i64)] { + let lp = format!("foo,t1={tag} f1=\"{}\" {ts}", "x".repeat(50)); + let val = WriteValidator::initialize(db.clone(), Arc::clone(&catalog)).unwrap(); + let lines = val + .v1_parse_lines_and_catalog_updates( + &lp, + false, + time_provider.now(), + Precision::Nanosecond, + ) + .unwrap() + .commit_catalog_changes() + .await + .unwrap() + .unwrap_success() + .convert_lines_to_buffer(Gen1Duration::new_1m()); + let batch: WriteBatch = lines.into(); + ops.push(WalOp::Write(batch)); + } + let wal_contents = WalContents { + persist_timestamp_ms: 0, + min_timestamp_ns: 1, + max_timestamp_ns: 2, + wal_file_number: WalFileSequenceNumber::new(1), + ops, + snapshot: None, + }; + let end_time = + wal_contents.max_timestamp_ns + Gen1Duration::new_1m().as_duration().as_nanos() as i64; + + let snapshot_details = SnapshotDetails { + snapshot_sequence_number: SnapshotSequenceNumber::new(1), + end_time_marker: end_time, + first_wal_sequence_number: WalFileSequenceNumber::new(1), + last_wal_sequence_number: WalFileSequenceNumber::new(1), + forced: false, + }; + let details = queryable_buffer + .notify_and_snapshot(Arc::new(wal_contents), snapshot_details) + .await; + let _details = details.await.unwrap(); + + let db_schema = catalog.db_schema("testdb").unwrap(); + let table = db_schema.table_definition("foo").unwrap(); + let files = queryable_buffer + .persisted_files + .get_files(db_schema.id, table.table_id); + + // Both split chunks must be persisted... + assert_eq!( + files.len(), + 2, + "expected two persisted files, got {files:?}" + ); + // ...at DISTINCT paths... + assert_ne!( + files[0].path, files[1].path, + "split chunks persisted to the same parquet path" + ); + // ...and every record's size must match the object it points at (the invariant + // the shared path violated). + for f in &files { + let head = object_store + .head(&object_store::path::Path::from(f.path.as_str())) + .await + .expect("persisted object exists"); + assert_eq!( + head.size as u64, f.size_bytes, + "recorded size differs from object size for {}", + f.path + ); + } +} diff --git a/influxdb3_write/src/write_buffer/table_buffer.rs b/influxdb3_write/src/write_buffer/table_buffer.rs index a99b3ea8d2a..8708a12edef 100644 --- a/influxdb3_write/src/write_buffer/table_buffer.rs +++ b/influxdb3_write/src/write_buffer/table_buffer.rs @@ -259,11 +259,11 @@ fn var_col_max_bytes() -> usize { #[cfg(test)] #[derive(Debug)] -struct VarColMaxGuard(usize); +pub(crate) struct VarColMaxGuard(usize); #[cfg(test)] impl VarColMaxGuard { - fn new(cap: usize) -> Self { + pub(crate) fn new(cap: usize) -> Self { let prev = TEST_VAR_COL_MAX_BYTES.with(|c| { let prev = c.get(); c.set(cap);