Skip to content

Operators

Every page in this section marks one Flink operator (or operator family) as native, partial (native with specific, enumerated gaps), or unsupported — see Unsupported operators for the operators with no native path at all. Together these pages are the precise answer to "why didn't my query accelerate?" — everything not called out as a gap here runs natively.

Sink constraints documents the shared columnar NOT NULL boundary, Doris DROP and string/binary length policies, and the connector-specific admission that remains necessary around it. Sink upsert materialization covers ordered candidate state before admitted Doris Arrow writers.

The all-or-nothing island

A query accelerates only if it forms one fully-columnar island: every operator but a rowwise source/sink runs natively, exchanging Arrow batches, with the row↔Arrow transpose paid once at each host edge and never between native operators. One unsupported interior operator drags the whole query back to Flink — there's no partial acceleration of a single query. Use NativePlanner.explain(...) or -Dstreamfusion.logFallbackReasons=true to see the recorded reason(s) for a given plan.

What counts as a fallback. A fallback is something Flink executes that StreamFusion doesn't accelerate — a real gap that could be closed. It is not a fallback when Flink itself rejects the query in streaming (e.g. RANK/DENSE_RANK Top-N, non-time ORDER BY) — matching Flink by also not running it is parity, not a gap.

Optional native probe failures

Native admission predicates and fallback diagnostic probes can decline an otherwise valid Flink operator when a class is unavailable (NoClassDefFoundError), a method or field signature is incompatible (NoSuchMethodError, NoSuchFieldError, AbstractMethodError), or a native library/symbol is unavailable (UnsatisfiedLinkError). The recorded reason distinguishes these failures. A broken entry yields to later entries for the same operator shape. When none can replace it, Flink retains the host plan and the ordinary whole-island rule applies.

Diagnostics reuse an admission failure recorded for that node and entry during the current plan rewrite; they do not repeat the failed probe. Diagnostic-only failures are contained at the same boundary. Disabled operators skip their admission and diagnostic probes. There is no process-wide failure cache: another planning attempt can succeed after deployment repair.

This boundary covers optional acceleration probes, not SQL validation, user-function execution, or construction of an admitted replacement. Those failures still propagate, as do fatal JVM errors and other linkage errors. Falling back preserves host planning; it does not repair or certify an incompatible native deployment.

Optional planner extension initialization

Optional connector extensions register into separate temporary rule lists. Only a fully registered extension publishes its rules and source-sharing hook. A service-provider or linkage failure during discovery or registration skips that extension without retaining partial rules or disabling healthy extensions and built-in operators. Registration warnings identify the provider and include the original exception; discovery warnings include ServiceLoader's provider diagnostic. These initialization warnings are separate from per-operator fallback reasons.

Ordinary programming exceptions and fatal JVM errors still propagate, including when wrapped by ServiceLoader. This isolation does not supply a missing stock connector or make an incompatible connector usable: a query that actually requires it must still meet Flink's connector requirements. Extensions are initialized once per planner classloader; repairing a skipped provider requires a fresh planner classloader. No registration work is added to batch processing.

Nested values at the Arrow boundary

Reusable row-to-Arrow writers finish and reset nested ARRAY, MAP and ROW children together with their parent. Reusing a writer for another batch restarts map key/value positions at zero, including maps nested inside arrays or rows; null and empty containers retain their distinct representations.

Timestamp values and event time

Timestamp readers expose Flink's signed epoch milliseconds plus a non-negative nanosecond remainder within the millisecond. For example, -1 nanosecond is (-1, 999999), not (0, -1). Flink BinaryRow key encoding reads these components directly, including nested timestamp keys; it does not multiply milliseconds into an i64 nanosecond count. Event-time readers for sort, window-aggregate input and watermarks use the millisecond component. The JVM temporal-function bridge preserves both components for generated expressions and reads milliseconds for millisecond-only builtins. Readers accept Arrow second, millisecond, microsecond and nanosecond timestamp columns without interpreting the Arrow timezone label as a timezone conversion. The default SQL layout is a nullable timestamp struct containing non-null millis: BIGINT and nano_of_milli: INT buffers. Component metadata distinguishes it from user ROW values. This keeps Flink's full signed-millisecond range and fractional nanos in projections, expressions, keys, windows and saved state. Its millisecond view shares the original buffer, and its writer preserves hidden fractions even when the logical precision is three. Explicit Arrow unit casts check overflow; connector formats follow their host's documented precision and physical-unit limits. See the timestamp contract.

Global switches

  • -Dstreamfusion.native.enabled=false — master switch; run entirely on Flink.
  • -Dstreamfusion.operator.<name>.enabled=false — keep one specific operator on the host. See Configuration for the full flag surface.
  • Insert-only guard — every operator except the changelog-aware ones (GROUP BY, regular join, a CDC source, Calc, UNION ALL, Expand, UNNEST, changelog normalize, DropUpdateBefore, streaming Top-N/LIMIT) requires an insert-only input; a retracting/updating input falls it back.
  • table.optimizer.delta-join.strategy = FORCE — no substitutions are made in an optimizer block containing an ordinary join and no delta join. Flink validates this strategy after our pass across all statement roots; removing the ordinary joins could hide its intended rejection. Blocks containing a delta join, or no ordinary join, are not rejected by this FORCE guard; ordinary admission and island checks still apply. DeltaJoin remains unsupported. The guard is deliberately conservative per block: a regular-join block still falls back when a delta join exists only in another block, even if Flink's statement-wide check accepts the statement. The fallback reason names the optimizer block. AUTO and NONE are unaffected by this guard.

State-backend admission

Flink's changelog state backend is unsupported for native keyed operators on both supported lines. Enabling it makes an affected query fall back in full; stateless queries remain eligible. This is separate from the SQL changelog row kinds used by updating queries. On 1.18, stock RocksDB selections are automatically adapted to native RocksDB; unverified custom backends and custom RocksDB factories still cause keyed SQL to fall back.

Idle-state TTL

table.exec.state.ttl runs natively everywhere Flink applies StateTtlConfig: non-windowed GROUP BY, changelog normalize, deduplication, the regular join, Top-N/LIMIT, OVER, and the temporal join. Semantics match Flink exactly — every stored value carries its last-write wall-clock timestamp (reads never refresh it), expiry happens at last_write + ttl inclusive, and expired state reads as absent and is deleted on read. Each operator's page notes any operator-specific expiry-granularity wrinkle (e.g. the temporal join's single per-key deadline instead of per-row TTL).

The 1.18 development profile retains global idle-state retention but has no per-relation STATE_TTL hints, VARIANT, delta join or SESSION table function. Grouped session windows remain available. See Flink line compatibility for the explicit host N/A cases and validation status.