diff --git a/crates/adapterlib/src/utils/datafusion.rs b/crates/adapterlib/src/utils/datafusion.rs index 70fb60f14ca..6192fea2650 100644 --- a/crates/adapterlib/src/utils/datafusion.rs +++ b/crates/adapterlib/src/utils/datafusion.rs @@ -4,7 +4,7 @@ use arrow::array::Array; use datafusion::common::ScalarValue; use datafusion::common::arrow::array::{AsArray, RecordBatch}; use datafusion::execution::SessionStateBuilder; -use datafusion::execution::memory_pool::FairSpillPool; +use datafusion::execution::memory_pool::{FairSpillPool, MemoryLimit}; use datafusion::execution::runtime_env::{RuntimeEnv, RuntimeEnvBuilder}; use datafusion::logical_expr::sqlparser::parser::ParserError; use datafusion::prelude::{SQLOptions, SessionConfig, SessionContext}; @@ -22,30 +22,104 @@ use std::io::Error as IoError; use std::ops::ControlFlow; use std::path::{Path, PathBuf}; use std::sync::Arc; -use tracing::warn; +use tracing::{info, warn}; -/// In-memory sort threshold; above this, sorts spill to disk. 64 MiB. +/// Below this much buffered data, a sort partition concatenates its batches +/// and sorts them in place instead of sorting each batch and merging. 64 MiB. /// /// Powers of two align with page sizes (4 KiB / 2 MiB) the allocator /// hands back, so a `1 << 26` budget matches what the OS actually /// reserves rather than a round decimal value the OS rounds up anyway. const SORT_IN_PLACE_THRESHOLD_BYTES: usize = 1 << 26; -/// Memory withheld from the sort phase for the merge phase to use. 64 MiB. +/// Ceiling on the merge headroom reserved for spill-to-disk sorting per sort partition. 64 MiB. /// -/// Reserved per partition: a sort with N partitions pre-allocates -/// `N * SORT_SPILL_RESERVATION_BYTES` from the pool. -/// If the pool can't satisfy that, the query fails immediately -/// with `Resources exhausted`. `create_runtime_env` emits a startup warning -/// when the configured pool is below `workers * SORT_SPILL_RESERVATION_BYTES`. +/// Normally, we reserve a fraction (SORT_RESERVATION_POOL_DIVISOR) of the +/// datafusion memory pool divided by the number of partitions. Very large values +/// can waste memory, because datafusion will grab the entire reservation even +/// when sorting a few entries. Very large reservations are also unnecessary. +/// This constant bounds the reservation. /// /// Note: DataFusion 52.x emits noisy `WARN datafusion_physical_plan::spill: /// Record batch memory usage ... exceeds the expected limit ... by more /// than the allowed tolerance` lines during spilled sorts. The overage is /// typically a handful of bytes over a 4 KB tolerance -- upstream /// accounting drift, tracked at -/// Not a query failure -const SORT_SPILL_RESERVATION_BYTES: usize = 1 << 26; +/// Not a query failure. +const MAX_SORT_SPILL_RESERVATION_BYTES: usize = 1 << 26; + +/// A fraction of the datafusion memory pool used for spill sorting. +const SORT_RESERVATION_POOL_DIVISOR: usize = 4; + +/// Merge headroom to reserve per sort partition for spill-to-disk merge sorting, in bytes. +/// +/// DataFusion pre-books this much for *every* partition of *every* sort, on +/// the partition's first batch and before it compares a single row +/// (`ExternalSorter::reserve_memory_for_merge`). +/// +/// We derive it as a fraction of the pool size `pool / SORT_RESERVATION_POOL_DIVISOR` +/// leaving the rest for data. +fn sort_spill_reservation_bytes(pool_bytes: Option, partitions: usize) -> usize { + let Some(pool_bytes) = pool_bytes else { + return MAX_SORT_SPILL_RESERVATION_BYTES; + }; + (pool_bytes / SORT_RESERVATION_POOL_DIVISOR / partitions.max(1)) + .min(MAX_SORT_SPILL_RESERVATION_BYTES) +} + +/// Per-partition merge headroom below which sorting partitions in parallel +/// is not worth its risk. DataFusion's own default +/// `sort_spill_reservation_bytes`, and just above the point where parallel +/// sorts were measured to start failing. +const MIN_PARALLEL_SORT_RESERVATION_BYTES: usize = 10 * 1024 * 1024; + +/// Whether the pool can fund a merge sort. +/// +/// True if the reservation prescribed by `sort_spill_reservation_bytes` is +/// `>= MIN_PARALLEL_SORT_RESERVATION_BYTES`. +fn parallel_sort_fits_pool(pool_bytes: Option, partitions: usize) -> bool { + // One partition has nothing to coalesce, and an unbounded pool has + // nothing to run out of. + if partitions <= 1 || pool_bytes.is_none() { + return true; + } + sort_spill_reservation_bytes(pool_bytes, partitions) >= MIN_PARALLEL_SORT_RESERVATION_BYTES +} + +/// Size of the pool the sorts will actually charge, or `None` when no memory +/// budget is configured and the pool is unbounded. +/// +/// Read from the [`RuntimeEnv`] rather than re-derived from the pipeline +/// config so the number stays right for a caller that built its own env. +fn pool_bytes(runtime_env: &RuntimeEnv) -> Option { + match runtime_env.memory_pool.memory_limit() { + MemoryLimit::Finite(bytes) => Some(bytes), + MemoryLimit::Infinite | MemoryLimit::Unknown => None, + } +} + +/// Upper bound on the sort partitions a query in this session can run +/// concurrently, and so on the number of merge reservations it books. +/// +/// `target_partitions` bounds every scan DataFusion plans itself, which is +/// all a connector session runs. Only the ad-hoc engine escapes it, and it +/// passes its own `partition_floor`; see [`adhoc_partition_floor`]. +fn sort_partitions(target_partitions: usize, partition_floor: usize) -> usize { + target_partitions.max(partition_floor).max(1) +} + +/// Sort partitions the ad-hoc engine can open regardless of +/// `target_partitions`. +/// +/// `AdHocQueryExecution` reports one partition per reader in the snapshot, +/// and the snapshot holds one batch per *local* DBSP worker +/// (`OutputHandle::take_from_all` walks the local mailboxes). `workers` +/// counts the workers on every host, while the memory pool is per host, so +/// the local share is what a sort here actually runs. +fn adhoc_partition_floor(pipeline_config: &PipelineConfig) -> usize { + let workers = pipeline_config.global.workers as usize; + workers.div_ceil(pipeline_config.global.hosts.max(1)) +} /// Build the shared datafusion [`RuntimeEnv`] for a pipeline. /// @@ -59,7 +133,6 @@ pub fn create_runtime_env( if let Some(datafusion_memory_mb) = pipeline_config.global.resolved_datafusion_memory_mb() { let memory_bytes_max = datafusion_memory_mb * 1_000_000; builder = builder.with_memory_pool(Arc::new(FairSpillPool::new(memory_bytes_max as usize))); - warn_if_pool_too_small_for_adhoc_sort(pipeline_config, datafusion_memory_mb); } if let Some(storage) = &pipeline_config.storage_config { let path = PathBuf::from(storage.path.clone()).join(DATAFUSION_TEMP_DIR); @@ -83,55 +156,6 @@ pub fn create_runtime_env( }) } -/// Minimum DataFusion pool size, in MB, that can satisfy the ad-hoc -/// engine's per-partition sort reservation given `workers`. -/// -/// Ad-hoc sessions set `target_partitions = workers` -/// (see [`create_session_context`]), so an `ORDER BY` (or any other -/// sort-based operator) reserves `workers * SORT_SPILL_RESERVATION_BYTES` -/// from the pool *before* sorting any rows. The reservation is in -/// binary MiB (`1 << 26`); the pool is sized from the user-facing -/// `datafusion_memory_mb` (decimal MB). Compare in bytes, then -/// ceil-divide to MB so the warning's threshold is never lower than -/// the actual byte requirement. -fn min_pool_mb_for_adhoc_sort(workers: u64) -> u64 { - let needed_bytes = (SORT_SPILL_RESERVATION_BYTES as u64).saturating_mul(workers); - needed_bytes.div_ceil(1_000_000) -} - -/// Warn at startup when the DataFusion pool is too small to satisfy the -/// per-partition sort reservation for the ad-hoc query engine. -/// -/// If the pool can't satisfy that, the query fails on the first reservation -/// attempt with `Resources exhausted`. Surface this as a single startup -/// warning so the failure mode isn't silent. Connector sessions can override -/// `target_partitions`; their reservation budget is not checked here. -fn warn_if_pool_too_small_for_adhoc_sort(pipeline_config: &PipelineConfig, pool_mb: u64) { - let workers = pipeline_config.global.workers as u64; - // Degenerate configs (tests / synthetic) report `workers == 0`; nothing - // useful to say in that case and the message would print "0 MB". - if workers == 0 { - return; - } - let min_pool_mb = min_pool_mb_for_adhoc_sort(workers); - // `<=` not `<`: at exact equality every partition's reservation sums to - // the full pool with zero headroom. FairSpillPool's internal accounting - // takes a few bytes of overhead, so the last partition's reservation - // fails by a fraction of a MB. Empirically: pool=256 / workers=4 - // fails; 257 succeeds. - if pool_mb <= min_pool_mb { - let per_worker_mb = min_pool_mb_for_adhoc_sort(1); - warn!( - "DataFusion memory pool is {pool_mb} MB; sort-heavy ad-hoc \ - queries (ORDER BY, EXCEPT, hash joins) need at least \ - {min_pool_mb} MB ({workers} workers x {per_worker_mb} MB \ - reservation per worker). Such queries may fail at first \ - allocation with 'Resources exhausted'. Increase \ - 'datafusion_memory_mb' or reduce 'workers'." - ); - } -} - /// Remove leftovers from a previous process inside the scratch directory. /// /// DataFusion's `DiskManager` leaks its `datafusion-XXXXXX/` subdir if the @@ -186,22 +210,48 @@ fn clean_stale_scratch_entries(scratch_dir: &Path) { } } -/// `SessionContext` bound to the shared [`RuntimeEnv`], configured with the -/// pipeline's worker count and feldera's sort-spill thresholds. +/// `SessionContext` for the ad-hoc query engine, bound to the shared +/// [`RuntimeEnv`] and configured with feldera's sort-spill thresholds. +/// +/// Budgets sorts for one partition per local DBSP worker, since that is what +/// the ad-hoc scan opens whatever `target_partitions` says. pub fn create_session_context( pipeline_config: &PipelineConfig, runtime_env: Arc, ) -> SessionContext { - create_session_context_with(pipeline_config, runtime_env, |cfg| cfg) + let floor = adhoc_partition_floor(pipeline_config); + create_session_context_inner(pipeline_config, runtime_env, floor, |cfg| cfg) } -/// Like [`create_session_context`], with a hook to override individual +/// `SessionContext` for a connector, with a hook to override individual /// datafusion settings (e.g. parquet decoding) before the context is built. +/// +/// Sorts are budgeted for `target_partitions`, which bounds every scan +/// DataFusion plans. Unlike [`create_session_context`] there is no worker +/// floor: a connector session queries external tables, never the ad-hoc +/// snapshot, so the DBSP worker count does not bound its parallelism. +/// +/// The hook may raise or lower `target_partitions`. `sort_spill_reservation_bytes` +/// and `repartition_sorts` are derived from whatever partition count the hook +/// settles on and applied afterwards, so a hook that sets either of those two +/// options does not take effect. pub fn create_session_context_with( pipeline_config: &PipelineConfig, runtime_env: Arc, customize_config: F, ) -> SessionContext +where + F: FnOnce(SessionConfig) -> SessionConfig, +{ + create_session_context_inner(pipeline_config, runtime_env, 0, customize_config) +} + +fn create_session_context_inner( + pipeline_config: &PipelineConfig, + runtime_env: Arc, + partition_floor: usize, + customize_config: F, +) -> SessionContext where F: FnOnce(SessionConfig) -> SessionConfig, { @@ -212,13 +262,32 @@ where let session_config = SessionConfig::new() .with_target_partitions(workers as usize) .with_sort_in_place_threshold_bytes(SORT_IN_PLACE_THRESHOLD_BYTES) - .with_sort_spill_reservation_bytes(SORT_SPILL_RESERVATION_BYTES) .set( "datafusion.execution.planning_concurrency", &ScalarValue::UInt64(Some(workers)), ); let session_config = customize_config(session_config); + let partitions = sort_partitions(session_config.target_partitions(), partition_floor); + let pool_bytes = pool_bytes(&runtime_env); + let repartition_sorts = parallel_sort_fits_pool(pool_bytes, partitions); + if !repartition_sorts { + info!( + "DataFusion memory pool is {} MB across {partitions} sort partitions, too little to \ + sort them in parallel; sorts will coalesce their input first. This is slower but \ + lets a sort use the whole pool. Increase 'datafusion_memory_mb' or reduce 'workers' \ + to sort in parallel again.", + pool_bytes.unwrap_or(0) / 1_000_000, + ); + } + // The reservation stays sized for `partitions` even when the `repartition_sorts` + // flag is off. This flag does not govern sorts that arrive already + // partitioned (window functions, sort-merge joins), so a plan + // can still open one sorter per partition. + let session_config = session_config + .with_sort_spill_reservation_bytes(sort_spill_reservation_bytes(pool_bytes, partitions)) + .with_repartition_sorts(repartition_sorts); + let mut state = SessionStateBuilder::new() .with_config(session_config) .with_runtime_env(runtime_env) @@ -792,19 +861,268 @@ mod tests { ); } - /// Pins the boundary that drives the `warn_if_pool_too_small_for_adhoc_sort` - /// log line. If `SORT_SPILL_RESERVATION_BYTES` changes, the warning - /// threshold changes with it + /// Every partition books its merge headroom before reading a row, so the + /// aggregate has to fit the pool whatever the partition count is. This is + /// the property a fixed 64 MiB violated: 8 partitions booked 537 MB + /// against the 483 MB pool of a 9 GiB pipeline. #[test] - fn min_pool_mb_for_adhoc_sort_matches_reservation_times_workers() { - use super::min_pool_mb_for_adhoc_sort; - // SORT_SPILL_RESERVATION_BYTES is 64 MiB = 67_108_864 B; the - // resolved pool size is reported in decimal MB, so each worker's - // requirement ceil-divides to 68 MB. - assert_eq!(min_pool_mb_for_adhoc_sort(0), 0); - assert_eq!(min_pool_mb_for_adhoc_sort(1), 68); - assert_eq!(min_pool_mb_for_adhoc_sort(2), 135); - assert_eq!(min_pool_mb_for_adhoc_sort(8), 537); + fn sort_reservation_fits_the_pool() { + use super::{ + MAX_SORT_SPILL_RESERVATION_BYTES, SORT_RESERVATION_POOL_DIVISOR, + sort_spill_reservation_bytes, + }; + + let mut failures = Vec::new(); + for pool_mb in [12u64, 25, 100, 256, 483, 800, 2_000] { + for partitions in [1usize, 2, 4, 8, 16, 32, 64, 128] { + let pool = (pool_mb * 1_000_000) as usize; + let aggregate = sort_spill_reservation_bytes(Some(pool), partitions) * partitions; + let budget = pool / SORT_RESERVATION_POOL_DIVISOR; + if aggregate > budget { + failures.push(format!( + "pool={pool_mb} MB partitions={partitions}: {aggregate} B booked \ + exceeds the {budget} B budget" + )); + } + } + } + assert!(failures.is_empty(), "{}", failures.join("\n")); + + // Exact values, so an implementation that simply hands out nothing + // cannot satisfy the inequality above. + assert_eq!( + sort_spill_reservation_bytes(Some(483_000_000), 8), + 483_000_000 / 4 / 8, + ); + // A small pool shared by many partitions thins the reservation rather + // than clamping it up; clamping up is what would reintroduce + // `partitions * floor > pool`. + assert_eq!(sort_spill_reservation_bytes(Some(12_000_000), 64), 46_875); + // Degenerate partition counts must not divide by zero. + assert_eq!(sort_spill_reservation_bytes(Some(12_000_000), 0), 3_000_000); + // A pool that can fund the ceiling gets it, and an unbounded pool has + // nothing to divide. + assert_eq!( + sort_spill_reservation_bytes(Some(4_000_000_000), 8), + MAX_SORT_SPILL_RESERVATION_BYTES, + ); + assert_eq!( + sort_spill_reservation_bytes(None, 64), + MAX_SORT_SPILL_RESERVATION_BYTES, + ); + } + + /// The partition count a session budgets for has to be the one its + /// queries actually run, which differs between the two kinds of session. + /// Both settings derived from it are checked here, since both are wrong + /// together when the count is. + #[test] + fn sessions_budget_for_the_partitions_they_run() { + use super::create_session_context_with; + + /// `hosts` and `io_workers` are the two knobs that separate the + /// partition count from `workers`; `connector` picks the session kind. + struct Case { + name: &'static str, + connector: bool, + workers: u16, + hosts: usize, + io_workers: Option, + pool_mb: u64, + /// `target_partitions`, which bounds a connector's scans. + target_partitions: usize, + /// Partitions the session budgets sorts for. + sort_partitions: usize, + repartition_sorts: bool, + } + + let cases = [ + // The reported failure: `io_workers` unset, so `target_partitions` + // already equals the worker count and 483 MB funds 8 sorters. + Case { + name: "adhoc, defaults", + connector: false, + workers: 8, + hosts: 1, + io_workers: None, + pool_mb: 483, + target_partitions: 8, + sort_partitions: 8, + repartition_sorts: true, + }, + // The ad-hoc scan opens one partition per worker whatever + // `target_partitions` says, so a lower `io_workers` must not + // shrink the budget. + Case { + name: "adhoc, io_workers below workers", + connector: false, + workers: 16, + hosts: 1, + io_workers: Some(2), + pool_mb: 480, + target_partitions: 2, + sort_partitions: 16, + repartition_sorts: false, + }, + // A higher `io_workers` raises what DataFusion may repartition to. + Case { + name: "adhoc, io_workers above workers", + connector: false, + workers: 4, + hosts: 1, + io_workers: Some(12), + pool_mb: 480, + target_partitions: 12, + sort_partitions: 12, + repartition_sorts: false, + }, + // `workers` counts every host while the pool and the snapshot are + // per host, so a host budgets for its own share. Budgeting for all + // 32 would coalesce a sort this pool funds. + Case { + name: "adhoc, four hosts", + connector: false, + workers: 32, + hosts: 4, + io_workers: Some(4), + pool_mb: 480, + target_partitions: 4, + sort_partitions: 8, + repartition_sorts: true, + }, + // A connector queries external tables, never the ad-hoc snapshot, + // so nothing there opens a partition per worker. + Case { + name: "connector, io_workers below workers", + connector: true, + workers: 16, + hosts: 1, + io_workers: Some(2), + pool_mb: 480, + target_partitions: 2, + sort_partitions: 2, + repartition_sorts: true, + }, + ]; + + for case in cases { + let storage = TempStorage::new(&format!("feldera-df-session-{}", case.name)); + let cfg = pipeline_config( + RuntimeConfig { + workers: case.workers, + hosts: case.hosts, + io_workers: case.io_workers, + datafusion_memory_mb: Some(case.pool_mb), + max_rss_mb: Some(64_000), + ..Default::default() + }, + Some(storage.path()), + ); + let env = create_runtime_env(&cfg).unwrap(); + let ctx = if case.connector { + create_session_context_with(&cfg, env, |c| c) + } else { + create_session_context(&cfg, env) + }; + + let config = ctx.copied_config(); + let name = case.name; + assert_eq!( + config.target_partitions(), + case.target_partitions, + "{name}: target_partitions", + ); + // The expected partition count is stated, not re-derived from the + // production helper: deriving it would move expectation and actual + // together and the assertion could never fail. + assert_eq!( + config.options().execution.sort_spill_reservation_bytes, + case.pool_mb as usize * 1_000_000 / 4 / case.sort_partitions, + "{name}: reservation must divide by {} partitions", + case.sort_partitions, + ); + assert_eq!( + config.repartition_sorts(), + case.repartition_sorts, + "{name}: repartition_sorts", + ); + } + } + + /// End to end through a real plan: an `ORDER BY` over one row per worker, + /// on a pool smaller than a single fixed 64 MiB reservation. Catches a + /// reservation that is computed but never reaches the session, which the + /// unit tests above cannot see. + #[test] + fn order_by_succeeds_on_a_pool_smaller_than_the_old_reservation() { + use arrow::array::Int64Array; + use arrow::datatypes::{DataType, Field as ArrowField, Schema}; + use datafusion::common::arrow::array::RecordBatch; + use datafusion::datasource::MemTable; + use std::sync::Arc; + + const WORKERS: usize = 8; + let storage = TempStorage::new("feldera-datafusion-order-by-small-pool-test"); + let cfg = pipeline_config( + RuntimeConfig { + workers: WORKERS as u16, + // Below one 64 MiB reservation, so the first partition used + // to fail before comparing a row. + datafusion_memory_mb: Some(12), + max_rss_mb: Some(16_000), + ..Default::default() + }, + Some(storage.path()), + ); + let ctx = create_session_context(&cfg, create_runtime_env(&cfg).unwrap()); + + // One partition per DBSP worker, mirroring `AdHocQueryExecution`. + let schema = Arc::new(Schema::new(vec![ArrowField::new( + "cnt", + DataType::Int64, + false, + )])); + let partitions: Vec> = (0..WORKERS) + .map(|worker| { + vec![ + RecordBatch::try_new( + schema.clone(), + vec![Arc::new(Int64Array::from(vec![worker as i64]))], + ) + .unwrap(), + ] + }) + .collect(); + ctx.register_table( + "q1", + Arc::new(MemTable::try_new(schema, partitions).unwrap()), + ) + .unwrap(); + + let batches = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(super::execute_query_collect( + &ctx, + "select * from q1 order by cnt desc", + )) + .unwrap(); + // Assert the ordering, not just the row count: a plan that dropped + // the sort would return the same rows. + let sorted: Vec = batches + .iter() + .flat_map(|batch| { + batch + .column(0) + .as_any() + .downcast_ref::() + .expect("cnt column is Int64") + .values() + .to_vec() + }) + .collect(); + assert_eq!(sorted, (0..WORKERS as i64).rev().collect::>()); } /// Make sure random shapes for `filter`, `cdc_delete_filter`, and `cdc_order_by` diff --git a/crates/adapters/src/integrated/delta_table/input.rs b/crates/adapters/src/integrated/delta_table/input.rs index 97636cc89de..c1323e5b221 100644 --- a/crates/adapters/src/integrated/delta_table/input.rs +++ b/crates/adapters/src/integrated/delta_table/input.rs @@ -122,9 +122,11 @@ fn format_datafusion_error( format!( "{base}\n\ DataFusion memory pool is exhausted. Current datafusion memory pool limit: {current_pool_limit}. \ - Consider setting or increasing 'datafusion_memory_mb' in the pipeline runtime config. \ - If raising the budget is not an option, reduce 'io_workers' / 'workers' or \ - set the env var 'DELTA_DF_TARGET_PARTITIONS=1' to lower per-scan parallelism.\n" + The query needs more memory than the pool holds; consider setting or increasing \ + 'datafusion_memory_mb' in the pipeline runtime config. If raising the budget is not \ + an option, reduce 'io_workers' / 'workers' or set the env var \ + 'DELTA_DF_TARGET_PARTITIONS=1' to scan fewer files at a time, which lowers how much \ + of the scan is held in memory at once.\n" ) } else { base diff --git a/crates/feldera-types/src/config.rs b/crates/feldera-types/src/config.rs index 6fe11d9683d..c604153006f 100644 --- a/crates/feldera-types/src/config.rs +++ b/crates/feldera-types/src/config.rs @@ -866,10 +866,10 @@ pub struct RuntimeConfig { /// Pipelines that don't run heavy ad-hoc / Delta / Iceberg workloads /// can leave this unset. /// - /// Sort/aggregate-heavy ad-hoc queries (especially at high `workers` - /// counts) should set this explicitly. An under-sized pool surfaces as - /// `ResourcesExhausted` on the failing query — the pipeline keeps - /// running and only that query fails. + /// Set this explicitly for ad-hoc queries, or Delta / Iceberg scans, + /// over data too large for the default share. A pool that cannot hold + /// the query surfaces as `ResourcesExhausted` on that query alone; the + /// pipeline keeps running. /// /// No pool limit applied if no overall budget is configured. /// diff --git a/openapi.json b/openapi.json index 2c7ede8458c..df5628ad856 100644 --- a/openapi.json +++ b/openapi.json @@ -12424,7 +12424,7 @@ "datafusion_memory_mb": { "type": "integer", "format": "int64", - "description": "DataFusion memory pool size, in MB, shared by the ad-hoc query\nengine and the Delta Lake / Iceberg connectors.\n\nCarved out of `max_rss_mb` (falling back to\n`resources.memory_mb_max`); the remainder goes to the DBSP circuit,\nso the two do not double-book RAM.\n\nUnset: defaults to 5% of the effective budget, capped at 2 GB.\nPipelines that don't run heavy ad-hoc / Delta / Iceberg workloads\ncan leave this unset.\n\nSort/aggregate-heavy ad-hoc queries (especially at high `workers`\ncounts) should set this explicitly. An under-sized pool surfaces as\n`ResourcesExhausted` on the failing query — the pipeline keeps\nrunning and only that query fails.\n\nNo pool limit applied if no overall budget is configured.\n\nSee [documentation on the pipeline's memory usage](https://docs.feldera.com/operations/memory)\nfor more details.", + "description": "DataFusion memory pool size, in MB, shared by the ad-hoc query\nengine and the Delta Lake / Iceberg connectors.\n\nCarved out of `max_rss_mb` (falling back to\n`resources.memory_mb_max`); the remainder goes to the DBSP circuit,\nso the two do not double-book RAM.\n\nUnset: defaults to 5% of the effective budget, capped at 2 GB.\nPipelines that don't run heavy ad-hoc / Delta / Iceberg workloads\ncan leave this unset.\n\nSet this explicitly for ad-hoc queries, or Delta / Iceberg scans,\nover data too large for the default share. A pool that cannot hold\nthe query surfaces as `ResourcesExhausted` on that query alone; the\npipeline keeps running.\n\nNo pool limit applied if no overall budget is configured.\n\nSee [documentation on the pipeline's memory usage](https://docs.feldera.com/operations/memory)\nfor more details.", "default": null, "nullable": true, "minimum": 0 @@ -14443,7 +14443,7 @@ "datafusion_memory_mb": { "type": "integer", "format": "int64", - "description": "DataFusion memory pool size, in MB, shared by the ad-hoc query\nengine and the Delta Lake / Iceberg connectors.\n\nCarved out of `max_rss_mb` (falling back to\n`resources.memory_mb_max`); the remainder goes to the DBSP circuit,\nso the two do not double-book RAM.\n\nUnset: defaults to 5% of the effective budget, capped at 2 GB.\nPipelines that don't run heavy ad-hoc / Delta / Iceberg workloads\ncan leave this unset.\n\nSort/aggregate-heavy ad-hoc queries (especially at high `workers`\ncounts) should set this explicitly. An under-sized pool surfaces as\n`ResourcesExhausted` on the failing query — the pipeline keeps\nrunning and only that query fails.\n\nNo pool limit applied if no overall budget is configured.\n\nSee [documentation on the pipeline's memory usage](https://docs.feldera.com/operations/memory)\nfor more details.", + "description": "DataFusion memory pool size, in MB, shared by the ad-hoc query\nengine and the Delta Lake / Iceberg connectors.\n\nCarved out of `max_rss_mb` (falling back to\n`resources.memory_mb_max`); the remainder goes to the DBSP circuit,\nso the two do not double-book RAM.\n\nUnset: defaults to 5% of the effective budget, capped at 2 GB.\nPipelines that don't run heavy ad-hoc / Delta / Iceberg workloads\ncan leave this unset.\n\nSet this explicitly for ad-hoc queries, or Delta / Iceberg scans,\nover data too large for the default share. A pool that cannot hold\nthe query surfaces as `ResourcesExhausted` on that query alone; the\npipeline keeps running.\n\nNo pool limit applied if no overall budget is configured.\n\nSee [documentation on the pipeline's memory usage](https://docs.feldera.com/operations/memory)\nfor more details.", "default": null, "nullable": true, "minimum": 0 diff --git a/python/feldera/testutils.py b/python/feldera/testutils.py index e3d6d9d5746..3460ddaeb01 100644 --- a/python/feldera/testutils.py +++ b/python/feldera/testutils.py @@ -2,7 +2,6 @@ import base64 import logging -import math import os import urllib.error import urllib.parse @@ -149,31 +148,6 @@ def feldera_bearer_token() -> Optional[str]: FELDERA_TEST_NUM_WORKERS = int(os.environ.get("FELDERA_TEST_NUM_WORKERS", "8")) FELDERA_TEST_NUM_HOSTS = int(os.environ.get("FELDERA_TEST_NUM_HOSTS", "1")) -# Ad-hoc queries reserve 64 MiB per worker for sort spill before sorting a -# single row (SORT_SPILL_RESERVATION_BYTES in -# adapterlib/src/utils/datafusion.rs), so a `datafusion_memory_mb` pool -# smaller than `workers * 64 MiB` fails any ORDER BY/EXCEPT/hash-join ad-hoc -# query with "Resources exhausted", independent of data size. This holds -# regardless of `hosts`: the reservation is sized from the *total* configured -# `workers`, whether they all run on one host or are split across several -# (the coordinator's own ad-hoc engine also uses the total worker count as -# its partition count -- see crates/coord/src/adhoc.rs). -_ADHOC_SORT_RESERVATION_MB_PER_WORKER = (64 * 1024 * 1024) / 1_000_000 - - -def min_datafusion_memory_mb( - workers: int = FELDERA_TEST_NUM_WORKERS, - hosts: int = FELDERA_TEST_NUM_HOSTS, - headroom_mb: int = 512, -) -> int: - """Minimum `datafusion_memory_mb` for a sort-heavy ad-hoc query to avoid - "Resources exhausted" with the given number of workers, plus - `headroom_mb` for the query's actual data. `hosts` doesn't change the - result (see module comment above); it's accepted so callers can pass - both test parameters without worrying about which one matters. - """ - return math.ceil(workers * _ADHOC_SORT_RESERVATION_MB_PER_WORKER) + headroom_mb - class _LazyClient: "Construct the FelderaClient only when accessed as opposed to when imported." diff --git a/python/tests/runtime/test_udp.py b/python/tests/runtime/test_udp.py index 4edb94f0fb1..7c38f0a3a7b 100644 --- a/python/tests/runtime/test_udp.py +++ b/python/tests/runtime/test_udp.py @@ -4,11 +4,7 @@ from tests import TEST_CLIENT from tests.platform.helper import PipelineTestCase from feldera.runtime_config import Resources, RuntimeConfig -from feldera.testutils import ( - FELDERA_TEST_NUM_WORKERS, - FELDERA_TEST_NUM_HOSTS, - min_datafusion_memory_mb, -) +from feldera.testutils import FELDERA_TEST_NUM_WORKERS, FELDERA_TEST_NUM_HOSTS # Test user-defined preprocessor @@ -101,22 +97,19 @@ def test_local(self): tracing = { version = "0.1.40" } """ - # Scales with FELDERA_TEST_NUM_WORKERS/_HOSTS so the ad-hoc ORDER BY - # query below doesn't fail with "Resources exhausted" if either - # changes. See min_datafusion_memory_mb. - datafusion_memory_mb = min_datafusion_memory_mb( - FELDERA_TEST_NUM_WORKERS, FELDERA_TEST_NUM_HOSTS - ) - pipeline = PipelineBuilder( TEST_CLIENT, name=self.register_for_cleanup("test_udps"), sql=sql, udf_rust=udfs, udf_toml=toml, + # A 512 MB pool across 8 workers is the shape that used to fail: + # the ad-hoc ORDER BY below pre-booked 64 MiB per worker, more + # than the whole pool, before reading a row. Keep it small so the + # test keeps covering that. runtime_config=RuntimeConfig( - datafusion_memory_mb=datafusion_memory_mb, - resources=Resources(memory_mb_min=datafusion_memory_mb + 512), + datafusion_memory_mb=512, + resources=Resources(memory_mb_min=1024), workers=FELDERA_TEST_NUM_WORKERS, hosts=FELDERA_TEST_NUM_HOSTS, ), diff --git a/python/tests/workloads/test_now.py b/python/tests/workloads/test_now.py index 2e98dcf31e8..395801f467c 100644 --- a/python/tests/workloads/test_now.py +++ b/python/tests/workloads/test_now.py @@ -5,12 +5,9 @@ from feldera.pipeline import Pipeline from feldera.runtime_config import Resources from feldera.testutils import ( - FELDERA_TEST_NUM_HOSTS, - FELDERA_TEST_NUM_WORKERS, ViewSpec, build_pipeline, log, - min_datafusion_memory_mb, validate_outputs, unique_pipeline_name, ) @@ -99,14 +96,9 @@ def test_now(self): ), ] - # 12288 MB covers this test's data volume (>2GB ad-hoc query storage, - # peaks above 5GB memory on arm64); take the max with the per-worker - # sort-reservation floor (see min_datafusion_memory_mb) so a future - # bump to FELDERA_TEST_NUM_WORKERS can't undersize it. - datafusion_memory_mb = max( - min_datafusion_memory_mb(FELDERA_TEST_NUM_WORKERS, FELDERA_TEST_NUM_HOSTS), - 12288, - ) + # 12288 MB covers this test's data volume (>2GB ad-hoc query + # storage, peaks above 5GB memory on arm64). + datafusion_memory_mb = 12288 pipeline = build_pipeline( unique_pipeline_name("now-test"), tables,