Skip to content

Error Handling

Scalar UDFs can also set @geneva.udf(timeout=...) to enforce a per-row timeout in seconds. Timeout failures surface as ordinary TimeoutError, so existing on_error, retry, and skip_on_error() behavior applies unchanged.

This timeout support is scalar-only and uses Unix SIGALRM / setitimer semantics at runtime. It also requires execution on the worker process main thread. It works inside MultiProcessBatchApplier worker subprocesses because each child executes batches on its own main thread, but it still uses process-global signal state. UDFs or libraries that install their own SIGALRM handlers/timers, or that depend on main-thread signal behavior while coordinating work in background threads, may be incompatible.

Fatal Worker Errors

UDF backfill also classifies fatal worker loss into public Geneva exception types:

  • FatalWorkerTransientError
  • FatalWorkerOOMError
  • FatalWorkerCrashError
  • FatalWorkerExitError

By default, fatal worker errors use a driver-side retry-then-skip policy. The current unfinished task is attempted up to 3 times in total. If those attempts are exhausted, Geneva bisects the task until it isolates each failing row, writes NULL only for those rows, and continues the job. The bisected children retain the exhausted attempt count rather than starting a new retry budget. When error logging is enabled, Geneva records one error per failed row, and filtered backfills preserve successful selected values as well as existing values outside the filter. This driver-only default does not change the fail-fast behavior of ordinary UDF exceptions.

The retry rule matches FatalWorkerError, so newly introduced fatal-worker subclasses receive the default recovery behavior without an allowlist update. The exceptions are: FatalWorkerCrashError skips retry but still uses row isolation; FatalWorkerOOMError keeps the separate shrinking recovery described below; and deterministic integrity failures (ShortFragmentWriteError, CorruptCheckpointError, CheckpointCoverageError, and MergeFallbackTargetError) remain job-fatal by default.

FatalWorkerOOMError is also recovered by default: the unfinished part of the failing task is split into balanced tasks and resubmitted, so each retry runs with a smaller peak memory footprint. The target fanout follows the current ActorPool submission capacity, with a minimum of 2 and an unfinished-row cap: 2 below 1,000 rows, 10 from 1,000 through 999,999 rows, and 100 from 1,000,000 rows onward. A saturated pool therefore starts with a binary split, while idle actors can consume a wider long-tail split without creating unnecessarily tiny tasks. The pipeline refills all available actor slots after a split, while keeping at most one task queued behind the actors. Retries that strictly shrink every unfinished window are progress and proceed freely. A non-shrinking replacement for a multi-row task is rejected immediately, while a retry with no uncovered replacement is charged against the bounded retry budget. Once a task reaches a single-row window and can no longer shrink, each retry is also charged against the job-level budget (default: 10 such retries per job, 3 for the same exact range), and exhausting it fails the job fast with an "OOM recovery budget exceeded" error. The budget is configured through geneva_oom_recovery_budget settings (enabled, max_total_oom_recoveries, max_same_range_oom_recoveries, and the optional fixed target_split_fanout override). Setting enabled=false restores the previous fail-fast behavior for unhandled OOMs.

If the user provides on_error and does not explicitly match a fatal worker-loss type, Geneva still applies the corresponding default behavior. If the user does explicitly match one, the user configuration wins for that error type; for example, Fail(...) restores immediate job failure. Explicit fail-fast policies such as fail_fast() or an empty matcher list also restore immediate job failure for every exception.

Examples:

from geneva import Retry, Fail, udf
from geneva.errors import FatalWorkerTransientError

@udf(
    data_type=pa.int64(),
    on_error=[Retry(FatalWorkerTransientError, max_attempts=5)],
)
def retry_infra_failures(x: int) -> int:
    return x * 2


@udf(
    data_type=pa.int64(),
    on_error=[Fail(FatalWorkerTransientError)],
)
def fail_on_infra_failures(x: int) -> int:
    return x * 2

Exception Matchers

geneva.debug.error_store.Retry

Bases: ExceptionMatcher

Retry on matching exceptions with backoff

