diff --git a/docs/connectors/paimon.md b/docs/connectors/paimon.md index a04e1292..3f28666a 100644 --- a/docs/connectors/paimon.md +++ b/docs/connectors/paimon.md @@ -273,8 +273,11 @@ or `table.optimizer.reuse-sub-plan-enabled` restores independent native readers. `PaimonSourceSharingTest` checks three-sink projection union, identical projections, repeated compilation, disabled sharing, distinct scans, mixed native/Flink branches, and snapshot-to-tail -results including deletes and nulls for both formats, with chained and network edges. It also -checks window closure on every shared branch after a new commit. +results including deletes and nulls for both formats, with chained and network edges. The admitted +ABS projection participates in three-consumer sharing; a STRING-to-BOOLEAN TRY_CAST projection +retains a separate host branch and two native consumers. Both shapes execute snapshot/tail +comparisons against Flink. The suite also checks window closure on every shared branch after a +new commit. Paimon's unchanged `ContinuousFileStoreITCase.testSourceReuseWithScanPushDown` passes its `Reused` assertion and its filter/limit separation assertions. Cross-sink sharing requires the deployed planner hook; installing a program into an already constructed stock planner still optimizes each diff --git a/docs/flink-compatibility.md b/docs/flink-compatibility.md index e362e05e..d94b0f40 100644 --- a/docs/flink-compatibility.md +++ b/docs/flink-compatibility.md @@ -73,6 +73,10 @@ The following are host-line differences, not missing native substitutions: session windows remain a separate supported planner construct. - Several newer scalar functions and `TO_TIMESTAMP_LTZ` string/default-precision overloads are absent from 1.18. Tests mark only those specific host constructs N/A. + The function-coverage regressions specifically exclude BTRIM, binary STARTSWITH/ENDSWITH/ELT, + PRINTF, REGEXP_COUNT/INSTR/SUBSTR and REGEXP_EXTRACT_ALL on that released line. STR_TO_MAP, + Base64 decoding, exact numerics, extrema and the other available forms still run against 1.18's + own implementation rather than borrowing 2.2 behavior. - The 1.18 host planner cannot consume update/delete streams in window TVF aggregation. Those SQL parity cases are N/A; native retraction handling still has operator-level tests. - 1.18 has released host code-generation defects for nullable `TINYINT`/`SMALLINT` array lookup, @@ -84,6 +88,15 @@ The following are host-line differences, not missing native substitutions: on Flink because the native boundary cannot treat arbitrary bytes as a one-byte fixed vector. The fallback parity tests check the complete bytes, including UTF-16 encodings and nulls. +Scalar SQL regression tests assert native routing, resolved output schemas and collected results +for admitted exact-numeric ABS, Java-backed string extrema and dynamic trims, BOOLEAN IF, +PARSE_URL, dynamic SHA2, FROM_BASE64 and the released binary prefix/suffix overloads. +The remaining floating-extrema, binary-backed string, oversized SHA2/ELT and fixed BINARY-result +restrictions retain explicit fallback checks. Generic Calc/filter rejection fixtures use +STRING-to-BOOLEAN TRY_CAST, which still falls back, rather than functions already admitted by +the planner. A fallback assertion failure does not establish a result mismatch: routing and +host/native result parity both need to pass. + ## JSON and formats The shared nested ARRAY/ROW JSON parity fixtures use each release line's collection-source diff --git a/docs/operators/calc-filter.md b/docs/operators/calc-filter.md index 14698b79..a4d2a834 100644 --- a/docs/operators/calc-filter.md +++ b/docs/operators/calc-filter.md @@ -19,9 +19,18 @@ default via a JVM upcall (and why that's not a fallback), what's opt-in, and wha fallback. - **Unsupported function/operator** outside the admitted set normally declines the whole Calc. - A [SQL/JSON Calc](#sqljson-evaluation) can instead use Flink generation for its complete program, + A [SQL/JSON Calc](#sqljson-evaluation) or a Calc containing the collection-returning string + functions below can instead use Flink generation for its complete program, subject to the host code generator and the verified batch bridge types. +## LIKE ESCAPE and SIMILAR TO + +LIKE/NOT LIKE with an explicit literal or per-row ESCAPE, and SIMILAR TO/NOT SIMILAR TO, +run Flink's generated SQL-pattern evaluator through the batch JVM bridge. Pattern grammar, +Unicode, NULLs and invalid-escape failures are Flink's own. AND/OR consumers fuse with these +expressions so invalid patterns are not evaluated on rows that Flink short-circuits; filtered-out +batches do not evaluate projections. Existing two-argument LIKE keeps its current native path. + ## COALESCE `COALESCE` retains the first non-NULL operand without evaluating it again. When an operand @@ -45,8 +54,9 @@ This retains Flink's branch casts, call counts and code-generation evaluation or user-defined function named IF retains its own implementation. Unsupported children retain the existing admission rules. SQL parity tests cover exact numerics, strings, temporal and binary values, nullable conditions, nested expressions, errors and multiple batches. -The admitted result types are numeric, character, DATE, TIME, plain TIMESTAMP and binary. -BOOLEAN, TIMESTAMP_LTZ and complex result types retain fallback because their IF overloads +The admitted result types are BOOLEAN, numeric, character, DATE, TIME, plain TIMESTAMP and binary. +BOOLEAN branches preserve TRUE/FALSE/NULL conditions, nullable results, nested predicates and +unselected failing branches across batches. TIMESTAMP_LTZ and complex result types retain fallback because their IF overloads are not registered by the released Flink code generator. The release benchmark below measures this coverage change against the previous full Flink @@ -113,7 +123,8 @@ As with Flink's generated random fields, stream state belongs to the running ope not keyed checkpoint state. Floating-point unary negation also runs natively, including `-RAND(seed)`, and preserves signed -zero, infinities, NaN and NULL. Integer and DECIMAL unary negation retain their existing fallback. +zero, infinities, NaN and NULL. Integer and DECIMAL unary negation use Flink-generated +expressions through the batch JVM bridge inside native Calc. Release diagnostic on Apple M1 Max, JDK 17/Flink 2.2.1: two million rows, parallelism 1, two warmups and five interleaved trials, with rowwise source/sink and both transposes asserted: @@ -249,8 +260,9 @@ These functions run entirely in Rust by default, in projections, predicates, and | `SHA224(s)`, `SHA256(s)`, `SHA384(s)`, `SHA512(s)` | Lowercase hexadecimal SHA-2 of the UTF-8 bytes; NULL input produces NULL. | | `SHA2(s, bit_length)` | The two-argument form with a literal bit length of 224, 256, 384, or 512; equivalent to the corresponding fixed-width function. | -A non-literal or NULL `SHA2` bit length, and other bit lengths, are not admitted. A dynamic bit length -falls back; literal values are checked exactly, including `BIGINT`, without truncating to 32 bits. +A dynamic `SHA2` bit length uses Flink-generated code through the batch JVM bridge, preserving +NULLs and the released host's unsupported-algorithm failure. Literal values are checked exactly, +including `BIGINT`, without truncating to 32 bits; unsupported or NULL literal widths fall back. For example, `SHA2(s, CAST(4294967520 AS BIGINT))` falls back and retains Flink's unsupported-algorithm failure instead of being treated as SHA-224. Flink 2.2 does not expose hash overloads with an explicit character set. Binary/collection concatenation is outside this @@ -465,6 +477,10 @@ These scalar rules are separate from grouping-key equality and sort ordering. ## Casts +BOOLEAN to STRING, bounded VARCHAR and CHAR runs Flink's cast executor through the batch JVM +bridge inside native Calc. TRUE/FALSE use uppercase text, NULL stays NULL, and bounded targets +retain Flink's truncation and CHAR padding. TRY_CAST follows the same released cast rules. + Native, unconditionally, with no host involvement: - **Widening numeric** — integer→wider integer, integer→float/double, float→double. @@ -513,7 +529,8 @@ an unselected failing cast. Default-mode casts nested under AND/OR still fall ba that Flink's row short-circuiting suppresses errors on unselected rows; legacy-mode casts can compose under AND/OR because malformed input returns NULL. A bare expression encoder without table configuration declines this cast instead of guessing the mode. -BOOLEAN-to-string and BOOLEAN TRY_CAST remain unsupported. +STRING-to-BOOLEAN TRY_CAST remains unsupported. The reverse BOOLEAN-to-character casts use +the host-exact path described above. ### Integer/string casts @@ -590,6 +607,18 @@ expressions; see [temporal functions](temporal-functions.md). ## Decimal arithmetic +### Exact unary and integral functions + +Integer and DECIMAL unary minus, ABS and SIGN execute Flink's generated expression code through +the existing batch JVM bridge. Integral FLOOR, CEIL and TRUNCATE use the same path, including +per-row TRUNCATE positions. This retains resolved widths, decimal precision/scale, NULL handling, +and Java overflow behavior, including ABS of the minimum INT/BIGINT value. Adjacent supported +host expressions fuse before crossing the Arrow boundary. These are host-evaluated functions +inside a columnar native Calc, not pure-Rust kernels. + +DECIMAL FLOOR/CEIL retain fallback: the released-host collection path has unverified decimal +precision behavior. Existing floating-point gates and decimal TRUNCATE/ROUND kernels are unchanged. + ### Decimal ROUND, TRUNCATE and literals `ROUND(decimal_column[, literal_integer_scale])` runs with compatibility overrides disabled. @@ -673,9 +702,13 @@ benchmark results and do not disable otherwise verified expressions. ### STARTSWITH +Binary overloads of STARTSWITH and ENDSWITH use released Flink code through the batch JVM +bridge. They compare bytes directly, retaining empty-prefix/suffix and NULL behavior, and never +decode binary inputs as text. Literal and runtime binary operands are supported within the +generated expression. + Two character arguments, literal or column. Matches a literal prefix, including Unicode and empty strings; any NULL argument returns NULL. Wildcard characters have no special meaning. -Binary operands fall back. ### ENDSWITH @@ -735,7 +768,18 @@ Character strings and binary columns are encoded as padded RFC 4648 Base64 witho Strings use their UTF-8 bytes; binary inputs preserve every byte, including invalid UTF-8. Both overloads share the direct-output encoder. Empty input stays empty and NULL propagates. VARBINARY literals are native; fixed-size BINARY literals retain the literal encoder's fallback. -FROM_BASE64 falls back. + +`FROM_BASE64` accepts character and binary inputs through Flink-generated evaluation. Invalid +encoding raises the same host exception; empty input and NULL retain their host results. Decoded +STRING values may contain arbitrary bytes. Final scalar STRING projections therefore travel as +Arrow Binary and are read as Flink StringData without UTF-8 normalization. Consumers such as +comparison, length, casts and re-encoding fuse in Flink before Arrow export. Sensitive STRING +results crossing another relational operator, feeding a native columnar sink, or contained in +whole-Calc complex outputs retain fallback rather than exposing invalid Arrow Utf8. Row sinks +can consume the final decoded bytes through the Arrow-to-RowData transpose. + +`IS_DECIMAL`, `IS_DIGIT` and `IS_ALPHA` use the same generated evaluator, preserving Flink syntax +and Unicode classification, including non-nullable FALSE for NULL and empty input. ### UNHEX @@ -743,11 +787,18 @@ Character inputs produce BYTES. Either hex letter case is accepted; invalid byte ### GREATEST -Integers, BOOLEAN and matching-precision/scale DECIMAL are native, with strict NULL propagation. Strings require ASCII literals or CASE results composed entirely of ASCII literals. Unrestricted string columns fall back: Flink uses UTF-16 order for Java-backed strings and byte order after binary materialization. Floating point and mixed decimal scales fall back. +Integers, BOOLEAN and matching-precision/scale DECIMAL retain their Rust kernels and strict NULL +propagation. Character results, mixed exact numerics, TIMESTAMP and TIMESTAMP_LTZ use +Flink-generated expressions through the batch JVM bridge, preserving coercions and nanoseconds. +Runtime strings use that bridge when the Calc reads an external DataStream whose conversion +produces Java-backed strings. Unknown or binary-backed input representations retain fallback: +Flink can use UTF-16 ordering before serialization and byte ordering afterward. Floating-point +extrema retain fallback. ### LEAST -Uses the same type and ASCII-proof gates as GREATEST, with strict NULL propagation and minimum comparison. +Uses the same kernels, generated-expression paths and string-representation gates as GREATEST, +with strict NULL propagation and minimum comparison. ### INITCAP @@ -759,7 +810,11 @@ Three character arguments. Mappings use Unicode codepoints, not graphemes. The f ### BTRIM -One-argument space trimming and two-argument character-set trimming with a literal set are native. Empty sets preserve the input and NULL propagates. Column trim sets fall back because Flink can change their meaning after an exchange when the first set character is a space. +One-argument space trimming and two-argument character-set trimming with a literal set retain +their Rust kernels. Empty sets preserve the input and NULL propagates. Per-row sets use Flink's +generated evaluator when the Calc reads a proven external Java-string conversion. Unknown or +binary-backed representations retain fallback because Flink can change set behavior after an +exchange when the first set character is a space. ### TRIM @@ -770,7 +825,11 @@ Column trim sets fall back for the same Flink representation-dependent behavior ### ELT -An INTEGER index and character alternatives are admitted. The index is 1-based; out-of-range and NULL indices return NULL. Only the selected alternative's NULL matters. Other index types and binary alternatives fall back: Flink casts its boxed index to Integer after its bounds check. Explicit casts to INTEGER follow the existing cast rules. +The binary-result overload also runs through Flink-generated code. Dynamic INT indices retain +one-based selection, out-of-range NULLs and NULL operands. Multiple binary result columns retain +independent byte arrays across rows and batches. + +An INTEGER index and character alternatives are admitted. The index is 1-based; out-of-range and NULL indices return NULL. Only the selected alternative's NULL matters. Other index types fall back: Flink casts its boxed index to Integer after its bounds check. Explicit casts to INTEGER follow the existing cast rules. ### URL_ENCODE @@ -780,6 +839,22 @@ Character strings use Java form encoding: space becomes `+`, ASCII alphanumerics Character strings and integer positions, widened to BIGINT without losing bits. Preserves Java UTF-16 positions, length narrowing/overflow, and substring errors. Non-positive or beyond-end starts return the source; zero/negative lengths omit the suffix. Split surrogate pairs encode as `?`, like Flink. Any NULL argument returns NULL. +### PARSE_URL + +The two- and three-argument overloads use Flink's generated evaluator through the batch JVM +bridge, including runtime URL parts and query keys. Java URL component spelling, raw percent +escapes, duplicate query keys, absent components and NULL/invalid-input behavior remain Flink's. +No Rust URL normalization or query decoding is substituted. + +### PRINTF + +PRINTF uses Flink's generated formatter through the batch JVM bridge for supported scalar +arguments, including STRING, integral and DECIMAL values and per-row formats. Argument indices, +width, precision, locale, NULLs and invalid-format results follow the selected Flink runtime. +Because character formatting can produce isolated UTF-16 surrogates, consumers fuse with PRINTF +before Arrow conversion; a sensitive string crossing another operator retains the existing +representation-protection fallback. Direct final projections are supported. + ### URL_DECODE One character argument is native. Form decoding preserves JDK UTF-8 replacement grouping and returns NULL for malformed escapes. The planner selects the runtime JDK rule: JDK 17/21 (and pre-25 runtimes) use Integer.parseInt, accepting signed one-digit escapes and BMP Unicode hex digits; JDK 25+ uses ASCII-only HexFormat rules. @@ -829,8 +904,9 @@ date/time, interval, supported timestamp types, and recursively nested ARRAY/ROW those leaves. Nested arguments use Flink internal views over the imported Arrow batch; generated results are copied into owned Arrow output vectors before that batch closes. The callback still crosses JNI once per batch, including multi-column results and filtering. -MAP/MULTISET boundary values, including maps nested inside an array or row, remain explicit -fallback. Constructing containers internally is allowed when the resulting boundary types are +MAPs with character keys and character values are also admitted, including inside ARRAY/ROW. +Other MAP and MULTISET boundary types remain explicit fallback. +Constructing containers internally is allowed when the resulting boundary types are admitted. Unsupported host code generation and UDF signatures retain explicit fallback. Flink 2.2.1 rejects dynamic JSON_EXISTS paths; that host failure is preserved. No configuration opt-in is required. @@ -1239,7 +1315,11 @@ Same input and boundary rules as LPAD, with padding appended on the right. Dynam ### SPLIT_INDEX -Character separators and TINYINT/SMALLINT/INTEGER indices may be dynamic. Indices are zero-based; negative/out-of-range indices, empty input, or any NULL produce NULL. Whole separators preserve empty tokens. An empty separator uses Java Character.isWhitespace, including tabs and line separators but excluding non-breaking spaces. Numeric separators and BIGINT indices fall back. +Character separators and TINYINT/SMALLINT/INTEGER indices may be dynamic. Indices are zero-based; negative/out-of-range indices, empty input, or any NULL produce NULL. Whole separators preserve empty tokens. An empty separator uses Java Character.isWhitespace, including tabs and line separators but excluding non-breaking spaces. Numeric separator overloads use Flink-generated code and interpret the integer as a character code. BIGINT indices with character separators fall back. + +Generated scalar helpers keep operand computations within the same Flink evaluator, retaining +intermediate StringData representation. This also preserves the released host's failure for a +computed empty trim set instead of silently changing it through a string conversion. ### Temporal parsing, extraction and rounding @@ -1250,14 +1330,32 @@ by default. See the complete [temporal function inventory](temporal-functions.md ### LTRIM -One-argument space trimming and two-argument trimming with a literal Unicode character set are native. Empty sets preserve the input; NULL propagates. Dynamic trim sets fall back because Flink semantics depend on whether strings are Java-backed or binary-backed. +One-argument space trimming and literal Unicode sets retain their Rust kernels. Per-row sets +use the same generated evaluator and source-representation gate as BTRIM. Empty sets preserve +the input and NULL propagates. ### RTRIM -Uses the same literal-set gate as LTRIM, trimming from the right. Dynamic trim sets fall back; one-argument space trimming is native. +Uses the same literal kernels and per-row set admission as LTRIM, trimming from the right. ## Case folding & regex +`REGEXP_EXTRACT_ALL` and `STR_TO_MAP` use Flink-generated evaluation. ARRAY/MAP results use +the whole-Calc batch JVM route with owned Arrow output vectors; scalar consumers can stay in +one generated expression. Capture groups, unmatched optional groups, empty matches, invalid +patterns/indices, NULL containers/elements, regex map delimiters, duplicate keys and missing +values follow released Flink. Filters run before projections, including all-filtered batches. +ARRAY and MAP lookups and cardinality inside the Calc retain the host's internal values. +Character results from these functions crossing another relational operator retain the string +identity fallback, as do whole-Calc character outputs containing arbitrary Base64-decoded bytes. +The operator remains columnar; these functions themselves execute on the JVM. + +REGEXP, REGEXP_REPLACE, REGEXP_COUNT, REGEXP_INSTR and REGEXP_SUBSTR use Flink-generated +expressions through the batch JVM bridge for literal and per-row patterns. Java lookaround, +backreferences, UTF-16 positions, empty matches, literal replacement strings, invalid-pattern +behavior and resolved INT/BOOLEAN/STRING types are retained. AND/OR consumers fuse with these +calls to preserve Flink evaluation order. These functions do not use Rust's regex engine. + **Native by default — not a fallback.** `UPPER`/`LOWER` and `REGEXP_EXTRACT` run natively by default via a columnar JVM upcall to Flink's own string routines — `BinaryStringData` case folding and `SqlFunctionUtils.regexpExtract` — so the result is byte-identical to the host, and the rest of the @@ -1314,9 +1412,9 @@ A number of otherwise-admitted functions decline when called with an argument sh implementation can't handle, even though the function itself is supported: - An **unsupported literal type** anywhere in the expression. -- **`TRIM`** — dynamic trim sets; all directions with literal sets are native. +- **`TRIM`** — dynamic SQL trim sets; all directions with literal sets are native. Per-row + BTRIM/LTRIM/RTRIM sets use their documented external-Java-string admission. - **`POSITION`** — a `FROM` start offset. -- **`SPLIT_INDEX`** — the numeric separator overload. - **`CURRENT_WATERMARK`** — requires a Calc watermark context; unsupported in standalone join or UNNEST residuals. - **Collection subscripts:** non-INT ARRAY indexes and literal indexes below one; runtime MAP keys of floating, collection or mismatched types; nullable non-compact decimal/timestamp MAP keys. @@ -1326,6 +1424,38 @@ implementation can't handle, even though the function itself is supported: See [Configuration](../configuration.md) for the full `allowIncompatible` flag surface referenced throughout this page. +## Generated helper performance + +The generated functions above extend the expressions that can stay inside a columnar island. +They are correctness/coverage work, not standalone scalar speedups: Flink still evaluates each +row inside the batch callback. A release-only diagnostic on Apple M4 Pro on 2026-09-21 used 200,000 rows, +64-byte text/binary payloads, parallelism 1, one warmup and three alternating measured runs per +engine. Matched-source identity controls and both row/Arrow transposes remain in the measured +path. Every query verifies native Calc admission before timing. + +| Expression workload | Flink seconds | Native seconds | Throughput ratio | +|---|---:|---:|---:| +| BIGINT ABS | 0.094 | 0.117 | 0.806x | +| DECIMAL SIGN | 0.096 | 0.153 | 0.630x | +| Runtime STRING GREATEST | 0.103 | 0.145 | 0.709x | +| BOOLEAN IF | 0.095 | 0.104 | 0.911x | +| BOOLEAN to STRING | 0.093 | 0.110 | 0.844x | +| LIKE ESCAPE | 0.106 | 0.132 | 0.799x | +| REGEXP_COUNT | 0.138 | 0.171 | 0.808x | +| PARSE_URL | 0.134 | 0.174 | 0.771x | +| PRINTF BIGINT | 0.168 | 0.184 | 0.913x | +| Dynamic BTRIM | 0.117 | 0.178 | 0.659x | +| IS_ALPHA | 0.096 | 0.136 | 0.706x | +| Binary STARTSWITH | 0.089 | 0.131 | 0.681x | +| REGEXP_EXTRACT_ALL | 0.200 | 0.279 | 0.718x | + +These short local timings measure the cost of the new coverage. They do not establish a gain +for an entire native pipeline, and every standalone workload here is slower than Flink. The +bridge is retained to allow composition with existing native operators; future performance +claims require measuring that complete pipeline. Reproduce with `ScalarFunctionBenchmark` +under `-Pbench`, `SF_BENCHMARK=true`, `-Dscalar.rows=200000 -Dscalar.bytes=64` and +`-Dscalar.warmup=1 -Dscalar.runs=3`, selecting the corresponding `scalar.functions` names. + ## Flink 1.18 compatibility The 1.18 development build disables unverified Jackson buffer emulation and runs SQL/JSON through diff --git a/src/main/java/tech/streamfusion/planner/CalcOutputTypeCheck.java b/src/main/java/tech/streamfusion/planner/CalcOutputTypeCheck.java index b0ee3f6b..d9f9aab1 100644 --- a/src/main/java/tech/streamfusion/planner/CalcOutputTypeCheck.java +++ b/src/main/java/tech/streamfusion/planner/CalcOutputTypeCheck.java @@ -61,18 +61,26 @@ static String mismatch(RexExpression encoded, RelDataType inputType, RowType dec if (notInferable != null) { return notInferable; } - return mismatch(Data.importSchema(allocator, outputSchema, null).getFields(), declared); + return mismatch(Data.importSchema(allocator, outputSchema, null).getFields(), declared, encoded); } catch (NativeException compileFailure) { return "expression does not compile natively: " + compileFailure.getMessage(); } } - private static String mismatch(List inferred, RowType declared) { + private static String mismatch(List inferred, RowType declared, RexExpression encoded) { if (inferred.size() != declared.getFieldCount()) { return inferred.size() + " projections for " + declared.getFieldCount() + " declared columns"; } for (int i = 0; i < inferred.size(); i++) { Field actual = inferred.get(i); + if (encoded.isBinaryStringProjection(i) + && actual.getType() instanceof org.apache.arrow.vector.types.pojo.ArrowType.Binary + && (declared.getTypeAt(i).getTypeRoot() + == org.apache.flink.table.types.logical.LogicalTypeRoot.VARCHAR + || declared.getTypeAt(i).getTypeRoot() + == org.apache.flink.table.types.logical.LogicalTypeRoot.CHAR)) { + continue; + } if (!ArrowConversion.readsAs(actual, declared.getTypeAt(i))) { return String.format( "projection `%s` evaluates natively as %s but the plan declares %s", diff --git a/src/main/java/tech/streamfusion/planner/FlinkExpressionFunction.java b/src/main/java/tech/streamfusion/planner/FlinkExpressionFunction.java index 6c906150..4d59e860 100644 --- a/src/main/java/tech/streamfusion/planner/FlinkExpressionFunction.java +++ b/src/main/java/tech/streamfusion/planner/FlinkExpressionFunction.java @@ -45,8 +45,17 @@ public interface Evaluator extends Function { LogicalType[] argumentTypes, ReadableConfig config, ClassLoader classLoader) { + this(expression, argumentTypes, config, classLoader, false); + } + + FlinkExpressionFunction( + RexNode expression, + LogicalType[] argumentTypes, + ReadableConfig config, + ClassLoader classLoader, + boolean binaryStringResult) { this( - scalarBody(expression, argumentTypes, config, classLoader), + scalarBody(expression, argumentTypes, config, classLoader, binaryStringResult), argumentTypes, config, classLoader); @@ -72,7 +81,8 @@ private static Body scalarBody( RexNode expression, LogicalType[] argumentTypes, ReadableConfig config, - ClassLoader classLoader) { + ClassLoader classLoader, + boolean binaryStringResult) { var context = new Context(config, classLoader); var generator = new ExprCodeGenerator(context, false); generator.bindInput(RowType.of(argumentTypes), "input", scala.Option.empty()); @@ -85,6 +95,7 @@ private static Body scalarBody( + result.nullTerm() + ") { return null; }\nreturn " + result.resultTerm() + + (binaryStringResult ? ".toBytes()" : "") + ";\n", null); } diff --git a/src/main/java/tech/streamfusion/planner/HostStringInputs.java b/src/main/java/tech/streamfusion/planner/HostStringInputs.java new file mode 100644 index 00000000..b8c8a837 --- /dev/null +++ b/src/main/java/tech/streamfusion/planner/HostStringInputs.java @@ -0,0 +1,45 @@ +package tech.streamfusion.planner; + +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.core.TableScan; +import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalDataStreamScan; +import org.apache.flink.table.planner.plan.schema.TableSourceTable; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.logical.LogicalTypeFamily; + +/** Proof for functions whose Flink behavior depends on Java-backed versus binary-backed strings. */ +final class HostStringInputs { + private HostStringInputs() {} + + static boolean areJavaBacked(RelNode input) { + if (input instanceof StreamPhysicalDataStreamScan scan) { + return hasExternalStrings(scan.dataStreamTable().dataType()); + } + if (!(input instanceof TableScan scan)) return false; + TableSourceTable table = scan.getTable().unwrap(TableSourceTable.class); + if (table == null) return false; + Object source = table.tableSource(); + if (!source.getClass().getName().equals( + "org.apache.flink.table.planner.connectors.ExternalDynamicSource")) return false; + try { + // ExternalDynamicSource is package-private and exposes no physical-conversion accessor. + var field = source.getClass().getDeclaredField("physicalDataType"); + if (!field.trySetAccessible()) return false; + return hasExternalStrings((DataType) field.get(source)); + } catch (ReflectiveOperationException | RuntimeException unavailable) { + return false; + } + } + + private static boolean hasExternalStrings(DataType type) { + if (org.apache.flink.table.data.RowData.class.isAssignableFrom(type.getConversionClass()) + || org.apache.flink.table.data.ArrayData.class.isAssignableFrom(type.getConversionClass()) + || org.apache.flink.table.data.MapData.class.isAssignableFrom(type.getConversionClass())) { + return false; + } + if (type.getLogicalType().is(LogicalTypeFamily.CHARACTER_STRING)) { + return type.getConversionClass() == String.class; + } + return type.getChildren().stream().allMatch(HostStringInputs::hasExternalStrings); + } +} diff --git a/src/main/java/tech/streamfusion/planner/JsonStringIdentity.java b/src/main/java/tech/streamfusion/planner/JsonStringIdentity.java index 25f977bd..9f8d4bd8 100644 --- a/src/main/java/tech/streamfusion/planner/JsonStringIdentity.java +++ b/src/main/java/tech/streamfusion/planner/JsonStringIdentity.java @@ -10,7 +10,7 @@ import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalCalc; import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalSink; -/** Keeps Java UTF-16 JSON results inside a generated expression or at the final output boundary. */ +/** Keeps sensitive Java UTF-16 results inside a generated expression or at the final output boundary. */ final class JsonStringIdentity { private JsonStringIdentity() {} @@ -24,10 +24,14 @@ static boolean containsSensitiveString(RexNode expression) { return containsSensitiveString(access.getReferenceExpr()); } if (!(expression instanceof RexCall call)) return false; + if (call.getOperator().getName().equals("REGEXP_EXTRACT_ALL") + || call.getOperator().getName().equals("STR_TO_MAP")) return true; if (SqlTypeFamily.CHARACTER.contains(call.getType()) && (call.getOperator().getName().equals("JSON_VALUE") || call.getOperator().getName().equals("JSON_QUERY") - || call.getOperator().getName().equals("JSON_UNQUOTE"))) { + || call.getOperator().getName().equals("JSON_UNQUOTE") + || call.getOperator().getName().equals("PRINTF") + || call.getOperator().getName().equals("FROM_BASE64"))) { return true; } return call.getOperands().stream().anyMatch(JsonStringIdentity::containsSensitiveString); @@ -67,7 +71,27 @@ private static boolean crossesOperatorBoundary( return false; } - private static boolean containsCharacter(RelDataType type) { + static boolean containsBinaryString(RexNode expression) { + if (expression instanceof RexFieldAccess access) { + return containsBinaryString(access.getReferenceExpr()); + } + return expression instanceof RexCall call + && (call.getOperator().getName().equals("FROM_BASE64") + || call.getOperands().stream().anyMatch(JsonStringIdentity::containsBinaryString)); + } + + static boolean projectsBinaryString(RelNode node) { + if (node instanceof StreamPhysicalCalc calc) { + var program = calc.getProgram(); + for (var projection : program.getProjectList()) { + RexNode expression = program.expandLocalRef(projection); + if (containsCharacter(expression.getType()) && containsBinaryString(expression)) return true; + } + } + return node.getInputs().stream().anyMatch(JsonStringIdentity::projectsBinaryString); + } + + static boolean containsCharacter(RelDataType type) { if (SqlTypeFamily.CHARACTER.contains(type)) return true; if (type.getComponentType() != null && containsCharacter(type.getComponentType())) return true; if (type.getKeyType() != null && containsCharacter(type.getKeyType())) return true; diff --git a/src/main/java/tech/streamfusion/planner/PhysicalPlanScan.java b/src/main/java/tech/streamfusion/planner/PhysicalPlanScan.java index 68cd3eaa..2531dfb4 100644 --- a/src/main/java/tech/streamfusion/planner/PhysicalPlanScan.java +++ b/src/main/java/tech/streamfusion/planner/PhysicalPlanScan.java @@ -160,6 +160,13 @@ private RelNode substitute(RelNode root, Set repeatedSources, boolean fi // Pass 1 substitutes native (columnar) operators. int previousSubstitutions = substitutions; RelNode substituted = rewrite(root, new PlanContext(this, repeatedSources)); + if (root instanceof org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalSink + && substituted instanceof ColumnarInput + && JsonStringIdentity.projectsBinaryString(root)) { + substitutions = previousSubstitutions; + recordFallback("binary-backed STRING requires a row sink"); + return root; + } // Whole-query all-or-nothing: every native operator but a source/sink is Arrow → Arrow. // If any operator other than a source (a leaf) or the sink (the plan root) is still row-wise, // the diff --git a/src/main/java/tech/streamfusion/planner/RexExpression.java b/src/main/java/tech/streamfusion/planner/RexExpression.java index 616b9a0d..fdab6caa 100644 --- a/src/main/java/tech/streamfusion/planner/RexExpression.java +++ b/src/main/java/tech/streamfusion/planner/RexExpression.java @@ -151,6 +151,7 @@ final class RexExpression { private final List udfIdSlots = new ArrayList<>(); private final List projectionRoots = new ArrayList<>(); + private final java.util.Set binaryStringProjections = new java.util.HashSet<>(); private int conditionRoot = -1; private String[] outputNames = new String[0]; // Why the encode declined, set at the first (innermost) un-admitted node; null if it succeeded. @@ -167,6 +168,7 @@ final class RexExpression { private RexNode projectionRoot; private int binaryUdfCalls; private boolean rowFusion; + private boolean javaStringInputs; private final java.util.Set statefulUdfEvaluations = new java.util.HashSet<>(); private ClassLoader expressionClassLoader = RexExpression.class.getClassLoader(); @@ -330,7 +332,8 @@ private record CalcEncoding(RexExpression encoder, boolean supported) {} private static CalcEncoding tryEncodeCalc(Calc calc) { RexExpression encoder = forCalc(calc); boolean supported = encoder.emitCalc(calc); - if (!supported && containsSqlJson(calc.getProgram())) { + if (!supported + && (containsSqlJson(calc.getProgram()) || containsCollectionStringFunction(calc.getProgram()))) { // A failed native attempt may have populated pools and UDF bindings. Generate the complete // Calc in a fresh encoder so short-circuiting and row evaluation order stay with Flink. encoder = forCalc(calc); @@ -345,6 +348,7 @@ private static RexExpression forCalc(Calc calc) { org.apache.flink.table.planner.utils.ShortcutUtils.unwrapClassLoader(calc); encoder.configure(org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig(calc)); encoder.watermarkAvailable = true; + encoder.javaStringInputs = HostStringInputs.areJavaBacked(calc.getInput()); return encoder; } @@ -461,6 +465,8 @@ private static int rowCalcTypeCode(RelDataType type) { boolean nested = switch (type.getSqlTypeName()) { case ARRAY -> rowCalcTypeCode(type.getComponentType()) >= 0; + case MAP -> SqlTypeFamily.CHARACTER.contains(type.getKeyType()) + && SqlTypeFamily.CHARACTER.contains(type.getValueType()); case ROW -> type.getFieldList().stream().allMatch(field -> rowCalcTypeCode(field.getType()) >= 0); default -> false; @@ -468,6 +474,22 @@ private static int rowCalcTypeCode(RelDataType type) { return nested ? tech.streamfusion.operator.NativeUdf.TYPE_INTERNAL : hostCastTypeCode(type); } + private static boolean containsCollectionStringFunction(RexProgram program) { + return program.getCondition() != null + && containsCollectionStringFunction(program.expandLocalRef(program.getCondition())) + || program.getProjectList().stream() + .anyMatch(project -> containsCollectionStringFunction(program.expandLocalRef(project))); + } + + private static boolean containsCollectionStringFunction(RexNode node) { + if (node instanceof org.apache.calcite.rex.RexFieldAccess access) { + return containsCollectionStringFunction(access.getReferenceExpr()); + } + return node instanceof RexCall call + && (java.util.Set.of("REGEXP_EXTRACT_ALL", "STR_TO_MAP").contains(call.getOperator().getName()) + || call.getOperands().stream().anyMatch(RexExpression::containsCollectionStringFunction)); + } + private boolean emitRowCalc(Calc calc) { rowFusion = true; RexProgram program = calc.getProgram(); @@ -514,6 +536,10 @@ public RexNode visitInputRef(RexInputRef input) { RexUtil.expandSearch( calc.getCluster().getRexBuilder(), null, program.expandLocalRef(ref)); if (!validateGeneratedExpression(projection)) return false; + if (JsonStringIdentity.containsCharacter(projection.getType()) + && JsonStringIdentity.containsBinaryString(projection)) { + return reject("binary-backed STRING requires a final scalar projection"); + } if (rowCalcTypeCode(projection.getType()) < 0) return reject("row-fused UDF output type is not supported: " + projection.getType()); projections.add(projection.accept(remap)); @@ -570,6 +596,10 @@ int[] projectionRoots() { return toIntArray(projectionRoots); } + boolean isBinaryStringProjection(int index) { + return binaryStringProjections.contains(index); + } + /** The condition tree's root node index, or -1 if the Calc has no condition. */ int conditionRoot() { return conditionRoot; @@ -845,6 +875,10 @@ private boolean emitLiteral(RexLiteral literal) { } private boolean emitCall(RexCall call) { + if ((call.getKind() == SqlKind.AND || call.getKind() == SqlKind.OR) + && containsExactScalarFunction(call)) { + return emitHostExpression(call, true); + } if (call.getOperands().stream() .anyMatch( operand -> @@ -895,7 +929,7 @@ && hasImplicitStringNumericCast(call)) { strings.add(unixTimeFormat); return emit(call.getOperands().get(0)); } - if (needsTemporalFunction(call) || needsExactPower(call)) { + if (needsTemporalFunction(call) || needsExactPower(call) || needsExactScalarFunction(call)) { return emitHostExpression(call, false); } if (call.getKind() == SqlKind.MINUS_PREFIX) { @@ -982,7 +1016,7 @@ && hasImplicitStringNumericCast(call)) { == org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable.IF) { if (call.getOperands().size() != 3) return reject("IF requires three operands"); boolean supportedType = switch (call.getType().getSqlTypeName()) { - case TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, REAL, DOUBLE, DECIMAL, + case BOOLEAN, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, REAL, DOUBLE, DECIMAL, CHAR, VARCHAR, DATE, TIME, TIMESTAMP, BINARY, VARBINARY -> true; default -> false; }; @@ -2170,7 +2204,7 @@ private static boolean hostCastSupported(RelDataType sourceType, RelDataType res boolean targetNumeric = numericRank(target) >= 0 || target == SqlTypeName.DECIMAL; boolean sourceFloat = source == SqlTypeName.FLOAT || source == SqlTypeName.REAL || source == SqlTypeName.DOUBLE; - return (sourceNumeric && targetString) + return ((sourceNumeric || source == SqlTypeName.BOOLEAN) && targetString) || (sourceString && targetNumeric) || (sourceString && targetString) || (sourceFloat && target == SqlTypeName.DECIMAL); @@ -2293,6 +2327,51 @@ && extractField(unit) != null) { }; } + private static boolean needsExactScalarFunction(RexCall call) { + if (call.getKind() == SqlKind.LIKE && call.getOperands().size() == 3 + || call.getKind() == SqlKind.SIMILAR) return true; + if (call.getOperands().isEmpty()) return false; + SqlTypeName input = call.getOperands().get(0).getType().getSqlTypeName(); + boolean integral = switch (input) { + case TINYINT, SMALLINT, INTEGER, BIGINT -> true; + default -> false; + }; + boolean exact = integral || input == SqlTypeName.DECIMAL; + if (call.getKind() == SqlKind.MINUS_PREFIX) return exact; + return switch (call.getOperator().getName().toUpperCase(Locale.ROOT)) { + case "ABS", "SIGN" -> exact; + case "FLOOR", "CEIL", "CEILING" -> integral && call.getOperands().size() == 1; + case "TRUNCATE" -> integral; + case "TRY_CAST" -> input == SqlTypeName.BOOLEAN && SqlTypeFamily.CHARACTER.contains(call.getType()); + case "REGEXP", "REGEXP_REPLACE", "REGEXP_COUNT", "REGEXP_INSTR", "REGEXP_SUBSTR" -> true; + case "REGEXP_EXTRACT_ALL", "STR_TO_MAP" -> true; + case "PARSE_URL" -> true; + case "PRINTF" -> true; + case "STARTSWITH", "ENDSWITH" -> SqlTypeFamily.BINARY.contains(call.getOperands().get(0).getType()); + case "ELT" -> input == SqlTypeName.INTEGER && SqlTypeFamily.BINARY.contains(call.getType()); + case "FROM_BASE64", "IS_DECIMAL", "IS_DIGIT", "IS_ALPHA" -> true; + case "SHA2" -> call.getOperands().size() == 2 && !(call.getOperands().get(1) instanceof RexLiteral); + case "SPLIT_INDEX" -> call.getOperands().size() == 3 + && SqlTypeFamily.INTEGER.contains(call.getOperands().get(1).getType()); + case "BTRIM", "LTRIM", "RTRIM" -> + call.getOperands().size() == 2 && !(call.getOperands().get(1) instanceof RexLiteral); + case "GREATEST", "LEAST" -> + SqlTypeFamily.CHARACTER.contains(call.getType()) + || temporalTypeCode(call.getType()) >= 0 + || call.getType().getSqlTypeName() == SqlTypeName.DECIMAL + && call.getOperands().stream().anyMatch( + operand -> !org.apache.calcite.sql.type.SqlTypeUtil.equalSansNullability( + operand.getType(), call.getType())); + default -> false; + }; + } + + private static boolean containsExactScalarFunction(RexNode node) { + return node instanceof RexCall call + && (needsExactScalarFunction(call) + || call.getOperands().stream().anyMatch(RexExpression::containsExactScalarFunction)); + } + private static long nativeRoundingWidth(RexCall call) { String name = call.getOperator().getName(); if (!("FLOOR".equals(name) || "CEIL".equals(name) || "CEILING".equals(name)) @@ -2344,6 +2423,8 @@ private static boolean containsScalarUdf(RexNode node) { } private boolean emitHostExpression(RexCall call, boolean fuseConsumers) { + fuseConsumers |= needsExactScalarFunction(call); + if (!validateHostStringInputs(call)) return false; if (fuseConsumers && !validateGeneratedExpression(call)) return false; List arguments = new ArrayList<>(); List types = new ArrayList<>(); @@ -2355,6 +2436,17 @@ private boolean emitHostExpression(RexCall call, boolean fuseConsumers) { return reject(e.getMessage()); } int returnCode = hostCastTypeCode(call.getType()); + boolean binaryStringResult = + SqlTypeFamily.CHARACTER.contains(call.getType()) + && JsonStringIdentity.containsBinaryString(call); + if (binaryStringResult) { + if (call != projectionRoot) { + return reject("binary-backed STRING requires a final scalar projection"); + } + // Flink strings may contain arbitrary bytes, which Arrow Utf8 must never carry. + returnCode = tech.streamfusion.operator.NativeUdf.TYPE_BINARY; + binaryStringProjections.add(projectionRoots.size() - 1); + } if (returnCode < 0) { return reject("unsupported generated-expression result type " + call.getType()); } @@ -2370,7 +2462,8 @@ private boolean emitHostExpression(RexCall call, boolean fuseConsumers) { expression, types.toArray(org.apache.flink.table.types.logical.LogicalType[]::new), temporalConfig, - expressionClassLoader); + expressionClassLoader, + binaryStringResult); eval = FlinkExpressionFunction.class.getMethod("eval", Object[].class); } catch (Exception e) { return reject("host expression cannot be generated: " + e.getMessage()); @@ -2409,7 +2502,8 @@ private RexNode hostExpressionArguments( if (node instanceof RexCall call && (fuseConsumers || needsTemporalFunction(call) && nativeUnixTimeFormat(call) == null - || needsExactPower(call))) { + || needsExactPower(call) + || needsExactScalarFunction(call))) { List operands = new ArrayList<>(); for (RexNode operand : call.getOperands()) { operands.add(hostExpressionArguments(operand, arguments, types, codes, fuseConsumers)); @@ -2821,6 +2915,7 @@ private Method checkedUdfMethod(RexCall call) { private boolean validateGeneratedExpression(RexNode node) { if (!(node instanceof RexCall call)) return true; + if (!validateHostStringInputs(call)) return false; if (call.getOperator() instanceof org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction function && function.getDefinition() instanceof org.apache.flink.table.functions.ScalarFunction @@ -2830,6 +2925,25 @@ && checkedUdfMethod(call) == null) { return call.getOperands().stream().allMatch(this::validateGeneratedExpression); } + private boolean validateHostStringInputs(RexNode node) { + if (!(node instanceof RexCall call)) return true; + String name = call.getOperator().getName().toUpperCase(Locale.ROOT); + boolean stringOrdering = switch (call.getKind()) { + case LESS_THAN, LESS_THAN_OR_EQUAL, GREATER_THAN, GREATER_THAN_OR_EQUAL -> + call.getOperands().stream().anyMatch(operand -> SqlTypeFamily.CHARACTER.contains(operand.getType())); + default -> false; + }; + boolean dynamicTrim = java.util.Set.of("BTRIM", "LTRIM", "RTRIM").contains(name) + && call.getOperands().size() == 2 && !(call.getOperands().get(1) instanceof RexLiteral); + if ((dynamicTrim || stringOrdering || (name.equals("GREATEST") || name.equals("LEAST")) + && SqlTypeFamily.CHARACTER.contains(call.getType())) + && !javaStringInputs + && call.getOperands().stream().anyMatch(operand -> !isAsciiLiteralResult(operand))) { + return reject("string representation before Arrow is not proven Java-backed"); + } + return call.getOperands().stream().allMatch(this::validateHostStringInputs); + } + /** * Emits a Flink user {@link org.apache.flink.table.functions.ScalarFunction} call as a JVM-upcall * node (op {@link #KIND_UDF}). The function is registered in {@link diff --git a/src/test/java-flink1.18/tech/streamfusion/compat/FlinkTestCapabilities.java b/src/test/java-flink1.18/tech/streamfusion/compat/FlinkTestCapabilities.java index fc7e68f4..adf53ff7 100644 --- a/src/test/java-flink1.18/tech/streamfusion/compat/FlinkTestCapabilities.java +++ b/src/test/java-flink1.18/tech/streamfusion/compat/FlinkTestCapabilities.java @@ -11,6 +11,11 @@ private FlinkTestCapabilities() {} private static final java.util.Set ABSENT_SQL_FUNCTIONS = java.util.Set.of( "BTRIM", + "PRINTF", + "REGEXP_COUNT", + "REGEXP_INSTR", + "REGEXP_SUBSTR", + "REGEXP_EXTRACT_ALL", "ENDS_WITH", "ENDSWITH", "SPLIT", diff --git a/src/test/java/tech/streamfusion/BuiltinFunctionParity.java b/src/test/java/tech/streamfusion/BuiltinFunctionParity.java new file mode 100644 index 00000000..75c1ea42 --- /dev/null +++ b/src/test/java/tech/streamfusion/BuiltinFunctionParity.java @@ -0,0 +1,70 @@ +package tech.streamfusion; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static tech.streamfusion.compat.FlinkTestSources.fromData; + +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.function.Supplier; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.table.runtime.typeutils.ExternalTypeInfo; +import org.apache.flink.table.types.DataType; +import org.apache.flink.types.Row; +import tech.streamfusion.planner.NativePlanner; + +final class BuiltinFunctionParity { + private BuiltinFunctionParity() {} + + static TableEnvironment environment(DataType type, List rows) { + return environment(type, rows, "UTC"); + } + + static TableEnvironment environment(DataType type, List rows, String zone) { + var env = StreamExecutionEnvironment.getExecutionEnvironment(); + env.setParallelism(1); + var table = StreamTableEnvironment.create(env); + table.getConfig().setLocalTimeZone(ZoneId.of(zone)); + table.createTemporaryView( + "src", + fromData(env, rows, ExternalTypeInfo.of(type)), + Schema.newBuilder().fromRowDataType(type).build()); + return table; + } + + static void assertParity(Supplier environment, String sql) throws Exception { + var host = environment.get(); + var accelerated = environment.get(); + var scan = NativePlanner.install(accelerated); + assertEquals( + host.sqlQuery(sql).getResolvedSchema(), accelerated.sqlQuery(sql).getResolvedSchema()); + var expected = collect(host, sql); + var actual = collect(accelerated, sql); + assertTrue(scan.substitutions() > 0, "No native substitution: " + scan.fallbackReasons()); + assertTrue( + scan.operatorTypes().stream().anyMatch(name -> name.contains("Calc")), + "No native Calc: " + scan.operatorTypes()); + assertEquals(expected, actual, sql); + } + + private static List> collect(TableEnvironment table, String sql) throws Exception { + List> rows = new ArrayList<>(); + try (var iterator = table.executeSql(sql).collect()) { + while (iterator.hasNext()) { + Row row = iterator.next(); + List fields = new ArrayList<>(); + for (int i = 0; i < row.getArity(); i++) { + fields.add(NativeParity.comparableValue(row.getField(i))); + } + rows.add(fields); + } + } + rows.sort(Comparator.comparing(Object::toString)); + return rows; + } +} diff --git a/src/test/java/tech/streamfusion/FlinkAdditionalStringBuiltinsSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkAdditionalStringBuiltinsSqlHarnessTest.java new file mode 100644 index 00000000..80bc18a7 --- /dev/null +++ b/src/test/java/tech/streamfusion/FlinkAdditionalStringBuiltinsSqlHarnessTest.java @@ -0,0 +1,94 @@ +package tech.streamfusion; + +import static org.apache.flink.table.api.DataTypes.*; + +import java.nio.charset.StandardCharsets; +import java.util.List; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.types.Row; +import org.junit.jupiter.api.Test; + +class FlinkAdditionalStringBuiltinsSqlHarnessTest { + @Test + void classificationNumericSeparatorsAndDynamicHashes() throws Exception { + BuiltinFunctionParity.assertParity(this::environment, + "SELECT IS_DECIMAL(s), IS_DIGIT(s), IS_ALPHA(s), SPLIT_INDEX(s,sep,n), SHA2(s,bits) FROM src"); + BuiltinFunctionParity.assertParity(this::environment, + "SELECT SPLIT_INDEX(s,sep,n) FROM src WHERE IS_ALPHA(s) OR IS_DIGIT(s)"); + } + + @Test + void base64StringAndBinaryInputsPreserveInvalidEncodingAndNulls() throws Exception { + BuiltinFunctionParity.assertParity(this::base64Environment, + "SELECT FROM_BASE64(s), FROM_BASE64(b), FROM_BASE64(s) = '?', " + + "CHAR_LENGTH(FROM_BASE64(s)), TO_BASE64(FROM_BASE64(s)), " + + "CAST(FROM_BASE64(s) AS BYTES) FROM src"); + BuiltinFunctionParity.assertParity(this::base64Environment, + "SELECT FROM_BASE64(s), SHA2(s,bits) FROM src WHERE n = 99"); + } + + @Test + void decodedBytesReachTheRowSinkWithoutUtf8Normalization() throws Exception { + for (boolean nativeEnabled : new boolean[] {false, true}) { + var table = (org.apache.flink.table.api.bridge.java.StreamTableEnvironment) base64Environment(); + var scan = nativeEnabled ? tech.streamfusion.planner.NativePlanner.install(table) : null; + var query = table.sqlQuery("SELECT FROM_BASE64(s) AS v FROM src"); + var output = table.toDataStream( + query, ROW(FIELD("v", STRING())).bridgedTo(org.apache.flink.table.data.RowData.class)); + List actual = new java.util.ArrayList<>(); + try (var rows = output.map(row -> row.isNullAt(0) ? "NULL" + : java.util.Base64.getEncoder().encodeToString(row.getString(0).toBytes())) + .returns(org.apache.flink.api.common.typeinfo.Types.STRING).executeAndCollect()) { + rows.forEachRemaining(actual::add); + } + actual.sort(String::compareTo); + org.junit.jupiter.api.Assertions.assertEquals( + List.of("", "/w==", "7aCA", "AA==", "NULL", "aGVsbG8="), actual); + if (nativeEnabled) org.junit.jupiter.api.Assertions.assertTrue(scan.substitutions() > 0, + scan.fallbackReasons().toString()); + } + } + + @Test + void invalidBase64RetainsTheHostFailure() throws Exception { + NativeFailureParity.run(this::environment, "SELECT FROM_BASE64(s) FROM src") + .assertFailure(IllegalArgumentException.class, "Illegal base64 character", + NativeFailureParity.Phase.ROW_EVALUATION, NativeFailureParity.Route.NATIVE); + } + + @Test + void unsupportedHashWidthsRetainFailuresAndShortCircuiting() throws Exception { + BuiltinFunctionParity.assertParity(this::invalidHashEnvironment, + "SELECT n = 0 OR SHA2(s,bits) = 'x', n <> 0 AND SHA2(s,bits) = 'x' FROM src"); + NativeFailureParity.run(this::invalidHashEnvironment, "SELECT SHA2(s,bits) FROM src") + .assertFailure(java.security.NoSuchAlgorithmException.class, "SHA-128", NativeFailureParity.Phase.ROW_EVALUATION, + NativeFailureParity.Route.NATIVE); + } + + private TableEnvironment invalidHashEnvironment() { + return BuiltinFunctionParity.environment(ROW(FIELD("s", STRING()), FIELD("bits", INT()), + FIELD("n", INT())), List.of(Row.of("test", 128, 0))); + } + + private TableEnvironment base64Environment() { + List rows = new java.util.ArrayList<>(); + for (String value : new String[] {"aGVsbG8=", "/w==", "7aCA", "AA==", "", null}) { + rows.add(Row.of(value, value == null ? null : value.getBytes(StandardCharsets.US_ASCII), 0, 256)); + } + return BuiltinFunctionParity.environment( + ROW(FIELD("s", STRING()), FIELD("b", BYTES()), FIELD("n", INT()), FIELD("bits", INT())), rows); + } + + private TableEnvironment environment() { + return BuiltinFunctionParity.environment( + ROW(FIELD("s", STRING()), FIELD("b", BYTES()), FIELD("sep", INT()), FIELD("n", INT()), + FIELD("bits", INT())), + List.of(Row.of("aGVsbG8=", "aGVsbG8=".getBytes(StandardCharsets.US_ASCII), 73, 0, 224), + Row.of("/w==", "/w==".getBytes(StandardCharsets.US_ASCII), 256, 1, 256), + Row.of("!invalid!", new byte[] {(byte) 255, 0}, 0, -1, 384), + Row.of("123", new byte[0], 49, 0, 512), Row.of("-12.50", null, 46, 1, 256), + Row.of("abc", null, 98, 0, 256), Row.of("\u0661\u0662", null, -1, 0, 256), + Row.of("\u4e2d", null, 44, 0, 256), Row.of("", null, 44, 0, null), + Row.of(null, null, null, null, 256))); + } +} diff --git a/src/test/java/tech/streamfusion/FlinkBinaryStringBuiltinsSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkBinaryStringBuiltinsSqlHarnessTest.java new file mode 100644 index 00000000..5ce26283 --- /dev/null +++ b/src/test/java/tech/streamfusion/FlinkBinaryStringBuiltinsSqlHarnessTest.java @@ -0,0 +1,51 @@ +package tech.streamfusion; + +import static org.apache.flink.table.api.DataTypes.*; + +import java.util.ArrayList; +import java.util.List; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.types.Row; +import org.junit.jupiter.api.Test; + +class FlinkBinaryStringBuiltinsSqlHarnessTest { + @org.junit.jupiter.api.BeforeEach + void requireHostFunctions() { + tech.streamfusion.compat.FlinkTestCapabilities.requireSqlFunction("STARTSWITH"); + tech.streamfusion.compat.FlinkTestCapabilities.requireSqlFunction("ELT"); + } + + @Test + void binaryPredicatesAndSelectionPreserveEveryByteAcrossBatches() throws Exception { + BuiltinFunctionParity.assertParity(this::environment, + "SELECT STARTSWITH(b,p), ENDSWITH(b,q), ELT(n,b,p,q), ELT(2,b,q,p), " + + "STARTSWITH(ELT(n,b,p,q),p), ENDSWITH(b,X'00FF'), ELT(n,b,X'FF00') FROM src"); + BuiltinFunctionParity.assertParity(this::environment, + "SELECT ELT(n,b,p,q), ELT(1,q,p,b) FROM src WHERE STARTSWITH(b,p) OR ENDSWITH(b,q)"); + BuiltinFunctionParity.assertParity(this::environment, + "SELECT ELT(n,b,p,q), STARTSWITH(b,p) FROM src WHERE n = 99"); + } + + @Test + void binaryResultCompositionRetainsBytesAndNulls() throws Exception { + BuiltinFunctionParity.assertParity(this::environment, + "SELECT TO_BASE64(ELT(n,b,p,q)), " + + "CASE WHEN STARTSWITH(b,p) THEN ELT(n,b,p,q) ELSE q END FROM src"); + } + + private TableEnvironment environment() { + byte[] empty = new byte[0]; + byte[] high = {0, (byte) 255, (byte) 128, 0}; + List samples = List.of( + Row.of(high, new byte[] {0, (byte) 255}, new byte[] {(byte) 128, 0}, 1), + Row.of(high, high, empty, 2), Row.of(empty, empty, high, 3), + Row.of(high, empty, high, 0), Row.of(high, null, empty, -1), + Row.of(null, high, null, 4), Row.of(high, high, high, null), + Row.of(new byte[] {1, 2, 3, 4, 5}, high, high, Integer.MAX_VALUE), + Row.of(empty, null, high, Integer.MIN_VALUE)); + List rows = new ArrayList<>(); + for (int i = 0; i < 5003; i++) rows.add(Row.copy(samples.get(i % samples.size()))); + return BuiltinFunctionParity.environment( + ROW(FIELD("b", BYTES()), FIELD("p", BYTES()), FIELD("q", BYTES()), FIELD("n", INT())), rows); + } +} diff --git a/src/test/java/tech/streamfusion/FlinkBooleanIfSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkBooleanIfSqlHarnessTest.java new file mode 100644 index 00000000..84445df2 --- /dev/null +++ b/src/test/java/tech/streamfusion/FlinkBooleanIfSqlHarnessTest.java @@ -0,0 +1,39 @@ +package tech.streamfusion; + +import static org.apache.flink.table.api.DataTypes.*; + +import java.util.ArrayList; +import java.util.List; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.types.Row; +import org.junit.jupiter.api.Test; + +class FlinkBooleanIfSqlHarnessTest { + @Test + void nullableConditionsAndBranchesSpanBatches() throws Exception { + BuiltinFunctionParity.assertParity( + this::environment, + "SELECT IF(c,a,b), IF(c, IF(a,b,c), a), IF(c, TRUE, FALSE) FROM src"); + } + + @Test + void predicatesAndUnselectedDivisionRetainShortCircuiting() throws Exception { + BuiltinFunctionParity.assertParity( + this::environment, + "SELECT IF(n = 0, TRUE, 10 / n > 1) FROM src WHERE IF(c,a,b)"); + BuiltinFunctionParity.assertParity( + this::environment, + "SELECT IF(c,a, 10 / n > 1) FROM src WHERE n = 99"); + } + + private TableEnvironment environment() { + Boolean[] values = {true, false, null}; + List rows = new ArrayList<>(); + for (int i = 0; i < 5003; i++) { + rows.add(Row.of(values[i % 3], values[i / 3 % 3], values[i / 9 % 3], i % 3)); + } + return BuiltinFunctionParity.environment( + ROW(FIELD("c", BOOLEAN()), FIELD("a", BOOLEAN()), FIELD("b", BOOLEAN()), + FIELD("n", INT())), rows); + } +} diff --git a/src/test/java/tech/streamfusion/FlinkBooleanStringCastSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkBooleanStringCastSqlHarnessTest.java new file mode 100644 index 00000000..d1e00cc9 --- /dev/null +++ b/src/test/java/tech/streamfusion/FlinkBooleanStringCastSqlHarnessTest.java @@ -0,0 +1,33 @@ +package tech.streamfusion; + +import static org.apache.flink.table.api.DataTypes.*; + +import java.util.List; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.types.Row; +import org.junit.jupiter.api.Test; + +class FlinkBooleanStringCastSqlHarnessTest { + @Test + void castsPreserveCaseLengthsPaddingAndNulls() throws Exception { + BuiltinFunctionParity.assertParity( + this::environment, + "SELECT CAST(b AS STRING), CAST(b AS VARCHAR(2)), CAST(b AS VARCHAR(5)), " + + "CAST(b AS CHAR(2)), CAST(b AS CHAR(8)), TRY_CAST(b AS VARCHAR(1)) FROM src"); + } + + @Test + void consumersAndEmptyBatchesMatchFlink() throws Exception { + BuiltinFunctionParity.assertParity( + this::environment, + "SELECT CONCAT(CAST(b AS STRING), '!') FROM src WHERE CAST(b AS VARCHAR(1)) = 'T'"); + BuiltinFunctionParity.assertParity( + this::environment, "SELECT CAST(b AS CHAR(8)) FROM src WHERE n = 99"); + } + + private TableEnvironment environment() { + return BuiltinFunctionParity.environment( + ROW(FIELD("b", BOOLEAN()), FIELD("n", INT())), + List.of(Row.of(true, 0), Row.of(false, 1), Row.of(null, 2))); + } +} diff --git a/src/test/java/tech/streamfusion/FlinkBtrimSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkBtrimSqlHarnessTest.java index 1302ef65..6a7812ee 100644 --- a/src/test/java/tech/streamfusion/FlinkBtrimSqlHarnessTest.java +++ b/src/test/java/tech/streamfusion/FlinkBtrimSqlHarnessTest.java @@ -20,10 +20,9 @@ private static void parity(String sql) throws Exception { } @Test - void dynamicTrimSetsFallBack() throws Exception { - NativeParity.assertFallbackReasonContains( + void javaBackedDynamicTrimSetsMatchHost() throws Exception { + BuiltinFunctionParity.assertParity( TextTimeFunctionTestInputs::parameters, - "SELECT id, BTRIM(s, p) FROM inputs", - "literal trim set"); + "SELECT id, BTRIM(s, p) FROM inputs"); } } diff --git a/src/test/java/tech/streamfusion/FlinkCalcSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkCalcSqlHarnessTest.java index 2459ec0b..072bac1c 100644 --- a/src/test/java/tech/streamfusion/FlinkCalcSqlHarnessTest.java +++ b/src/test/java/tech/streamfusion/FlinkCalcSqlHarnessTest.java @@ -154,12 +154,16 @@ void floatToIntCastMatchesHost() throws Exception { FlinkCalcSqlHarnessTest::castEnvironment, "SELECT CAST(dbl AS INT) FROM c"); } + @Test + void parseUrlProjectionMatchesHost() throws Exception { + BuiltinFunctionParity.assertParity( + FlinkCalcSqlHarnessTest::environment, "SELECT PARSE_URL(s, 'HOST') FROM f"); + } + @Test void unsupportedProjectionFunctionFallsBack() throws Exception { - // A function the expression encoder does not admit makes the whole Calc fall back, and the - // fallback reason names the offending function (ticket 29). NativeParity.assertFallbackReasonContains( - FlinkCalcSqlHarnessTest::environment, "SELECT PARSE_URL(s, 'HOST') FROM f", "PARSE_URL"); + FlinkCalcSqlHarnessTest::environment, "SELECT TRY_CAST(s AS BOOLEAN) FROM f", "TRY_CAST"); } @Test @@ -354,16 +358,15 @@ void positionMatchesHost() throws Exception { @Test void absFloatMatchesHost() throws Exception { // ABS over a double expression (the E-notation literal forces DOUBLE; goes negative for some - // rows). Integer ABS stays on host (overflow edge). + // rows). NativeParity.assertParity( FlinkCalcSqlHarnessTest::environment, "SELECT ABS(v - 25.5E0) FROM f"); } @Test - void absIntegerFallsBack() throws Exception { - // Integer ABS is not admitted (INT_MIN overflow edge), so it falls back, naming ABS. - NativeParity.assertFallbackReasonContains( - FlinkCalcSqlHarnessTest::environment, "SELECT ABS(v) FROM f", "ABS"); + void absIntegerMatchesHost() throws Exception { + BuiltinFunctionParity.assertParity( + FlinkCalcSqlHarnessTest::environment, "SELECT ABS(v), ABS(v - 25) FROM f"); } @Test diff --git a/src/test/java/tech/streamfusion/FlinkCollectionStringBuiltinsSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkCollectionStringBuiltinsSqlHarnessTest.java new file mode 100644 index 00000000..b436ee52 --- /dev/null +++ b/src/test/java/tech/streamfusion/FlinkCollectionStringBuiltinsSqlHarnessTest.java @@ -0,0 +1,113 @@ +package tech.streamfusion; + +import static org.apache.flink.table.api.DataTypes.*; + +import java.util.ArrayList; +import java.util.List; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.types.Row; +import org.junit.jupiter.api.Test; + +class FlinkCollectionStringBuiltinsSqlHarnessTest { + @Test + void regexArraysRetainCapturesNullElementsAndInvalidPatterns() throws Exception { + tech.streamfusion.compat.FlinkTestCapabilities.requireSqlFunction("REGEXP_EXTRACT_ALL"); + BuiltinFunctionParity.assertParity(this::regexEnvironment, + "SELECT REGEXP_EXTRACT_ALL(s,p,n), REGEXP_EXTRACT_ALL(s,p), " + + "REGEXP_EXTRACT_ALL(s,p,n)[1], CARDINALITY(REGEXP_EXTRACT_ALL(s,p,n)) FROM src"); + BuiltinFunctionParity.assertParity(this::regexEnvironment, + "SELECT REGEXP_EXTRACT_ALL(s,p,n)[1] FROM src " + + "WHERE CARDINALITY(REGEXP_EXTRACT_ALL(s,p,n)) > 0"); + BuiltinFunctionParity.assertParity(this::regexEnvironment, + "SELECT REGEXP_EXTRACT_ALL(s,p,n) FROM src WHERE n = 99"); + } + + @Test + void mapsRetainDuplicateKeysMissingValuesAndRegexDelimiters() throws Exception { + BuiltinFunctionParity.assertParity(this::mapEnvironment, + "SELECT STR_TO_MAP(s), STR_TO_MAP(s,p,q), STR_TO_MAP(s,p,q)['k1'], " + + "CARDINALITY(STR_TO_MAP(s,p,q)), STR_TO_MAP(s,p,q)['missing'] FROM src"); + BuiltinFunctionParity.assertParity(this::mapEnvironment, + "SELECT STR_TO_MAP(s,p,q) FROM src WHERE STR_TO_MAP(s,p,q)['k1'] IS NOT NULL"); + BuiltinFunctionParity.assertParity(this::mapEnvironment, + "SELECT STR_TO_MAP(s,p,q) FROM src WHERE n = 99"); + } + + @Test + void invalidMapDelimiterRetainsJavaRegexFailureAndEmptyBatchBehavior() throws Exception { + java.util.function.Supplier input = () -> BuiltinFunctionParity.environment( + ROW(FIELD("s", STRING()), FIELD("p", STRING()), FIELD("n", INT())), + List.of(Row.of("k=v", "[", 0))); + NativeFailureParity.run(input, "SELECT STR_TO_MAP(s,p,'=') FROM src") + .assertFailure(java.util.regex.PatternSyntaxException.class, "Unclosed character class", + NativeFailureParity.Phase.ROW_EVALUATION, NativeFailureParity.Route.NATIVE); + BuiltinFunctionParity.assertParity(input, "SELECT STR_TO_MAP(s,p,'=') FROM src WHERE n = 99"); + } + + @Test + void nativeSinksDoNotConsumeBinaryTransportAsUtf8() { + var table = BuiltinFunctionParity.environment(ROW(FIELD("s", STRING())), List.of(Row.of("/w=="))); + table.executeSql("CREATE TABLE pq (v STRING) WITH " + + "('connector'='filesystem', 'path'='/tmp/unused-base64-sink', 'format'='parquet')"); + var scan = tech.streamfusion.planner.NativePlanner.install(table); + table.explainSql("INSERT INTO pq SELECT s FROM src"); + org.junit.jupiter.api.Assertions.assertTrue(scan.substitutions() > 0, scan::explainSummary); + table.explainSql("INSERT INTO pq SELECT FROM_BASE64(s) FROM src"); + org.junit.jupiter.api.Assertions.assertEquals(0, scan.substitutions()); + org.junit.jupiter.api.Assertions.assertTrue( + scan.fallbackReasons().stream().anyMatch(reason -> reason.contains("requires a row sink")), + scan::explainSummary); + } + + @Test + void existingStringMapInputsSurviveWholeCalcEvaluation() throws Exception { + java.util.Map values = new java.util.HashMap<>(); + values.put("k", "value"); + values.put("null", null); + values.put(null, "null key"); + BuiltinFunctionParity.assertParity(() -> BuiltinFunctionParity.environment( + ROW(FIELD("s", STRING()), FIELD("m", MAP(STRING(), STRING()))), + List.of(Row.of("k=v", values), Row.of("", java.util.Map.of()), Row.of(null, null))), + "SELECT STR_TO_MAP(s), m, m['k'] FROM src"); + } + + @Test + void utf16ConsumersStayWithinFlinkEvaluation() throws Exception { + tech.streamfusion.compat.FlinkTestCapabilities.requireSqlFunction("REGEXP_EXTRACT_ALL"); + BuiltinFunctionParity.assertParity(this::mapEnvironment, + "SELECT STR_TO_MAP(s,'','')['?'], " + + "REGEXP_EXTRACT_ALL(s,p,0)[1] = '?', " + + "CASE WHEN STR_TO_MAP(s,p,q)['k1'] = '?' THEN 1 ELSE 0 END FROM src"); + } + + private TableEnvironment regexEnvironment() { + List samples = List.of(Row.of("abcdeabde", "ab((c)|(.?))de", 2), + Row.of("100-200, 300-400", "(\\d+)-(\\d+)", 1), + Row.of("abc", "", 0), Row.of("abc", "z+", 0), + Row.of("abc", "(", 0), Row.of("abc", "(abc)", -1), + Row.of("abc", "(abc)", 2), Row.of("abc", "(?<=a)(b)", 1), + Row.of("abab", "(ab)\\1", 1), Row.of(null, "a", 0), + Row.of("abc", null, 0), Row.of("abc", "(a)", null), + Row.of("\ud83d\ude00", ".", 0)); + return BuiltinFunctionParity.environment( + ROW(FIELD("s", STRING()), FIELD("p", STRING()), FIELD("n", INT())), repeat(samples)); + } + + private TableEnvironment mapEnvironment() { + List samples = List.of(Row.of("k1=v1,k2=v2,k1=last", ",", "=", 0), + Row.of("k1:v1;k2: v2", ";", ":", 1), + Row.of("k1$$v1|k2$$ v2", "\\|", "\\$\\$", 2), + Row.of("k1=,k2,=v,", ",", "=", 3), Row.of("", ",", "=", 4), + Row.of("\ud83d\ude00", "", "", 5), Row.of(null, ",", "=", 6), + Row.of("k1=v1", null, "=", 7), Row.of("k1=v1", ",", null, 8)); + return BuiltinFunctionParity.environment( + ROW(FIELD("s", STRING()), FIELD("p", STRING()), FIELD("q", STRING()), FIELD("n", INT())), + repeat(samples)); + } + + private static List repeat(List samples) { + List rows = new ArrayList<>(); + for (int i = 0; i < 5003; i++) rows.add(Row.copy(samples.get(i % samples.size()))); + return rows; + } +} diff --git a/src/test/java/tech/streamfusion/FlinkDynamicTrimSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkDynamicTrimSqlHarnessTest.java new file mode 100644 index 00000000..45f96d0a --- /dev/null +++ b/src/test/java/tech/streamfusion/FlinkDynamicTrimSqlHarnessTest.java @@ -0,0 +1,52 @@ +package tech.streamfusion; + +import static org.apache.flink.table.api.DataTypes.*; + +import java.util.ArrayList; +import java.util.List; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.types.Row; +import org.junit.jupiter.api.Test; + +class FlinkDynamicTrimSqlHarnessTest { + @org.junit.jupiter.api.BeforeEach + void requireHostFunction() { + tech.streamfusion.compat.FlinkTestCapabilities.requireSqlFunction("BTRIM"); + } + + @Test + void perRowSetsPreserveWhitespaceUnicodeAndNullsAcrossBatches() throws Exception { + BuiltinFunctionParity.assertParity(this::environment, + "SELECT BTRIM(s,LEFT(c,1)), LTRIM(s,LEFT(c,1)), RTRIM(s,LEFT(c,1)) FROM src WHERE c <> ''"); + BuiltinFunctionParity.assertParity(this::environment, + "SELECT BTRIM(s,c), LTRIM(s,c), RTRIM(s,c), BTRIM(s,'ab') FROM src"); + BuiltinFunctionParity.assertParity(this::environment, + "SELECT BTRIM(s,c) FROM src WHERE BTRIM(s,c) = 'XYZ'"); + BuiltinFunctionParity.assertParity(this::environment, + "SELECT LTRIM(s,c), RTRIM(s,c) FROM src WHERE n = 99"); + } + + @Test + void computedEmptySetRetainsTheReleasedHostFailure() throws Exception { + NativeFailureParity.run(this::environment, "SELECT BTRIM(s,LEFT(c,1)) FROM src") + .assertFailure(ArithmeticException.class, "/ by zero", NativeFailureParity.Phase.ROW_EVALUATION, + NativeFailureParity.Route.NATIVE); + } + + @Test + void unprovenBinaryStringRepresentationsKeepSafeFallback() throws Exception { + NativeParity.assertFallbackReasonContains(FlinkExtremaBuiltinsSqlHarnessTest::binaryEnvironment, + "SELECT BTRIM(a,b), LTRIM(a,b), RTRIM(a,b) FROM src", "string representation"); + } + + private TableEnvironment environment() { + List samples = List.of(Row.of("aabaXYZbaa", "ab", 0), Row.of(" abaXYZba ", " ab", 1), + Row.of("\ud83d\ude00XYZ\ud83d\ude00", "\ud83d\ude00", 2), Row.of("abc", "", 3), + Row.of("", "abc", 4), Row.of("aaa", "a", 5), Row.of(null, "ab", 6), + Row.of("abc", null, 7), Row.of("\t abc\t", "\t ", 8)); + List rows = new ArrayList<>(); + for (int i = 0; i < 5003; i++) rows.add(Row.copy(samples.get(i % samples.size()))); + return BuiltinFunctionParity.environment( + ROW(FIELD("s", STRING()), FIELD("c", STRING()), FIELD("n", INT())), rows); + } +} diff --git a/src/test/java/tech/streamfusion/FlinkEltSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkEltSqlHarnessTest.java index 66d5d16f..ea0f46c2 100644 --- a/src/test/java/tech/streamfusion/FlinkEltSqlHarnessTest.java +++ b/src/test/java/tech/streamfusion/FlinkEltSqlHarnessTest.java @@ -22,7 +22,9 @@ void unverifiedOverloadsFallBack() throws Exception { "SELECT ELT(CAST(4294967297 AS BIGINT), s, f) FROM texts", "ELT"); NativeParity.assertFallbackReasonContains( - StringFunctionTestInputs::text, "SELECT ELT(i, X'AB', X'CD') FROM texts", "ELT"); + StringFunctionTestInputs::text, + "SELECT ELT(i, X'AB', X'CD') FROM texts", + "unsupported generated-expression result type BINARY(1)"); } private static void parity(String sql) throws Exception { diff --git a/src/test/java/tech/streamfusion/FlinkEndsWithSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkEndsWithSqlHarnessTest.java index 88825d07..21bf6aa0 100644 --- a/src/test/java/tech/streamfusion/FlinkEndsWithSqlHarnessTest.java +++ b/src/test/java/tech/streamfusion/FlinkEndsWithSqlHarnessTest.java @@ -24,11 +24,10 @@ void predicatesAndNonNullableArguments() throws Exception { } @Test - void unverifiedOverloadsFallBack() throws Exception { - NativeParity.assertFallbackReasonContains( + void binarySuffixMatchesHost() throws Exception { + BuiltinFunctionParity.assertParity( StringFunctionTestInputs::search, - "SELECT ENDSWITH(binary_value, X'FF') FROM searches", - "ENDSWITH requires"); + "SELECT ENDSWITH(binary_value, X'FF') FROM searches"); } private static void parity(String sql) throws Exception { diff --git a/src/test/java/tech/streamfusion/FlinkExactNumericBuiltinsSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkExactNumericBuiltinsSqlHarnessTest.java new file mode 100644 index 00000000..dc1963d9 --- /dev/null +++ b/src/test/java/tech/streamfusion/FlinkExactNumericBuiltinsSqlHarnessTest.java @@ -0,0 +1,42 @@ +package tech.streamfusion; + +import static org.apache.flink.table.api.DataTypes.*; + +import java.math.BigDecimal; +import java.util.List; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.types.Row; +import org.junit.jupiter.api.Test; + +class FlinkExactNumericBuiltinsSqlHarnessTest { + @Test + void exactUnaryFunctionsPreserveTypesNullsAndOverflow() throws Exception { + BuiltinFunctionParity.assertParity( + this::environment, + "SELECT -i, -n, -d, ABS(i), ABS(n), ABS(d), SIGN(i), SIGN(n), SIGN(d), " + + "FLOOR(i), FLOOR(n), CEIL(i), CEIL(n), TRUNCATE(i), TRUNCATE(n, -1) FROM src"); + } + + @Test + void decimalConsumersAndFilteredBatchesRetainExactValues() throws Exception { + BuiltinFunctionParity.assertParity( + this::environment, + "SELECT ABS(-d), ROUND(-d, 2), MOD(-d, 3), TRUNCATE(i, digits), " + + "TRUNCATE(n, digits) FROM src WHERE i IS NOT NULL"); + BuiltinFunctionParity.assertParity( + this::environment, "SELECT ABS(-d), SIGN(n) FROM src WHERE i = 99"); + } + + private TableEnvironment environment() { + return BuiltinFunctionParity.environment( + ROW(FIELD("i", INT()), FIELD("n", BIGINT()), FIELD("d", DECIMAL(38, 9)), + FIELD("digits", INT())), + List.of( + Row.of(-3, -4L, new BigDecimal("-12.345000000"), -1), + Row.of(Integer.MIN_VALUE, Long.MIN_VALUE, + new BigDecimal("-99999999999999999999999999999.999999999"), 0), + Row.of(Integer.MAX_VALUE, Long.MAX_VALUE, new BigDecimal("0.000000001"), 2), + Row.of(0, 0L, BigDecimal.ZERO.setScale(9), -2), + Row.of(null, null, null, null))); + } +} diff --git a/src/test/java/tech/streamfusion/FlinkExtremaBuiltinsSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkExtremaBuiltinsSqlHarnessTest.java new file mode 100644 index 00000000..aed39ffa --- /dev/null +++ b/src/test/java/tech/streamfusion/FlinkExtremaBuiltinsSqlHarnessTest.java @@ -0,0 +1,70 @@ +package tech.streamfusion; + +import static org.apache.flink.table.api.DataTypes.*; + +import java.math.BigDecimal; +import java.time.Instant; +import java.time.LocalDateTime; +import java.util.List; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.types.Row; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +class FlinkExtremaBuiltinsSqlHarnessTest { + @Test + void binaryBackedStringsRetainFlinkWithoutChangingTheirOrdering() throws Exception { + NativeParity.assertFallbackReasonContains( + FlinkExtremaBuiltinsSqlHarnessTest::binaryEnvironment, + "SELECT GREATEST(a,b), LEAST(a,b) FROM src", + "string representation"); + } + + static TableEnvironment binaryEnvironment() { + var env = org.apache.flink.streaming.api.environment.StreamExecutionEnvironment + .getExecutionEnvironment(); + env.setParallelism(1); + var table = org.apache.flink.table.api.bridge.java.StreamTableEnvironment.create(env); + var type = ROW(FIELD("a", STRING()), FIELD("b", STRING())); + var rowType = (org.apache.flink.table.types.logical.RowType) type.getLogicalType(); + List rows = List.of( + org.apache.flink.table.data.GenericRowData.of( + org.apache.flink.table.data.binary.BinaryStringData.fromBytes( + "\ud83d\ude00".getBytes(java.nio.charset.StandardCharsets.UTF_8)), + org.apache.flink.table.data.binary.BinaryStringData.fromBytes( + "\ue000".getBytes(java.nio.charset.StandardCharsets.UTF_8)))); + table.createTemporaryView("src", tech.streamfusion.compat.FlinkTestSources.fromData( + env, rows, org.apache.flink.table.runtime.typeutils.InternalTypeInfo.of(rowType)), + org.apache.flink.table.api.Schema.newBuilder().fromRowDataType(type).build()); + return table; + } + + @ParameterizedTest + @ValueSource(strings = {"UTC", "Asia/Shanghai", "America/Los_Angeles"}) + void runtimeStringsCoercionsAndNanosecondTimestamps(String zone) throws Exception { + BuiltinFunctionParity.assertParity( + () -> environment(zone), + "SELECT GREATEST(a,b), LEAST(a,b), GREATEST(i,d), LEAST(i,d), " + + "GREATEST(t,u), LEAST(t,u), GREATEST(lt,lu), LEAST(lt,lu) FROM src"); + BuiltinFunctionParity.assertParity( + () -> environment(zone), + "SELECT GREATEST(a,b), LEAST(i,d) FROM src WHERE GREATEST(i,d) > 0"); + } + + private TableEnvironment environment(String zone) { + var beforeEpoch = LocalDateTime.parse("1969-12-31T23:59:59.999999999"); + var future = LocalDateTime.parse("9999-12-31T23:59:59.999999999"); + return BuiltinFunctionParity.environment( + ROW(FIELD("a", STRING()), FIELD("b", STRING()), FIELD("i", INT()), + FIELD("d", DECIMAL(12,2)), FIELD("t", TIMESTAMP(9)), FIELD("u", TIMESTAMP(9)), + FIELD("lt", TIMESTAMP_LTZ(9)), FIELD("lu", TIMESTAMP_LTZ(9))), + List.of( + Row.of("abc", "def", 2, new BigDecimal("1.25"), beforeEpoch, future, + Instant.ofEpochSecond(-1, 999999999), Instant.ofEpochSecond(1)), + Row.of("\ud83d\ude00", "\ue000", -2, new BigDecimal("-1.25"), future, beforeEpoch, + Instant.ofEpochSecond(1), Instant.ofEpochSecond(-1, 999999999)), + Row.of(null, "z", null, new BigDecimal("0.00"), null, beforeEpoch, null, Instant.EPOCH), + Row.of("", null, 0, null, beforeEpoch, null, Instant.EPOCH, null)), zone); + } +} diff --git a/src/test/java/tech/streamfusion/FlinkFilterSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkFilterSqlHarnessTest.java index 0edee3c6..bf489310 100644 --- a/src/test/java/tech/streamfusion/FlinkFilterSqlHarnessTest.java +++ b/src/test/java/tech/streamfusion/FlinkFilterSqlHarnessTest.java @@ -110,12 +110,19 @@ void isNotNullFilterMatchesHost() throws Exception { } @Test - void unsupportedFunctionFallsBack() throws Exception { - // A function the expression encoder does not admit makes the whole filter fall back to the host. - NativeParity.assertFallback( + void absoluteValueFilterMatchesHost() throws Exception { + BuiltinFunctionParity.assertParity( FlinkFilterSqlHarnessTest::environment, "SELECT * FROM f WHERE ABS(v) > 20"); } + @Test + void unsupportedFunctionFallsBack() throws Exception { + NativeParity.assertFallbackReasonContains( + FlinkFilterSqlHarnessTest::environment, + "SELECT * FROM f WHERE TRY_CAST(CAST(v AS STRING) AS BOOLEAN)", + "TRY_CAST"); + } + @Test void filterCarriesTimestampColumnMatchesHost() throws Exception { // The row carries a TIMESTAMP column through the whole-row converter while filtering on another. diff --git a/src/test/java/tech/streamfusion/FlinkGreatestSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkGreatestSqlHarnessTest.java index 6deb6f92..ff39a7bb 100644 --- a/src/test/java/tech/streamfusion/FlinkGreatestSqlHarnessTest.java +++ b/src/test/java/tech/streamfusion/FlinkGreatestSqlHarnessTest.java @@ -30,11 +30,10 @@ private static void parity(String sql) throws Exception { } @Test - void unrestrictedUnicodeStringsFallBack() throws Exception { - NativeParity.assertFallbackReasonContains( + void javaBackedUnicodeStringsMatchHost() throws Exception { + BuiltinFunctionParity.assertParity( () -> TextTimeFunctionTestInputs.textRows("\ufffd", "\ud83d\ude00", null), - "SELECT id, GREATEST(s, '\ud83d\ude00') FROM inputs", - "ASCII-provable"); + "SELECT id, GREATEST(s, '\ud83d\ude00') FROM inputs"); } @Test diff --git a/src/test/java/tech/streamfusion/FlinkLeastSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkLeastSqlHarnessTest.java index ae4ca76f..2a6ec87c 100644 --- a/src/test/java/tech/streamfusion/FlinkLeastSqlHarnessTest.java +++ b/src/test/java/tech/streamfusion/FlinkLeastSqlHarnessTest.java @@ -26,11 +26,10 @@ private static void parity(String sql) throws Exception { } @Test - void unrestrictedUnicodeStringsFallBack() throws Exception { - NativeParity.assertFallbackReasonContains( + void javaBackedUnicodeStringsMatchHost() throws Exception { + BuiltinFunctionParity.assertParity( () -> TextTimeFunctionTestInputs.textRows("\ufffd", "\ud83d\ude00", null), - "SELECT id, LEAST(s, '\ud83d\ude00') FROM inputs", - "ASCII-provable"); + "SELECT id, LEAST(s, '\ud83d\ude00') FROM inputs"); } @Test diff --git a/src/test/java/tech/streamfusion/FlinkLtrimSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkLtrimSqlHarnessTest.java index 5b5cccb5..507d869b 100644 --- a/src/test/java/tech/streamfusion/FlinkLtrimSqlHarnessTest.java +++ b/src/test/java/tech/streamfusion/FlinkLtrimSqlHarnessTest.java @@ -19,10 +19,9 @@ void trimsLiteralUnicodeCharacterSetsAndPropagatesNulls() throws Exception { } @Test - void dynamicTrimSetsFallBack() throws Exception { - NativeParity.assertFallbackReasonContains( + void javaBackedDynamicTrimSetsMatchHost() throws Exception { + BuiltinFunctionParity.assertParity( TextTimeFunctionTestInputs::parameters, - "SELECT id, LTRIM(s, p) FROM inputs", - "literal trim set"); + "SELECT id, LTRIM(s, p) FROM inputs"); } } diff --git a/src/test/java/tech/streamfusion/FlinkParseUrlSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkParseUrlSqlHarnessTest.java new file mode 100644 index 00000000..2cb8fa04 --- /dev/null +++ b/src/test/java/tech/streamfusion/FlinkParseUrlSqlHarnessTest.java @@ -0,0 +1,40 @@ +package tech.streamfusion; + +import static org.apache.flink.table.api.DataTypes.*; + +import java.util.ArrayList; +import java.util.List; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.types.Row; +import org.junit.jupiter.api.Test; + +class FlinkParseUrlSqlHarnessTest { + @Test + void rawComponentsAndDynamicArgumentsMatchJavaUrl() throws Exception { + BuiltinFunctionParity.assertParity(this::environment, + "SELECT PARSE_URL(u,p), PARSE_URL(u,p,k), PARSE_URL(u,'QUERY','q'), " + + "PARSE_URL(u,'HOST'), PARSE_URL(u,'PATH'), PARSE_URL(u,'AUTHORITY') FROM src"); + } + + @Test + void consumersAndEmptyBatchesRetainNullAndRawQueryValues() throws Exception { + BuiltinFunctionParity.assertParity(this::environment, + "SELECT CONCAT(PARSE_URL(u,p), '!') FROM src " + + "WHERE PARSE_URL(u,'QUERY','q') = 'a%2Bb+c'"); + BuiltinFunctionParity.assertParity(this::environment, + "SELECT PARSE_URL(u,p,k) FROM src WHERE n = 99"); + } + + private TableEnvironment environment() { + String url = "http://user:pw@EXAMPLE.com:80/a/../b?q=a%2Bb+c&q=second#fragment"; + List rows = new ArrayList<>(); + for (String part : new String[] {"HOST", "PATH", "QUERY", "REF", "PROTOCOL", "FILE", + "AUTHORITY", "USERINFO", "host", "UNKNOWN"}) rows.add(Row.of(url, part, "q", 0)); + rows.addAll(List.of(Row.of(url, "QUERY", "missing", 1), Row.of(url, "QUERY", null, 2), + Row.of(url, null, "q", 3), Row.of(null, "HOST", "q", 4), + Row.of("not a url", "HOST", "q", 5), Row.of("http://example.org", "PATH", "q", 6), + Row.of("http://example.org/?q=&a.b=1", "QUERY", "a.b", 7))); + return BuiltinFunctionParity.environment(ROW(FIELD("u", STRING()), FIELD("p", STRING()), + FIELD("k", STRING()), FIELD("n", INT())), rows); + } +} diff --git a/src/test/java/tech/streamfusion/FlinkPatternPredicatesSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkPatternPredicatesSqlHarnessTest.java new file mode 100644 index 00000000..be7b1fba --- /dev/null +++ b/src/test/java/tech/streamfusion/FlinkPatternPredicatesSqlHarnessTest.java @@ -0,0 +1,58 @@ +package tech.streamfusion; + +import static org.apache.flink.table.api.DataTypes.*; + +import java.util.List; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.types.Row; +import org.junit.jupiter.api.Test; + +class FlinkPatternPredicatesSqlHarnessTest { + @Test + void dynamicEscapesNegationUnicodeAndNulls() throws Exception { + BuiltinFunctionParity.assertParity( + this::environment, + "SELECT s LIKE p ESCAPE e, s NOT LIKE p ESCAPE e, " + + "s SIMILAR TO p ESCAPE e, s NOT SIMILAR TO p ESCAPE e FROM src"); + BuiltinFunctionParity.assertParity( + this::environment, "SELECT s FROM src WHERE s LIKE p ESCAPE e"); + } + + @Test + void similarGrammarAndRuntimeFilteredBatches() throws Exception { + BuiltinFunctionParity.assertParity( + this::environment, + "SELECT s SIMILAR TO '(a|b)%', s NOT SIMILAR TO '_*', " + + "s LIKE 'a!_b' ESCAPE '!' FROM src"); + BuiltinFunctionParity.assertParity( + this::environment, "SELECT s LIKE p ESCAPE e FROM src WHERE n = 99"); + } + + @Test + void invalidDynamicEscapeFailsOnlyOnEvaluatedRows() throws Exception { + BuiltinFunctionParity.assertParity( + () -> invalidEnvironment(), + "SELECT n <> 0 AND s LIKE p ESCAPE e, n = 0 OR s LIKE p ESCAPE e FROM src"); + var comparison = NativeFailureParity.run(this::invalidEnvironment, + "SELECT s LIKE p ESCAPE e FROM src"); + comparison.assertFailure(RuntimeException.class, "Invalid escape character", + NativeFailureParity.Phase.ROW_EVALUATION, NativeFailureParity.Route.NATIVE); + } + + private TableEnvironment environment() { + return environment(List.of( + Row.of("a_b", "a!_b", "!", 0), Row.of("a%b", "a!%b", "!", 1), + Row.of("\u4e2d\ud83d\ude00", "_%", "!", 2), Row.of("", "", "!", 3), + Row.of(null, "%", "!", 4), Row.of("abc", null, "!", 5), + Row.of("abc", "%", null, 6))); + } + + private TableEnvironment invalidEnvironment() { + return environment(List.of(Row.of("a_b", "a!_b", "!!", 0))); + } + + private TableEnvironment environment(List rows) { + return BuiltinFunctionParity.environment( + ROW(FIELD("s", STRING()), FIELD("p", STRING()), FIELD("e", STRING()), FIELD("n", INT())), rows); + } +} diff --git a/src/test/java/tech/streamfusion/FlinkPrintfSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkPrintfSqlHarnessTest.java new file mode 100644 index 00000000..613eedb0 --- /dev/null +++ b/src/test/java/tech/streamfusion/FlinkPrintfSqlHarnessTest.java @@ -0,0 +1,46 @@ +package tech.streamfusion; + +import static org.apache.flink.table.api.DataTypes.*; + +import java.math.BigDecimal; +import java.util.List; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.types.Row; +import org.junit.jupiter.api.Test; + +class FlinkPrintfSqlHarnessTest { + @org.junit.jupiter.api.BeforeEach + void requireHostFunction() { + tech.streamfusion.compat.FlinkTestCapabilities.requireSqlFunction("PRINTF"); + } + + @Test + void dynamicFormatsExactArgumentsAndFailuresMatchFlink() throws Exception { + BuiltinFunctionParity.assertParity(this::environment, + "SELECT PRINTF(f,n,d,s), PRINTF('%1$04d/%2$.2f/%3$s',n,d,s), " + + "PRINTF('%1$x',n), PRINTF('%1$+020d',n) FROM src"); + } + + @Test + void utf16ConsumersStayInsideFlinkUntilTheFinalOutput() throws Exception { + BuiltinFunctionParity.assertParity(this::environment, + "SELECT PRINTF('%c',cp), PRINTF('%c',cp) = '?', " + + "PRINTF('%c',cp) LIKE '_', CHAR_LENGTH(PRINTF('%c',cp)), " + + "CASE WHEN PRINTF('%c',cp) = '?' THEN 1 ELSE 0 END FROM src"); + BuiltinFunctionParity.assertParity(this::environment, + "SELECT n FROM src WHERE PRINTF('%c',cp) = '?'"); + BuiltinFunctionParity.assertParity(this::environment, + "SELECT PRINTF(f,n,d,s) FROM src WHERE cp = 99"); + } + + private TableEnvironment environment() { + return BuiltinFunctionParity.environment( + ROW(FIELD("f", STRING()), FIELD("n", BIGINT()), FIELD("d", DECIMAL(20,3)), + FIELD("s", STRING()), FIELD("cp", INT())), + List.of(Row.of("%1$04d|%2$.2f|%3$s", 42L, new BigDecimal("12.345"), "text", 65), + Row.of("%1$d", Long.MIN_VALUE, new BigDecimal("-12.345"), "\u4e2d", 0xd800), + Row.of("%q", Long.MAX_VALUE, new BigDecimal("0.000"), "", 0x1f600), + Row.of("%9$s", 0L, null, null, 0x110000), + Row.of(null, null, null, null, null))); + } +} diff --git a/src/test/java/tech/streamfusion/FlinkRegexBuiltinsSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkRegexBuiltinsSqlHarnessTest.java new file mode 100644 index 00000000..cfa43a0c --- /dev/null +++ b/src/test/java/tech/streamfusion/FlinkRegexBuiltinsSqlHarnessTest.java @@ -0,0 +1,49 @@ +package tech.streamfusion; + +import static org.apache.flink.table.api.DataTypes.*; + +import java.util.ArrayList; +import java.util.List; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.types.Row; +import org.junit.jupiter.api.Test; + +class FlinkRegexBuiltinsSqlHarnessTest { + @org.junit.jupiter.api.BeforeEach + void requireHostFunctions() { + tech.streamfusion.compat.FlinkTestCapabilities.requireSqlFunction("REGEXP_COUNT"); + tech.streamfusion.compat.FlinkTestCapabilities.requireSqlFunction("REGEXP_SUBSTR"); + } + + @Test + void javaPatternsTypesNullsAndLiteralReplacementSpanBatches() throws Exception { + BuiltinFunctionParity.assertParity( + this::environment, + "SELECT REGEXP(s,p), REGEXP_REPLACE(s,p,r), REGEXP_COUNT(s,p), " + + "REGEXP_INSTR(s,p), REGEXP_SUBSTR(s,p) FROM src"); + } + + @Test + void filterAndComposedConsumersMatchFlink() throws Exception { + BuiltinFunctionParity.assertParity( + this::environment, + "SELECT REGEXP_SUBSTR(s,p), CONCAT(REGEXP_REPLACE(s,p,r), '!'), " + + "REGEXP_COUNT(s,p) + 1 FROM src WHERE REGEXP(s,p)"); + BuiltinFunctionParity.assertParity( + this::environment, + "SELECT REGEXP_REPLACE(s,p,r), REGEXP_INSTR(s,p) FROM src WHERE n = 99"); + } + + private TableEnvironment environment() { + List samples = List.of( + Row.of("abc123abc", "abc", "X", 0), Row.of("ab", "a(?=b)", "$1", 1), + Row.of("aab", "(a)\\1", "\\$1", 2), Row.of("\u4e2da\ud83d\ude00", "a", "", 3), + Row.of("abc", "", "X", 4), Row.of("abc", "[", "X", 5), + Row.of(null, "a", "X", 6), Row.of("abc", null, "X", 7), + Row.of("abc", "a", null, 8), Row.of("", "z", "X", 9)); + List rows = new ArrayList<>(); + for (int i = 0; i < 5003; i++) rows.add(Row.copy(samples.get(i % samples.size()))); + return BuiltinFunctionParity.environment( + ROW(FIELD("s", STRING()), FIELD("p", STRING()), FIELD("r", STRING()), FIELD("n", INT())), rows); + } +} diff --git a/src/test/java/tech/streamfusion/FlinkRtrimSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkRtrimSqlHarnessTest.java index 0f10f03e..d242e577 100644 --- a/src/test/java/tech/streamfusion/FlinkRtrimSqlHarnessTest.java +++ b/src/test/java/tech/streamfusion/FlinkRtrimSqlHarnessTest.java @@ -19,10 +19,9 @@ void trimsLiteralUnicodeCharacterSetsAndPropagatesNulls() throws Exception { } @Test - void dynamicTrimSetsFallBack() throws Exception { - NativeParity.assertFallbackReasonContains( + void javaBackedDynamicTrimSetsMatchHost() throws Exception { + BuiltinFunctionParity.assertParity( TextTimeFunctionTestInputs::parameters, - "SELECT id, RTRIM(s, p) FROM inputs", - "literal trim set"); + "SELECT id, RTRIM(s, p) FROM inputs"); } } diff --git a/src/test/java/tech/streamfusion/FlinkStartsWithSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkStartsWithSqlHarnessTest.java index bb78e9a3..d8a27dd8 100644 --- a/src/test/java/tech/streamfusion/FlinkStartsWithSqlHarnessTest.java +++ b/src/test/java/tech/streamfusion/FlinkStartsWithSqlHarnessTest.java @@ -24,11 +24,10 @@ void predicatesAndNonNullableArguments() throws Exception { } @Test - void unverifiedOverloadsFallBack() throws Exception { - NativeParity.assertFallbackReasonContains( + void binaryPrefixMatchesHost() throws Exception { + BuiltinFunctionParity.assertParity( StringFunctionTestInputs::search, - "SELECT STARTSWITH(binary_value, X'FF') FROM searches", - "STARTSWITH requires"); + "SELECT STARTSWITH(binary_value, X'FF') FROM searches"); } private static void parity(String sql) throws Exception { diff --git a/src/test/java/tech/streamfusion/FlinkStringHashSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkStringHashSqlHarnessTest.java index 0f19cdce..11ec838a 100644 --- a/src/test/java/tech/streamfusion/FlinkStringHashSqlHarnessTest.java +++ b/src/test/java/tech/streamfusion/FlinkStringHashSqlHarnessTest.java @@ -54,10 +54,10 @@ void nestedFunctionsAndFiltersMatchHost() throws Exception { } @Test - void dynamicSha2BitLengthFallsBack() throws Exception { - NativeParity.assertFallbackReasonContains( + void dynamicSha2BitLengthMatchesHost() throws Exception { + BuiltinFunctionParity.assertParity( FlinkStringHashSqlHarnessTest::environment, - "SELECT SHA2(s, bits) FROM strings", "literal bit length"); + "SELECT SHA2(s, bits) FROM strings"); } @ParameterizedTest diff --git a/src/test/java/tech/streamfusion/FlinkToBase64SqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkToBase64SqlHarnessTest.java index 98719396..1bb75c65 100644 --- a/src/test/java/tech/streamfusion/FlinkToBase64SqlHarnessTest.java +++ b/src/test/java/tech/streamfusion/FlinkToBase64SqlHarnessTest.java @@ -16,12 +16,11 @@ void literalsNullsAndNonNullableArguments() throws Exception { } @Test - void unverifiedOverloadsFallBack() throws Exception { - NativeParity.assertFallbackReasonContains( + void fromBase64NullableBranchesMatchHost() throws Exception { + BuiltinFunctionParity.assertParity( StringFunctionTestInputs::encodings, "SELECT FROM_BASE64(CASE WHEN s IS NULL THEN CAST(NULL AS STRING) ELSE 'YQ==' END) FROM" - + " encodings", - "FROM_BASE64"); + + " encodings"); } private static void parity(String sql) throws Exception { diff --git a/src/test/java/tech/streamfusion/FlinkUrlDecodeSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkUrlDecodeSqlHarnessTest.java index f17d86b5..b9a35fe4 100644 --- a/src/test/java/tech/streamfusion/FlinkUrlDecodeSqlHarnessTest.java +++ b/src/test/java/tech/streamfusion/FlinkUrlDecodeSqlHarnessTest.java @@ -20,9 +20,9 @@ void urlDecoderMatchesJavaAcrossByteAndUnicodeDigitBoundaries() throws Exception } @Test - void unverifiedOverloadsFallBack() throws Exception { - NativeParity.assertFallbackReasonContains( - StringFunctionTestInputs::text, "SELECT PARSE_URL(u, 'HOST') FROM texts", "PARSE_URL"); + void parseUrlHostMatchesHost() throws Exception { + BuiltinFunctionParity.assertParity( + StringFunctionTestInputs::text, "SELECT PARSE_URL(u, 'HOST') FROM texts"); } private static void parity(String sql) throws Exception { diff --git a/src/test/java/tech/streamfusion/ScalarFunctionBenchmark.java b/src/test/java/tech/streamfusion/ScalarFunctionBenchmark.java index 26d9aa42..789a8dc3 100644 --- a/src/test/java/tech/streamfusion/ScalarFunctionBenchmark.java +++ b/src/test/java/tech/streamfusion/ScalarFunctionBenchmark.java @@ -58,6 +58,19 @@ String ddl() { private static final List SCALAR_FUNCTIONS = List.of( + new Query("EXACT_ABS_BIGINT", "bigint", "ABS(n)", "BIGINT"), + new Query("EXACT_SIGN_DECIMAL", "tt_decimal", "SIGN(n)", "DECIMAL(38,9)"), + new Query("GREATEST_RUNTIME_STRING", "text", "GREATEST(s, 'm')", "STRING"), + new Query("IF_BOOLEAN", "integer", "IF(n > 0, TRUE, FALSE)", "BOOLEAN"), + new Query("BOOLEAN_TO_STRING", "integer", "CAST(n > 0 AS STRING)", "STRING"), + new Query("LIKE_ESCAPE", "text", "s LIKE '%!_%' ESCAPE '!'", "BOOLEAN"), + new Query("REGEXP_COUNT", "text", "REGEXP_COUNT(s, 'a')", "INT"), + new Query("PARSE_URL", "text", "PARSE_URL(CONCAT('http://example.org/', s), 'PATH')", "STRING"), + new Query("PRINTF_BIGINT", "bigint", "PRINTF('n=%020d', n)", "STRING"), + new Query("BTRIM_DYNAMIC", "text", "BTRIM(s, LEFT(s, 1))", "STRING"), + new Query("IS_ALPHA", "text", "IS_ALPHA(s)", "BOOLEAN"), + new Query("STARTSWITH_BINARY", "tt_bytes", "STARTSWITH(b,b)", "BOOLEAN"), + new Query("REGEXP_EXTRACT_ALL", "text", "REGEXP_EXTRACT_ALL(s, '(a)', 1)", "ARRAY"), new Query("HASH_CODE_STRING", "text", "HASH_CODE(s)", "INT"), new Query("HASH_CODE_BIGINT", "bigint", "HASH_CODE(n)", "INT"), new Query("HASH_CODE_DECIMAL", "tt_decimal", "HASH_CODE(n)", "INT"), diff --git a/src/test/java/tech/streamfusion/planner/IfAdmissionTest.java b/src/test/java/tech/streamfusion/planner/IfAdmissionTest.java index 61718fab..02971152 100644 --- a/src/test/java/tech/streamfusion/planner/IfAdmissionTest.java +++ b/src/test/java/tech/streamfusion/planner/IfAdmissionTest.java @@ -1,5 +1,6 @@ package tech.streamfusion.planner; +import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import java.util.List; @@ -11,13 +12,24 @@ class IfAdmissionTest { @Test - void unregisteredHostOverloadsStayOutsideNativeCaseLowering() { + void booleanBranchesUseNativeCaseLowering() { + var types = new JavaTypeFactoryImpl(); + var rex = new RexBuilder(types); + var type = types.createSqlType(SqlTypeName.BOOLEAN); + var condition = rex.makeInputRef(type, 0); + var left = rex.makeInputRef(type, 1); + var right = rex.makeInputRef(type, 2); + var call = rex.makeCall(type, FlinkSqlOperatorTable.IF, List.of(condition, left, right)); + assertNotNull(RexExpression.encodeProjections(List.of(call), List.of("v"))); + } + + @Test + void unverifiedResultTypesStayOutsideNativeCaseLowering() { var types = new JavaTypeFactoryImpl(); var rex = new RexBuilder(types); var condition = rex.makeInputRef(types.createSqlType(SqlTypeName.BOOLEAN), 0); for (var type : List.of( - types.createSqlType(SqlTypeName.BOOLEAN), types.createSqlType(SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE, 3), types.createArrayType(types.createSqlType(SqlTypeName.INTEGER), -1))) { var branch = rex.makeInputRef(type, 1); diff --git a/src/test/java/tech/streamfusion/planner/JsonStringIdentityTest.java b/src/test/java/tech/streamfusion/planner/JsonStringIdentityTest.java index 8bd9f5b4..30382ef8 100644 --- a/src/test/java/tech/streamfusion/planner/JsonStringIdentityTest.java +++ b/src/test/java/tech/streamfusion/planner/JsonStringIdentityTest.java @@ -17,6 +17,18 @@ import org.junit.jupiter.api.Test; class JsonStringIdentityTest { + @Test + void formattedCharactersKeepTheirUtf16IdentityForConsumers() { + var types = new JavaTypeFactoryImpl(); + var rex = new RexBuilder(types); + var function = new org.apache.calcite.sql.SqlFunction("PRINTF", + org.apache.calcite.sql.SqlKind.OTHER_FUNCTION, null, null, null, + org.apache.calcite.sql.SqlFunctionCategory.STRING); + var call = rex.makeCall(types.createSqlType(SqlTypeName.VARCHAR), function, + List.of(rex.makeLiteral("%c"), rex.makeInputRef(types.createSqlType(SqlTypeName.INTEGER), 0))); + assertTrue(JsonStringIdentity.containsSensitiveString(call)); + } + @Test void onlyFinalRootsCanExposeJavaStringsWithoutAnotherConsumer() { var types = new JavaTypeFactoryImpl(); diff --git a/src/test/java/tech/streamfusion/planner/NativePlannerTest.java b/src/test/java/tech/streamfusion/planner/NativePlannerTest.java index 7a1acd3f..61e6d94f 100644 --- a/src/test/java/tech/streamfusion/planner/NativePlannerTest.java +++ b/src/test/java/tech/streamfusion/planner/NativePlannerTest.java @@ -102,18 +102,32 @@ void substitutesNativeOperatorForDoublingProjection() throws Exception { } @Test - void leavesUnsupportedProjectionToHostEngine() throws Exception { + void substitutesNativeAbsoluteValueProjection() throws Exception { TableEnvironment tEnv = TableEnvironment.create(EnvironmentSettings.inStreamingMode()); PhysicalPlanScan scan = NativePlanner.install(tEnv); - // ABS is not an admitted expression op, so the whole projection falls back to the host. List result = - collectInts(tEnv, "SELECT ABS(c0) AS a FROM (VALUES (3), (4), (5)) AS t(c0)"); + collectInts(tEnv, "SELECT ABS(c0) AS a FROM (VALUES (-3), (4), (-5)) AS t(c0)"); - assertEquals(0, scan.substitutions(), "an unsupported projection should not be substituted"); + assertTrue(scan.substitutions() > 0, "native operator was not substituted in"); assertEquals(List.of(3, 4, 5), result); } + @Test + void leavesUnsupportedProjectionToHostEngine() throws Exception { + TableEnvironment tEnv = TableEnvironment.create(EnvironmentSettings.inStreamingMode()); + PhysicalPlanScan scan = NativePlanner.install(tEnv); + List result = + collectInts( + tEnv, + "SELECT CASE WHEN TRY_CAST(CAST(c0 AS STRING) AS BOOLEAN) THEN c0 ELSE -c0 END" + + " FROM (VALUES (3), (4), (5)) AS t(c0)"); + + assertEquals(0, scan.substitutions(), "an unsupported projection should not be substituted"); + assertTrue(scan.fallbackReasons().stream().anyMatch(reason -> reason.contains("TRY_CAST"))); + assertEquals(List.of(-5, -4, -3), result); + } + /** * A self-join reads the same table twice. Sub-plan reuse stays enabled under the native planner, * scoped by digest barriers: the rowwise prefix (the scan) merges — the plan shows a {@code diff --git a/streamfusion-paimon/src/test/java/tech/streamfusion/paimon/PaimonSourceSharingTest.java b/streamfusion-paimon/src/test/java/tech/streamfusion/paimon/PaimonSourceSharingTest.java index 38bb65a0..f5080900 100644 --- a/streamfusion-paimon/src/test/java/tech/streamfusion/paimon/PaimonSourceSharingTest.java +++ b/streamfusion-paimon/src/test/java/tech/streamfusion/paimon/PaimonSourceSharingTest.java @@ -19,6 +19,9 @@ import tech.streamfusion.NativePlannerTestEnvironment; class PaimonSourceSharingTest { + private static final String FALLBACK_ID_EXPRESSION = + "IF(TRY_CAST(CAST(id AS STRING) AS BOOLEAN), -id, id)"; + @ParameterizedTest @ValueSource(strings = {"parquet", "orc"}) void sharedWatermarksCloseEveryBranchWindow(String format) throws Exception { @@ -121,13 +124,25 @@ void sharesProjectionsAcrossSinksButKeepsDifferentScansSeparate(String format) t assertEquals(3, occurrences(plan, "NativePaimonSource("), plan); sql.getConfig().set(option, "true"); } + var withAbs = sql.createStatementSet(); + withAbs.addInsertSql("INSERT INTO out1 SELECT v FROM t"); + withAbs.addInsertSql("INSERT INTO out2 SELECT n FROM t"); + withAbs.addInsertSql("INSERT INTO out3 SELECT ABS(id) FROM t"); + String absPlan = withAbs.compilePlan().explain(); + assertTrue(absPlan.contains("NativeShare(consumers=[3]"), absPlan); + assertEquals(1, occurrences(absPlan, "NativePaimonSource("), absPlan); + var mixed = sql.createStatementSet(); mixed.addInsertSql("INSERT INTO out1 SELECT v FROM t"); mixed.addInsertSql("INSERT INTO out2 SELECT n FROM t"); - mixed.addInsertSql("INSERT INTO out3 SELECT ABS(id) FROM t"); + mixed.addInsertSql("INSERT INTO out3 SELECT " + FALLBACK_ID_EXPRESSION + " FROM t"); String mixedPlan = mixed.compilePlan().explain(); assertTrue(mixedPlan.contains("NativeShare(consumers=[2]"), mixedPlan); assertTrue(mixedPlan.contains("TableSourceScan"), mixedPlan); + assertTrue( + tech.streamfusion.planner.NativePlanner.install(sql).fallbackReasons().stream() + .anyMatch(reason -> reason.contains("TRY_CAST")), + mixedPlan); var identical = sql.createStatementSet(); identical.addInsertSql("INSERT INTO out2 SELECT id FROM t"); @@ -201,12 +216,21 @@ void sharedSnapshotAndTailMatchFlink( statements.addInsertSql("INSERT INTO left_sink SELECT id, v FROM t"); statements.addInsertSql("INSERT INTO right_sink SELECT n, v FROM t"); statements.addInsertSql( - "INSERT INTO third_sink SELECT " + (fallback ? "ABS(id)" : "id") + " + n, v FROM t"); + "INSERT INTO third_sink SELECT " + + (fallback ? FALLBACK_ID_EXPRESSION : "ABS(id)") + + " + n, v FROM t"); var compiled = statements.compilePlan(); String plan = compiled.explain(); if (nativePlanner) { assertTrue(plan.contains("NativeShare(consumers=[" + (fallback ? 2 : 3) + "]"), plan); assertEquals(1, occurrences(plan, "NativePaimonSource("), plan); + if (fallback) { + assertTrue(plan.contains("TableSourceScan"), plan); + assertTrue( + tech.streamfusion.planner.NativePlanner.install(sql).fallbackReasons().stream() + .anyMatch(reason -> reason.contains("TRY_CAST")), + plan); + } } var job = compiled.execute().getJobClient().orElseThrow(); try {