Skip to content

[adapters] size the datafusion sort reservation from the pool #6870

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
ryzhyk wants to merge 1 commit into
base: main
Choose a base branch
from
Open

[adapters] size the datafusion sort reservation from the pool #6870

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
474 changes: 396 additions & 78 deletions crates/adapterlib/src/utils/datafusion.rs
View file
Open in desktop

Large diffs are not rendered by default.

8 changes: 5 additions & 3 deletions crates/adapters/src/integrated/delta_table/input.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
8 changes: 4 additions & 4 deletions crates/feldera-types/src/config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
///
Expand Down
4 changes: 2 additions & 2 deletions openapi.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
26 changes: 0 additions & 26 deletions python/feldera/testutils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@

import base64
import logging
import math
import os
import urllib.error
import urllib.parse
Expand Down Expand Up @@ -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."
Expand Down
21 changes: 7 additions & 14 deletions python/tests/runtime/test_udp.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
),
Expand Down
14 changes: 3 additions & 11 deletions python/tests/workloads/test_now.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)
Expand Down Expand Up @@ -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,
Expand Down
Loading