Parameters:

  • *exceptions (type[Exception], default: () ) –

    Exception types to match

  • match (str, default: None ) –

    Regex pattern to match in exception message. Simple strings work as substring matches (e.g., "rate limit"). Use (?i) for case-insensitive matching.

  • max_attempts (int, default: 3 ) –

    Maximum number of attempts (default: 3)

  • backoff (str, default: 'exponential' ) –

    Backoff strategy: "exponential" (default), "fixed", or "linear"

Examples:

Retry(ConnectionError, TimeoutError, max_attempts=3)
Retry(ValueError, match="rate limit", max_attempts=5)
Retry(APIError, match=r"429|rate.?limit")
Retry(APIError, match=r"(?i)rate limit")  # case-insensitive

max_attempts 

max_attempts: int = field(default=3)

backoff 

backoff: str = field(
    default="exponential", validator=_validate_backoff
)

geneva.debug.error_store.Skip

Bases: ExceptionMatcher

Skip row (return None) on matching exceptions

Parameters:

  • *exceptions (type[Exception], default: () ) –

    Exception types to match

  • match (str, default: None ) –

    Regex pattern to match in exception message

  • max_skip_count (int, default: None ) –

    Maximum number of rows that can be skipped before failing the job. Only used when passed via skip_on_error().

  • max_skip_fraction (float, default: None ) –

    Maximum fraction of rows (0.0-1.0) that can be skipped before failing the job. Only used when passed via skip_on_error().

Examples:

Skip(ValueError, KeyError)
Skip(ValueError, match="invalid input")

max_skip_count 

max_skip_count: int | None = field(default=None)

max_skip_fraction 

max_skip_fraction: float | None = field(default=None)

geneva.debug.error_store.Fail

Bases: ExceptionMatcher

Fail job immediately on matching exceptions

Parameters:

  • *exceptions (type[Exception], default: () ) –

    Exception types to match

  • match (str, default: None ) –

    Regex pattern to match in exception message

Examples:

Fail(AuthError)
Fail(ValueError, match="fatal")

Helper Functions

geneva.debug.error_store.retry_transient 

retry_transient(
    max_attempts: int = 3, backoff: str = "exponential"
) -> list[ExceptionMatcher]

Retry transient network errors (ConnectionError, TimeoutError, OSError).

Parameters:

  • max_attempts (int, default: 3 ) –

    Maximum number of attempts (default: 3)

  • backoff (str, default: 'exponential' ) –

    Backoff strategy: "exponential" (default), "fixed", or "linear"

Returns:

  • list[ExceptionMatcher] –

    Matcher list for use with on_error parameter

Examples:

@udf(data_type=pa.int32(), on_error=retry_transient())
@udf(data_type=pa.int32(), on_error=retry_transient(max_attempts=5))

geneva.debug.error_store.retry_all 

retry_all(
    max_attempts: int = 3, backoff: str = "exponential"
) -> list[ExceptionMatcher]

Retry any exception.

Parameters:

  • max_attempts (int, default: 3 ) –

    Maximum number of attempts (default: 3)

  • backoff (str, default: 'exponential' ) –

    Backoff strategy: "exponential" (default), "fixed", or "linear"

Returns:

  • list[ExceptionMatcher] –

    Matcher list for use with on_error parameter

Examples:

@udf(data_type=pa.int32(), on_error=retry_all())
@udf(data_type=pa.int32(), on_error=retry_all(max_attempts=5))

geneva.debug.error_store.skip_on_error 

skip_on_error(
    max_skip_count: int | None = None,
    max_skip_fraction: float | None = None,
) -> list[ExceptionMatcher]

Skip (return None) for any exception.

Parameters:

  • max_skip_count (int, default: None ) –

    Maximum number of rows that can be skipped before the job fails. If both max_skip_count and max_skip_fraction are set, whichever threshold is hit first triggers failure.

  • max_skip_fraction (float, default: None ) –

    Maximum fraction of rows (0.0-1.0) that can be skipped before the job fails. The fraction is computed as skipped / total_processed_so_far.

