Fuse exact string consumers¶
Native Calc can consume selected LPAD, RPAD and OVERLAY intermediates entirely in
Rust, without a JVM callback or an intermediate string allocation. This optimizes
expressions that already ran through the exact generated Flink evaluator; it does
not broaden SQL admission or add a configuration switch.
Why a UTF-8 kernel alone is insufficient¶
Released Flink 1.18.1 and 2.2.1 count UTF-16 units in these functions. Truncating a
supplementary character can leave an isolated surrogate. It must remain distinct
from ? until the consumer finishes, even though Java encodes both to the same
final UTF-8 bytes. For runtime input s = U+1F600:
SELECT LPAD(s, 1, 'x') = '?',
REPLACE(LPAD(s, 1, 'x'), '?', 'Z')
FROM src;
-- false, '?' at final UTF-8 output
Encoding the padding result before evaluating equality or replacement produces incorrect values. Original Java string caches can also contain UTF-16 units that cannot be reconstructed from their encoded bytes.
Technique and admission¶
A fixed set of borrowed UTF-8 spans, borrowed original Java UTF-16 spans and
individual surrogate units represents the intermediate. Equality compares intact
spans directly and visits UTF-16 units at split boundaries. Replacement writes
directly into the final Arrow offsets/value buffers, retaining adjacent surrogate
units until they can be joined or finally encoded with Java's ? replacement.
ASCII mapping uses bulk byte loops. Scalar operands remain scalar instead of
being expanded into repeated Arrow arrays; non-ASCII producer literals prepare
their UTF-16 units once per batch.
This follows Comet's scalar-operand and buffer-reuse discipline, including its
static_invoke/char_varchar_utils/read_side_padding.rs implementation. It uses
released Flink's UTF-16 semantics rather than Spark's code-point rules. No new
JNI interface, inter-operator string representation or external dependency is
introduced.
The planner recognizes these direct consumers:
| Consumer | Native conditions |
|---|---|
producer = literal, literal = producer |
Valid string literal, including NULL; Unicode and original Java caches supported |
producer <> literal, literal <> producer |
Same conditions as equality |
REPLACE(producer, old, new) |
old and new are each exactly one ASCII character; producer inputs have no non-ASCII original Java cache or non-ASCII string literal |
Producer arguments must be direct inputs, valid literals, or lossless integer widenings of direct inputs. LPAD/RPAD admit truncation and unchanged length, including their NULL, negative-length and empty-padding behavior. Growth retains the exact evaluator. OVERLAY admits three or four arguments, preserving UTF-16 positions and Flink's integer narrowing; overflowing substring cases retain the exact evaluator and its exception.
Certification applies to the batch. If a row needs an unverified path, the whole
expression uses its retained generated evaluator for that batch. Partially built
native output is discarded. This is execution inside native Calc, not a planner
fallback to a stock Calc. jvmFunctions in EXPLAIN describes the retained option
and does not count actual callbacks.
Dynamic comparison targets remain on the exact evaluator: a producer can decode its input and change the Java cache visible through an alias in the consumer. Other REPLACE operands, LIKE, surrounding CASE expressions and more complex producer arguments also retain exact generated evaluation. Inter-operator STRING identity remains a separate fallback tracked in issue #519. Issue #521 tracks the remaining fusion and performance work, including cached Unicode REPLACE.
Correctness evidence¶
- Release Rust core suite: 797 tests passed. Compiled Calc tests deliberately use an invalid evaluator ID for all nine admitted combinations, so an accidental JVM callback fails the test. They include cached and binary-only inputs, sliced batches, repeated batches, NULLs and empty batches.
- Focused SQL/planner suites: 49 passed on Flink 2.2.1/JDK 17; 46 passed and three existing capability skips on Flink 1.18.1/JDK 11. SQL tests require native work as well as matching values, including positive/negative equality filters.
- A local production-Calc differential check compared 35,909 admitted fixtures per release against unchanged generated Flink expressions. All matched; 2,120 raw fixtures per release explicitly retained exact evaluation. Those retained fixtures are not counted as pure-Rust successes.
- Unchanged released Flink
testLPad,testRPadandtestOverlaytests passed on both lines. These establish the reference semantics; the invalid-ID compiled tests independently establish the absence of callbacks.
Measurements¶
Local Apple M4 Pro / 24 GiB release-build measurements, 2026-10-26. The final component comparison uses JDK 17 for both Flink releases, one fresh JVM per engine/profile, 256 MiB initial and 512 MiB maximum Java heap, seven warmups and nine measured trials of 65,536 rows. JDK 11 correctness is tested separately; the embedded JVM benchmark could not start with JDK 11 on this host.
Stock Flink starts with prepared RowData; StreamFusion starts with equivalent prepared Arrow batches. The Rust timer wraps the production compiled Calc, allocation, intrinsic callbacks and result consumption. Input setup, initial JNI entry/final export and perimeter transposes are excluded from this component comparison. The previous path is the retained exact evaluator from the parent implementation, with three runtime arguments and inline consumer literals.
Inputs have 16 distinct values, NULL every seventh row, dynamic truncation or
overlay positions, and nominal 264-byte strings. Actual values include a prefix
and cardinality suffix. cold means binary-only StringData; cached retains the
original Java string. Values are nanoseconds per row: median [minimum-maximum].
| Flink | Expression / input, 1,024 rows per batch | Stock | Previous StreamFusion | Fused StreamFusion |
|---|---|---|---|---|
| 2.2.1 | LPAD equality, ASCII cold | 184.07 [154.06-206.30] | 250.72 [216.72-439.16] | 41.75 [37.48-44.43] |
| 2.2.1 | LPAD equality, CJK cached | 81.19 [68.34-125.36] | 219.25 [192.18-244.17] | 22.27 [21.68-23.58] |
| 2.2.1 | OVERLAY equality, supplementary cold | 284.05 [266.84-295.46] | 357.80 [314.90-376.09] | 95.49 [93.83-98.33] |
| 2.2.1 | LPAD REPLACE, ASCII cached | 183.84 [161.58-220.70] | 338.37 [288.62-372.99] | 109.98 [107.59-112.24] |
| 2.2.1 | LPAD REPLACE, supplementary cold | 328.35 [322.74-338.47] | 403.14 [389.80-431.42] | 172.73 [166.25-177.13] |
| 2.2.1 | OVERLAY REPLACE, ASCII cold | 277.40 [265.39-296.11] | 367.71 [341.21-433.44] | 71.48 [70.13-75.49] |
| 1.18.1 | LPAD equality, ASCII cold | 178.27 [149.68-196.07] | 256.28 [236.91-265.75] | 41.29 [39.27-43.35] |
| 1.18.1 | LPAD equality, CJK cached | 93.84 [68.95-116.65] | 224.83 [179.74-243.18] | 22.61 [22.15-22.97] |
| 1.18.1 | OVERLAY equality, supplementary cold | 278.92 [261.62-291.11] | 337.17 [317.72-371.52] | 99.55 [97.56-103.89] |
| 1.18.1 | LPAD REPLACE, ASCII cached | 174.86 [144.72-199.21] | 315.44 [285.79-363.77] | 106.11 [104.87-112.91] |
| 1.18.1 | LPAD REPLACE, supplementary cold | 329.91 [316.52-338.27] | 413.96 [385.11-461.43] | 172.26 [168.03-176.93] |
| 1.18.1 | OVERLAY REPLACE, ASCII cold | 285.47 [264.56-298.60] | 372.95 [340.67-435.74] | 71.41 [69.62-74.56] |
These admitted profiles are 1.65-4.41x faster than stock Flink and 2.33-9.95x faster than the previous StreamFusion evaluator. They are not a claim about all STRING expressions or every batch size. Remaining unfavorable measurements:
| Flink | Profile | Stock | Previous StreamFusion | Current StreamFusion |
|---|---|---|---|---|
| 2.2.1 | LPAD equality, cached ASCII, nominal 32 bytes, 16 rows | 39.78 [38.48-81.11] | 1339.03 [1234.66-2325.74] | 93.11 [89.47-96.46], native |
| 1.18.1 | Same small-batch profile | 42.06 [38.78-88.00] | 1251.60 [1206.23-1515.80] | 95.34 [92.73-105.59], native |
| 2.2.1 | OVERLAY REPLACE, cached CJK, 1,024 rows | 238.41 [232.06-259.62] | 373.51 [349.99-452.00] | 368.64 [345.99-434.81], retained evaluator |
| 1.18.1 | Same cached CJK profile | 227.76 [197.90-257.68] | 362.76 [351.95-445.05] | 378.37 [346.79-428.02], retained evaluator |
Small-batch compiled-Calc/array setup still exceeds stock's cached row loop, despite the improvement over the already-admitted evaluator. Cached Unicode replacement was profiled and remains on its existing evaluator because the proposed encoder did not outperform stock. Neither case adds new SQL coverage or qualifies as an acceleration over Flink. The fixed dispatch, encoding and boundary costs are part of issue #521.
Complete bounded job¶
The optional StringFusionPipelineBenchmark runs two million deterministic rows,
parallelism one, LPAD(s,n,'X') = 'a?', nominal 264-byte ASCII, NULL every seventh
row and a blackhole row sink. Both RowData-to-Arrow and Arrow-to-RowData remain in
the native INSERT plan and timer. Separate SQL tests establish output parity;
blackhole timings alone do not compare output values.
Flink 2.2.1/JDK 17, separate fresh JVMs with the same 2 GiB maximum heap, two warmups and five measured runs; both engines explain INSERT before timing:
| Engine | Median seconds | Minimum-maximum seconds |
|---|---|---|
| Stock Flink | 0.690 | 0.687-0.691 |
| Previous StreamFusion | 0.751 | 0.747-0.764 |
| Fused StreamFusion | 0.560 | 0.559-0.591 |
The current path reduces elapsed time by 19% versus Flink and 25% versus previous StreamFusion in this bounded job. These short local jobs include startup and planning; they do not establish sustained connector throughput. Earlier interleaved runs with a larger 8 GiB heap measured 0.509 seconds for Flink versus 0.611 seconds for the candidate. That unfavorable diagnostic is retained: heap, warmup and execution order materially affect short-job results. The controlled three-way comparison above uses identical resources and fresh JVMs throughout.
Reproduce¶
From the repository root, using JDK 17 and the pinned Rust toolchain:
mvn -pl streamfusion-runtime -Pbench -Dnative.build.skip=true \
-DskipTests test-compile dependency:build-classpath \
-Dmdep.outputFile=target/string-fusion.classpath
(cd native && cargo build --release --no-default-features -p streamfusion \
--lib --example string_fusion)
export SF_NATIVE_BENCH_CLASSPATH="$PWD/streamfusion-runtime/target/test-classes:$PWD/streamfusion-runtime/target/classes:$(cat streamfusion-runtime/target/string-fusion.classpath)"
export SF_NATIVE_LIB_DIR="$PWD/native/target/release"
native/target/release/examples/string_fusion 20 ascii 264 false 1024 64 7 9 flink
native/target/release/examples/string_fusion 20 ascii 264 false 1024 64 7 9 previous
native/target/release/examples/string_fusion 20 ascii 264 false 1024 64 7 9 native
Each command starts a fresh JVM. Operation codes 20/21/22 are LPAD/RPAD/OVERLAY
equality; 23/24/25 are inequality; 26/27/28 are single-ASCII-character REPLACE.
The remaining arguments select corpus (ascii, cjk, emoji), nominal width,
original Java cache, batch rows, loops, warmups, measured trials and engine.
CSV records contain operation, corpus, width, cache, batch rows, total timed rows,
trial, warmup flag, actual engine, total nanoseconds and consumed checksum.
retained explicitly means that the candidate called its exact evaluator.
Run the optional full job separately with SF_BENCHMARK=true, the bench Maven
profile and -Dtest=StringFusionPipelineBenchmark. The fusion.pipeline.rows,
fusion.pipeline.warmups, fusion.pipeline.runs, fusion.pipeline.expression
and fusion.pipeline.engines=stock|native|both properties select the workload.
Use identical heap and JVM settings for before/after comparisons, and a separate
worktree at the parent revision for the previous production encoder.