Returns:

  • list[ExceptionMatcher] –

    Matcher list for use with on_error parameter

Examples:

@udf(data_type=pa.int32(), on_error=skip_on_error())
@udf(data_type=pa.int32(), on_error=skip_on_error(max_skip_count=100))
@udf(data_type=pa.int32(), on_error=skip_on_error(max_skip_fraction=0.05))
@udf(
    data_type=pa.int32(),
    on_error=skip_on_error(max_skip_count=100, max_skip_fraction=0.05),
)

geneva.debug.error_store.fail_fast 

fail_fast() -> list[ExceptionMatcher]

Fail immediately on any exception (default behavior).

Returns:

  • list[ExceptionMatcher] –

    Empty matcher list (no special handling)

Examples:

@udf(data_type=pa.int32(), on_error=fail_fast())

Configuration

geneva.debug.error_store.ErrorHandlingConfig

Configuration for UDF error handling behavior

retry_config 

retry_config: UDFRetryConfig = field(factory=no_retry)

fault_isolation 

fault_isolation: FaultIsolation = field(default=FAIL_BATCH)

log_errors 

log_errors: bool = field(default=True)

log_retry_attempts 

log_retry_attempts: bool = field(default=False)

max_skip_count 

max_skip_count: int | None = field(default=None)

max_skip_fraction 

max_skip_fraction: float | None = field(default=None)

validate_compatibility 

validate_compatibility(map_task) -> None

Validate that this error config is compatible with the given task

Parameters:

  • map_task –

    The MapTask to validate against

Raises:

  • ValueError –

    If the config is incompatible with the task

geneva.debug.error_store.UDFRetryConfig

Retry configuration for UDF execution using tenacity semantics

retry 

retry: retry_base = field(
    factory=lambda: retry_if_exception_type(())
)

stop 

stop: stop_base = field(
    factory=lambda: stop_after_attempt(1)
)

wait 

wait: wait_base = field(
    factory=lambda: wait_exponential(
        multiplier=1, min=1, max=60
    )
)

before_sleep 

before_sleep: Callable[[RetryCallState], None] | None = (
    field(default=None)
)

after_attempt 

after_attempt: Callable[[RetryCallState], None] | None = (
    field(default=None)
)

reraise 

reraise: bool = field(default=True)

no_retry 

no_retry() -> UDFRetryConfig

No retries - fail immediately (default behavior)

retry_transient 

retry_transient(max_attempts: int = 3) -> UDFRetryConfig

Retry common transient errors (network, timeouts)

Parameters:

  • max_attempts (int, default: 3 ) –

    Maximum number of attempts including the initial try

geneva.debug.error_store.ErrorRecord

UDF execution error record, stored in geneva_errors table

error_id 

error_id: str = field(factory=lambda: str(uuid4()))

error_type 

error_type: str = field()

error_message 

error_message: str = field()

error_trace 

error_trace: str = field()

job_id 

job_id: str = field()

table_uri 

table_uri: str = field()

table_name 

table_name: str = field()

table_version 

table_version: Optional[int] = field(default=None)

column_name 

column_name: str = field()

udf_name 

udf_name: str = field()

udf_version 

udf_version: str = field()

input_columns 

input_columns: Optional[list[str]] = field(
    default=None, metadata={"pa_type": list_(string())}
)

output_columns 

output_columns: Optional[list[str]] = field(
    default=None, metadata={"pa_type": list_(string())}
)

actor_id 

actor_id: Optional[str] = field(default=None)

fragment_id 

fragment_id: Optional[int] = field(default=None)

batch_index 

batch_index: int = field()

row_address 

row_address: Optional[int] = field(default=None)

attempt 

attempt: int = field(default=1)

max_attempts 

max_attempts: int = field(default=1)

bisect_depth 

bisect_depth: Optional[int] = field(default=None)

timestamp 

timestamp: datetime = field(
    factory=dt_now_utc,
    metadata={"pa_type": timestamp("us", tz="UTC")},
)