diff --git a/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/FlinkDateTimeFormatConverter.java b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/FlinkDateTimeFormatConverter.java new file mode 100644 index 000000000..bb098ce18 --- /dev/null +++ b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/FlinkDateTimeFormatConverter.java @@ -0,0 +1,234 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.auron.flink.table.planner.converter; + +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; +import java.util.Optional; + +/** + * Translates a {@code java.text.SimpleDateFormat} pattern into the native parser's + * {@code strftime}-style format string, or reports that the pattern cannot be translated. + * + *

The native {@code Flink_UnixTimestamp} function replicates {@code SimpleDateFormat} lenient + * parsing only for a fixed set of numeric fields. This converter is the plan-time gate: it accepts + * a pattern only when every token maps to a field the native parser handles identically, and + * returns {@link Optional#empty()} otherwise so the whole {@code Calc} falls back to Flink's engine. + * + *

Accepted fields and their native specifiers: + *

+ * + *

Non-alphabetic characters are literals (a literal {@code %} is emitted as {@code %%}). + * {@code SimpleDateFormat} single-quote escaping applies, including the doubled {@code ''} that + * denotes a literal quote. Every other ASCII letter and every unlisted run length forces a fall + * back, because Java reserves all letters and the omitted forms either depend on a locale or on the + * clock at parse time. + * + *

Adjacency rule: run length is erased by the translation ({@code M} and {@code MM} both become + * {@code %m}), yet the native parser reads each numeric field at a canonical width (year 4; month, + * day, hour, minute, second 2) while Java's lenient scan window equals the run length. When two + * numeric fields are adjacent with no literal separator, those two widths must agree, so the left + * field's run length must equal its canonical width; otherwise the pattern falls back to avoid a + * silent divergence (e.g. {@code yyyyMd} on {@code 20201010} yields month 1 in Java but month 10 + * natively). + */ +public final class FlinkDateTimeFormatConverter { + + private FlinkDateTimeFormatConverter() { + // utility class + } + + /** + * Translates the given Java {@code SimpleDateFormat} pattern to the native {@code strftime}-style + * format string. + * + * @param javaPattern the Java date-time format pattern, never {@code null}. Null is rejected + * rather than reported as untranslatable: {@link Optional#empty()} means the user wrote a + * pattern outside the native surface and the {@code Calc} should fall back, whereas a null + * pattern means the caller never resolved one, which is a plumbing bug that a silent + * fallback would hide. + * @return the translated native format string, or {@link Optional#empty()} if any part of the + * pattern is outside the natively supported surface + * @throws NullPointerException if {@code javaPattern} is null + */ + public static Optional translate(String javaPattern) { + Objects.requireNonNull(javaPattern, "format pattern must not be null"); + List tokens = scan(javaPattern); + if (tokens == null) { + return Optional.empty(); + } + if (!adjacencyValid(tokens)) { + return Optional.empty(); + } + StringBuilder out = new StringBuilder(); + for (Token token : tokens) { + out.append(token.rendered); + } + return Optional.of(out.toString()); + } + + /** + * Walks the pattern into a token list, accumulating literal runs and emitting one token per + * pattern-letter field. Returns {@code null} on any unsupported letter, unsupported run length, + * or unterminated quote. + */ + private static List scan(String pattern) { + List tokens = new ArrayList<>(); + StringBuilder literal = new StringBuilder(); + int i = 0; + int n = pattern.length(); + while (i < n) { + char c = pattern.charAt(i); + if (c == '\'') { + if (i + 1 < n && pattern.charAt(i + 1) == '\'') { + literal.append('\''); + i += 2; + continue; + } + i++; + boolean closed = false; + while (i < n) { + char q = pattern.charAt(i); + if (q == '\'') { + if (i + 1 < n && pattern.charAt(i + 1) == '\'') { + literal.append('\''); + i += 2; + continue; + } + closed = true; + i++; + break; + } + literal.append(q); + i++; + } + if (!closed) { + return null; + } + } else if (isAsciiLetter(c)) { + int j = i; + while (j < n && pattern.charAt(j) == c) { + j++; + } + Field field = fieldFor(c, j - i); + if (field == null) { + return null; + } + flushLiteral(tokens, literal); + tokens.add(Token.field(field, j - i)); + i = j; + } else { + literal.append(c); + i++; + } + } + flushLiteral(tokens, literal); + return tokens; + } + + private static void flushLiteral(List tokens, StringBuilder literal) { + if (literal.length() > 0) { + tokens.add(Token.literal(literal.toString())); + literal.setLength(0); + } + } + + /** + * Checks the adjacency rule over the token list: for every field immediately followed by another + * field (no intervening literal), the left field's run length must equal its canonical width. + */ + private static boolean adjacencyValid(List tokens) { + for (int i = 0; i + 1 < tokens.size(); i++) { + Token left = tokens.get(i); + Token right = tokens.get(i + 1); + if (left.field != null && right.field != null && left.runLength != left.field.canonicalWidth) { + return false; + } + } + return true; + } + + private static boolean isAsciiLetter(char c) { + return (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z'); + } + + private static Field fieldFor(char letter, int runLength) { + switch (letter) { + case 'y': + return runLength == 3 || runLength == 4 ? Field.YEAR : null; + case 'M': + return runLength == 1 || runLength == 2 ? Field.MONTH : null; + case 'd': + return runLength == 1 || runLength == 2 ? Field.DAY : null; + case 'H': + return runLength == 1 || runLength == 2 ? Field.HOUR : null; + case 'm': + return runLength == 1 || runLength == 2 ? Field.MINUTE : null; + case 's': + return runLength == 1 || runLength == 2 ? Field.SECOND : null; + default: + return null; + } + } + + /** A supported numeric field: its native specifier and the width the native parser reads. */ + private enum Field { + YEAR("%Y", 4), + MONTH("%m", 2), + DAY("%d", 2), + HOUR("%H", 2), + MINUTE("%M", 2), + SECOND("%S", 2); + + private final String specifier; + private final int canonicalWidth; + + Field(String specifier, int canonicalWidth) { + this.specifier = specifier; + this.canonicalWidth = canonicalWidth; + } + } + + /** A scanned token: either a field (non-null {@link #field}) or a literal run. */ + private static final class Token { + private final Field field; + private final int runLength; + private final String rendered; + + private Token(Field field, int runLength, String rendered) { + this.field = field; + this.runLength = runLength; + this.rendered = rendered; + } + + static Token field(Field field, int runLength) { + return new Token(field, runLength, field.specifier); + } + + static Token literal(String raw) { + return new Token(null, 0, raw.replace("%", "%%")); + } + } +} diff --git a/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/FlinkNodeConverterUtils.java b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/FlinkNodeConverterUtils.java index 3e3fd209c..e7eda4ad8 100644 --- a/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/FlinkNodeConverterUtils.java +++ b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/FlinkNodeConverterUtils.java @@ -16,10 +16,14 @@ */ package org.apache.auron.flink.table.planner.converter; +import java.util.List; import org.apache.auron.flink.utils.SchemaConverters; +import org.apache.auron.protobuf.ArrowType; import org.apache.auron.protobuf.PhysicalCastNode; import org.apache.auron.protobuf.PhysicalExprNode; +import org.apache.auron.protobuf.PhysicalScalarFunctionNode; import org.apache.auron.protobuf.PhysicalTryCastNode; +import org.apache.auron.protobuf.ScalarFunction; import org.apache.calcite.rel.type.RelDataType; import org.apache.calcite.rel.type.RelDataTypeFactory; import org.apache.calcite.rel.type.RelDataTypeSystem; @@ -140,6 +144,27 @@ public static PhysicalExprNode wrapInCast(PhysicalExprNode expr, RelDataType tar .build(); } + /** + * Assembles a {@link PhysicalScalarFunctionNode} that routes to Auron's ext-function registry + * ({@link ScalarFunction#AuronExtFunctions}) by name. The arguments must already be converted to + * native expression nodes. + * + * @param name the registry name of the ext function (e.g. {@code "Flink_UnixTimestamp"}) + * @param args the already-converted argument expressions, in call order + * @param returnType the native Arrow return type of the function + * @return a {@link PhysicalExprNode} wrapping the scalar-function call + */ + public static PhysicalExprNode buildExtScalarFunctionNode( + String name, List args, ArrowType returnType) { + return PhysicalExprNode.newBuilder() + .setScalarFunction(PhysicalScalarFunctionNode.newBuilder() + .setName(name) + .setFun(ScalarFunction.AuronExtFunctions) + .addAllArgs(args) + .setReturnType(returnType)) + .build(); + } + private static boolean notApproxType(SqlTypeName typeName) { return typeName != SqlTypeName.FLOAT && typeName != SqlTypeName.DOUBLE; } diff --git a/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexCallConverter.java b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexCallConverter.java index 6d1282a68..f21a703ac 100644 --- a/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexCallConverter.java +++ b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexCallConverter.java @@ -16,10 +16,14 @@ */ package org.apache.auron.flink.table.planner.converter; +import java.time.ZoneId; +import java.util.Arrays; import java.util.EnumSet; import java.util.List; import java.util.Optional; import java.util.Set; +import org.apache.auron.protobuf.ArrowType; +import org.apache.auron.protobuf.EmptyMessage; import org.apache.auron.protobuf.PhysicalBinaryExprNode; import org.apache.auron.protobuf.PhysicalCaseNode; import org.apache.auron.protobuf.PhysicalExprNode; @@ -31,11 +35,13 @@ import org.apache.auron.protobuf.PhysicalWhenThen; import org.apache.calcite.rel.type.RelDataType; import org.apache.calcite.rex.RexCall; +import org.apache.calcite.rex.RexLiteral; import org.apache.calcite.rex.RexNode; import org.apache.calcite.sql.SqlKind; import org.apache.calcite.sql.fun.SqlLikeOperator; import org.apache.calcite.sql.type.SqlTypeUtil; import org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable; +import org.apache.flink.table.planner.utils.TableConfigUtils; /** * Converts a Calcite {@link RexCall} (operator expression) to an Auron native @@ -65,6 +71,14 @@ *

{@code CASE WHEN} (searched form) becomes a {@link PhysicalCaseNode} with * one {@link PhysicalWhenThen} per branch and a trailing else; each then/else * result is cast to the call's result type so all branches share one type. + * + *

{@code UNIX_TIMESTAMP} (matched by operator identity, like {@code TRY_CAST}) + * maps to a native {@code Flink_UnixTimestamp} ext-function call with arguments + * {@code [value, chronoFormat, zoneId]}. The single-argument form uses Flink's + * default format; the two-argument form is supported only when the format operand + * is a literal whose pattern {@link FlinkDateTimeFormatConverter} can translate, + * and the zero-argument form falls back. It additionally requires a session time + * zone the native function can resolve (see {@link #isNativelySupportedZone}). */ public class RexCallConverter implements FlinkRexNodeConverter { @@ -72,6 +86,9 @@ public class RexCallConverter implements FlinkRexNodeConverter { private static final Set BINARY_ARITHMETIC_KINDS = EnumSet.of(SqlKind.PLUS, SqlKind.MINUS, SqlKind.TIMES, SqlKind.DIVIDE, SqlKind.MOD); + /** Flink's default format for the single-argument {@code UNIX_TIMESTAMP(string)} form. */ + private static final String DEFAULT_UNIX_TIMESTAMP_FORMAT = "yyyy-MM-dd HH:mm:ss"; + /** All supported SqlKinds including unary and cast. */ private static final Set SUPPORTED_KINDS = EnumSet.of( SqlKind.PLUS, @@ -127,11 +144,14 @@ public Class getNodeClass() { @Override public boolean isSupported(RexNode node, ConverterContext context) { RexCall call = (RexCall) node; - // TRY_CAST has SqlKind OTHER_FUNCTION (not in SUPPORTED_KINDS), so it is - // matched by operator identity before the kind checks. + // TRY_CAST and UNIX_TIMESTAMP have SqlKind OTHER_FUNCTION (not in SUPPORTED_KINDS), so they + // are matched by operator identity before the kind checks. if (call.getOperator() == FlinkSqlOperatorTable.TRY_CAST) { return isCastTypeSupported(call.getOperands().get(0).getType(), call.getType()); } + if (call.getOperator() == FlinkSqlOperatorTable.UNIX_TIMESTAMP) { + return isUnixTimestampSupported(call, context); + } SqlKind kind = call.getKind(); if (!SUPPORTED_KINDS.contains(kind)) { return false; @@ -215,11 +235,14 @@ private static boolean isCastTypeSupported(RelDataType source, RelDataType targe @Override public PhysicalExprNode convert(RexNode node, ConverterContext context) { RexCall call = (RexCall) node; - // TRY_CAST has SqlKind OTHER_FUNCTION, which the switch below would route - // to the throwing default; match it by operator identity beforehand. + // TRY_CAST and UNIX_TIMESTAMP have SqlKind OTHER_FUNCTION, which the switch below would route + // to the throwing default; match them by operator identity beforehand. if (call.getOperator() == FlinkSqlOperatorTable.TRY_CAST) { return buildTryCast(call, context); } + if (call.getOperator() == FlinkSqlOperatorTable.UNIX_TIMESTAMP) { + return buildUnixTimestamp(call, context); + } SqlKind kind = call.getKind(); switch (kind) { case PLUS: @@ -390,6 +413,114 @@ private PhysicalExprNode buildTryCast(RexCall call, ConverterContext context) { return FlinkNodeConverterUtils.wrapInTryCast(operand, call.getType()); } + /** + * Returns {@code true} if this {@code UNIX_TIMESTAMP} call can run natively. The single-argument + * form uses Flink's default format; the two-argument form additionally requires the format + * operand to be a compile-time literal whose pattern the native parser handles identically + * (see {@link FlinkDateTimeFormatConverter}). Both also require a session time zone the native + * function can resolve (see {@link #isNativelySupportedZone}). + * + *

The zero-argument form is rejected here so it always falls back to Flink. It is a + * different function rather than a defaulted arity: it parses no input at all, and instead + * reads the wall clock once per record, so the operator is not deterministic and is never + * folded to a literal at plan time. It also has no value operand, which the native + * ext-function path needs to size its output against the batch. + */ + private static boolean isUnixTimestampSupported(RexCall call, ConverterContext context) { + List operands = call.getOperands(); + if (operands.isEmpty()) { + return false; + } + ZoneId zone = TableConfigUtils.getLocalTimeZone(context.getTableConfig()); + if (!isNativelySupportedZone(zone.getId())) { + return false; + } + if (operands.size() == 1) { + return true; + } + if (operands.size() == 2) { + RexNode format = operands.get(1); + if (!(format instanceof RexLiteral)) { + return false; + } + String javaFormat = ((RexLiteral) format).getValueAs(String.class); + return javaFormat != null + && FlinkDateTimeFormatConverter.translate(javaFormat).isPresent(); + } + return false; + } + + /** + * Returns {@code true} if the native function can resolve {@code zoneId}. It resolves a zone by + * exact-match lookup in the IANA time zone database, so two id families that Flink's + * {@code table.local-time-zone} accepts have to fall back: + * + *

+ * + *

The check has to happen at plan time: an unresolvable id fails inside the native call, + * and the Calc operator has no run-time fallback to catch it. + * + *

The membership test consults the JDK's copy of the database as a proxy for the one the + * native side resolves against. The two are updated independently and nothing in the build + * pins them together, so any further id family they stop agreeing on has to be excluded here + * the way {@code SystemV/*} is. + */ + private static boolean isNativelySupportedZone(String zoneId) { + return !zoneId.startsWith("SystemV/") && ZoneId.getAvailableZoneIds().contains(zoneId); + } + + /** + * Builds a native {@code Flink_UnixTimestamp} ext-function call. The native argument list is + * always {@code [value, chronoFormat, zoneId]}, and both of Flink's string-parsing arities are + * normalized onto that shape here: the value operand is converted recursively, the format is + * translated to the native specifier string (defaulting to Flink's + * {@code yyyy-MM-dd HH:mm:ss} when the call carries no format operand), and the session + * timezone is resolved at plan time. The native function therefore never has to default a + * missing format or timezone itself, and treats any other arity as a plumbing bug. + * + * @throws IllegalArgumentException if the call carries an arity this converter does not + * normalize, a format literal outside the natively supported surface, or a session time + * zone the native function cannot resolve (all unreachable via the factory, which gates on + * {@link #isSupported} first) + */ + private PhysicalExprNode buildUnixTimestamp(RexCall call, ConverterContext context) { + List operands = call.getOperands(); + if (operands.isEmpty() || operands.size() > 2) { + throw new IllegalArgumentException( + "UNIX_TIMESTAMP is native only in its 1-argument and 2-argument forms, got " + operands.size() + + " arguments"); + } + PhysicalExprNode value = convertOperand(operands.get(0), context); + + String javaFormat = operands.size() > 1 + ? ((RexLiteral) operands.get(1)).getValueAs(String.class) + : DEFAULT_UNIX_TIMESTAMP_FORMAT; + String chronoFormat = FlinkDateTimeFormatConverter.translate(javaFormat) + .orElseThrow(() -> new IllegalArgumentException("Unsupported UNIX_TIMESTAMP format: " + javaFormat)); + + ZoneId zone = TableConfigUtils.getLocalTimeZone(context.getTableConfig()); + if (!isNativelySupportedZone(zone.getId())) { + throw new IllegalArgumentException( + "UNIX_TIMESTAMP session time zone is not natively resolvable: " + zone.getId()); + } + + ArrowType bigIntType = ArrowType.newBuilder() + .setINT64(EmptyMessage.getDefaultInstance()) + .build(); + return FlinkNodeConverterUtils.buildExtScalarFunctionNode( + "Flink_UnixTimestamp", + Arrays.asList( + value, + RexLiteralConverter.stringLiteral(chronoFormat), + RexLiteralConverter.stringLiteral(zone.getId())), + bigIntType); + } + /** * Folds a Calcite n-ary {@code AND}/{@code OR} (operand count ≥ 2) into a * left-deep chain of binary nodes: {@code ((o0 op o1) op o2) ...}. Operands diff --git a/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexLiteralConverter.java b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexLiteralConverter.java index 90621a039..6bb520510 100644 --- a/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexLiteralConverter.java +++ b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexLiteralConverter.java @@ -21,6 +21,7 @@ import java.io.IOException; import java.math.BigDecimal; import java.util.EnumSet; +import java.util.Objects; import java.util.Set; import org.apache.arrow.memory.BufferAllocator; import org.apache.arrow.vector.VectorSchemaRoot; @@ -112,6 +113,41 @@ private static boolean isSupportedType(SqlTypeName typeName) { return SUPPORTED_TYPES.contains(typeName); } + /** + * Builds a literal expression node from a plain {@link String} that is not backed by a + * {@link RexNode}. Used for plan-time constants (such as a session timezone read from + * configuration) that must travel to the native side as a string argument. The value is + * serialized as a single-element {@code Utf8} Arrow record batch in IPC stream format, matching + * the encoding {@link #convert} produces for CHAR/VARCHAR {@link RexLiteral}s. + * + * @param value the constant string to encode, never {@code null}. A null value would mean the + * caller failed to resolve a plan-time constant, which is a plumbing bug rather than an + * unsupported expression, so it is rejected here instead of being encoded as a NULL literal. + * @return a {@link PhysicalExprNode} carrying the value as a native literal + * @throws NullPointerException if {@code value} is null + */ + public static PhysicalExprNode stringLiteral(String value) { + Objects.requireNonNull(value, "literal value must not be null"); + RowType rowType = RowType.of(new VarCharType(VarCharType.MAX_LENGTH)); + try (BufferAllocator allocator = + FlinkArrowUtils.getRootAllocator().newChildAllocator("literal", 0, Long.MAX_VALUE); + VectorSchemaRoot root = VectorSchemaRoot.create(FlinkArrowUtils.toArrowSchema(rowType), allocator)) { + + GenericRowData rowData = new GenericRowData(1); + rowData.setField(0, StringData.fromString(value)); + + FlinkArrowWriter writer = FlinkArrowWriter.create(root, rowType); + writer.write(rowData); + writer.finish(); + + return PhysicalExprNode.newBuilder() + .setLiteral(ScalarValue.newBuilder().setIpcBytes(ByteString.copyFrom(writeIpcBytes(root)))) + .build(); + } catch (IOException e) { + throw new IllegalStateException("Failed to serialize literal to Arrow IPC", e); + } + } + /** * Serializes the literal value as a single-element Arrow record batch in IPC stream format. * diff --git a/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java index 55ddad521..88870efd0 100644 --- a/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java +++ b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java @@ -189,7 +189,7 @@ protected Transformation translateToPlanInternal(PlannerBase planner, E return upstream; } - final Optional plan = tryBuildAuronPlan(inputRowType, outputRowType); + final Optional plan = tryBuildAuronPlan(config, inputRowType, outputRowType); if (!plan.isPresent()) { final boolean fallbackEnabled = AuronAdaptor.getInstance() @@ -263,13 +263,20 @@ private boolean fusedIntoSource(PlannerBase planner) { * composition throws — both signals are the same for the caller: fall back to Flink's codegen * Calc. * + *

The converter context is seeded with {@code config}, not {@code getPersistedConfig()}: the + * persisted config only carries the options this node declares in {@code @ExecNodeMetadata}, so + * it is empty here and would silently drop session settings such as {@code table.local-time-zone} + * that native converters need. The {@link ExecNodeConfig} in scope merges node-level overrides + * over the planner {@code TableConfig}, so it reflects the effective session configuration. + * + * @param config the effective exec-node configuration seeding the converter context * @param inputRowType the upstream row type used by the converter context * @param outputRowType the row type of this Calc's output * @return a composed plan, or empty if conversion failed */ - private Optional tryBuildAuronPlan(RowType inputRowType, RowType outputRowType) { - return NativePlanFusionBuilder.buildNativeCalcPlan( - getPersistedConfig(), projection, condition, inputRowType, outputRowType); + private Optional tryBuildAuronPlan( + ExecNodeConfig config, RowType inputRowType, RowType outputRowType) { + return NativePlanFusionBuilder.buildNativeCalcPlan(config, projection, condition, inputRowType, outputRowType); } /** diff --git a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/FlinkDateTimeFormatConverterTest.java b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/FlinkDateTimeFormatConverterTest.java new file mode 100644 index 000000000..f7f89b6eb --- /dev/null +++ b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/FlinkDateTimeFormatConverterTest.java @@ -0,0 +1,101 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.auron.flink.table.planner.converter; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.util.Optional; +import org.junit.jupiter.api.Test; + +/** Tests for {@link FlinkDateTimeFormatConverter}. */ +class FlinkDateTimeFormatConverterTest { + + /** The six supported letters translate at every accepted run length. */ + @Test + void testAcceptsSupportedPatterns() { + assertEquals(Optional.of("%Y-%m-%d %H:%M:%S"), translate("yyyy-MM-dd HH:mm:ss")); + assertEquals(Optional.of("%Y-%m-%d %H:%M:%S"), translate("yyyy-M-d H:m:s")); + assertEquals(Optional.of("%d-%m-%Y %H:%M:%S"), translate("dd-MM-yyyy HH:mm:ss")); + assertEquals(Optional.of("%Y/%m/%d"), translate("yyyy/MM/dd")); + assertEquals(Optional.of("%Y"), translate("yyy")); + assertEquals(Optional.of("%Y"), translate("yyyy")); + } + + /** Any letter outside the six-letter allowlist forces a fall back — Java reserves all letters, + * so a denylist would silently mis-treat an unknown letter as a literal. */ + @Test + void testRejectsUnsupportedLetters() { + assertFalse(translate("yyyy-MM-dd HH:mm:ss.SSS").isPresent()); + assertFalse(translate("yyyy-MMM-dd").isPresent()); + assertFalse(translate("yyyy-MM-dd E").isPresent()); + assertFalse(translate("yyyy-MM-dd a").isPresent()); + assertFalse(translate("yyyy-MM-dd z").isPresent()); + assertFalse(translate("yyyy-MM-dd Z").isPresent()); + assertFalse(translate("yyyy-MM-dd X").isPresent()); + } + + /** Run lengths outside each letter's accepted set fall back: the 2-digit year pivot is + * time-dependent, and over-long runs widen the lenient scan window past the native read. */ + @Test + void testRejectsUnsupportedRunLengths() { + assertFalse(translate("yy-MM-dd").isPresent()); + assertFalse(translate("yyyyy-MM-dd").isPresent()); + assertFalse(translate("yyyy-MM-dd HH:mm:sss").isPresent()); + } + + /** Single-quote escaping mirrors {@code SimpleDateFormat}, including the doubled {@code ''} + * literal quote and a literal {@code %} that must be doubled for the native format string. */ + @Test + void testQuoteEscaping() { + assertEquals(Optional.of("%Y-%m-%dT%H:%M:%S"), translate("yyyy-MM-dd'T'HH:mm:ss")); + assertEquals(Optional.of("%Y'%m"), translate("yyyy''MM")); + assertEquals(Optional.of("%Y%%%m"), translate("yyyy'%'MM")); + // An unterminated quote is not a valid pattern and falls back. + assertFalse(translate("yyyy'T").isPresent()); + } + + /** When two numeric fields are adjacent with no literal separator, the left field's run length + * must equal its native canonical width; otherwise Java's lenient window and the native + * canonical-width read diverge silently, so the pattern falls back. */ + @Test + void testAdjacencyRule() { + // Left field M has run length 1 but native reads canonical width 2 → fall back. + assertFalse(translate("yyyyMd").isPresent()); + // yyyyy is already rejected by run length, but is also an adjacency hazard (year 12020). + assertFalse(translate("yyyyyMMdd").isPresent()); + // Every left field already equals its canonical width → accepted. + assertEquals(Optional.of("%Y%m%d%H%M%S"), translate("yyyyMMddHHmmss")); + // Literal separators remove the adjacency between fields → accepted at run length 1. + assertEquals(Optional.of("%Y-%m-%d"), translate("yyyy-M-d")); + } + + /** + * Contract: a null pattern is a caller bug, not an untranslatable pattern, so it fails fast + * rather than returning empty. Returning empty would route it into the same fallback path as a + * legitimately unsupported pattern and hide the bug. + */ + @Test + void testNullPatternRejectedRatherThanReportedUntranslatable() { + assertThrows(NullPointerException.class, () -> translate(null)); + } + + private static Optional translate(String javaPattern) { + return FlinkDateTimeFormatConverter.translate(javaPattern); + } +} diff --git a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexCallConverterTest.java b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexCallConverterTest.java index a328653f8..b748dae06 100644 --- a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexCallConverterTest.java +++ b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexCallConverterTest.java @@ -18,12 +18,22 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import java.io.ByteArrayInputStream; +import java.io.IOException; import java.math.BigDecimal; import java.util.Arrays; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; +import org.apache.arrow.vector.VarCharVector; +import org.apache.arrow.vector.ipc.ArrowStreamReader; +import org.apache.auron.protobuf.ArrowType; import org.apache.auron.protobuf.PhysicalExprNode; +import org.apache.auron.protobuf.PhysicalScalarFunctionNode; import org.apache.auron.protobuf.PhysicalWhenThen; +import org.apache.auron.protobuf.ScalarFunction; import org.apache.calcite.jdbc.JavaTypeFactoryImpl; import org.apache.calcite.rel.type.RelDataType; import org.apache.calcite.rel.type.RelDataTypeFactory; @@ -72,7 +82,12 @@ void setUp() { new VarCharType() }, new String[] {"f0", "f1", "f2", "f3", "f4", "f5", "f6"}); - context = new ConverterContext(new Configuration(), null, getClass().getClassLoader(), inputType); + // The session time zone is pinned rather than left unset, because a converter that reads it + // rejects ids the native side cannot resolve. An unset zone resolves to the machine's + // default, which would make those cases depend on where the suite runs. + Configuration conf = new Configuration(); + conf.setString("table.local-time-zone", "UTC"); + context = new ConverterContext(conf, null, getClass().getClassLoader(), inputType); } @Test @@ -505,8 +520,116 @@ void testLikeWithExplicitEscapeIsUnsupported() { assertFalse(converter.isSupported(escapeLike, context)); } + // ---- UNIX_TIMESTAMP ---- + + @Test + void testUnixTimestampNodeShape() throws IOException { + RexNode call = makeCall(bigintType(), FlinkSqlOperatorTable.UNIX_TIMESTAMP, strRef(5)); + + PhysicalExprNode result = converter.convert(call, context); + + assertTrue(result.hasScalarFunction()); + PhysicalScalarFunctionNode fn = result.getScalarFunction(); + assertEquals("Flink_UnixTimestamp", fn.getName()); + assertEquals(ScalarFunction.AuronExtFunctions, fn.getFun()); + assertEquals(3, fn.getArgsCount(), "Args are always [value, chronoFormat, zoneId]"); + assertEquals(ArrowType.ArrowTypeEnumCase.INT64, fn.getReturnType().getArrowTypeEnumCase()); + assertTrue(fn.getArgs(0).hasColumn(), "First arg is the converted value operand"); + assertTrue(fn.getArgs(1).hasLiteral(), "Second arg is the format literal"); + assertTrue(fn.getArgs(2).hasLiteral(), "Third arg is the zone literal"); + // The default single-arg format translates to the native specifier string. + assertEquals("%Y-%m-%d %H:%M:%S", decodeStringLiteral(fn.getArgs(1))); + } + + @Test + void testUnixTimestampLiteralFormatIsSupportedAndTranslated() throws IOException { + RexNode fmt = REX_BUILDER.makeLiteral("yyyy/MM/dd"); + RexNode call = makeCall(bigintType(), FlinkSqlOperatorTable.UNIX_TIMESTAMP, strRef(5), fmt); + + assertTrue(converter.isSupported(call, context)); + PhysicalExprNode result = converter.convert(call, context); + assertEquals("%Y/%m/%d", decodeStringLiteral(result.getScalarFunction().getArgs(1))); + } + + /** The session timezone read from {@code TableConfig} reaches the node's third argument. This is + * the converter-level half of the config-propagation contract the shadowed {@code StreamExecCalc} + * enables. */ + @Test + void testUnixTimestampTimezonePropagatesToNode() throws IOException { + ConverterContext tzContext = contextWithZone("Asia/Shanghai"); + RexNode call = makeCall(bigintType(), FlinkSqlOperatorTable.UNIX_TIMESTAMP, strRef(5)); + + PhysicalExprNode result = converter.convert(call, tzContext); + + assertEquals( + "Asia/Shanghai", decodeStringLiteral(result.getScalarFunction().getArgs(2))); + } + + /** A fixed-offset session zone names an offset rather than a region: Flink accepts it, the + * native lookup cannot resolve it, so the gate rejects it and the builder refuses it outright + * rather than emitting a node that fails at run time. */ + @Test + void testUnixTimestampFixedOffsetZoneFallsBack() { + ConverterContext tzContext = contextWithZone("GMT-08:00"); + RexNode call = makeCall(bigintType(), FlinkSqlOperatorTable.UNIX_TIMESTAMP, strRef(5)); + + assertFalse(converter.isSupported(call, tzContext)); + assertThrows(IllegalArgumentException.class, () -> converter.convert(call, tzContext)); + } + + /** A legacy {@code SystemV/*} session zone still resolves in the JDK, so it reaches the gate + * looking like an ordinary region id, but the native lookup does not carry it. */ + @Test + void testUnixTimestampSystemVZoneFallsBack() { + RexNode call = makeCall(bigintType(), FlinkSqlOperatorTable.UNIX_TIMESTAMP, strRef(5)); + + assertFalse(converter.isSupported(call, contextWithZone("SystemV/PST8"))); + } + + /** The zero-argument form never reaches the native function: the gate rejects it so the Calc + * falls back, and the builder refuses it outright rather than reading an absent value operand. */ + @Test + void testUnixTimestampZeroArgFallsBack() { + RexNode call = makeCall(bigintType(), FlinkSqlOperatorTable.UNIX_TIMESTAMP); + + assertFalse(converter.isSupported(call, context)); + assertThrows(IllegalArgumentException.class, () -> converter.convert(call, context)); + } + + @Test + void testUnixTimestampNonLiteralFormatFallsBack() { + RexNode call = makeCall(bigintType(), FlinkSqlOperatorTable.UNIX_TIMESTAMP, strRef(5), strRef(6)); + + assertFalse(converter.isSupported(call, context)); + } + + @Test + void testUnixTimestampUnsupportedFormatTokenFallsBack() { + RexNode fmt = REX_BUILDER.makeLiteral("yyyy-MM-dd z"); + RexNode call = makeCall(bigintType(), FlinkSqlOperatorTable.UNIX_TIMESTAMP, strRef(5), fmt); + + assertFalse(converter.isSupported(call, context)); + } + // ---- Helpers ---- + /** Returns a copy of the shared context whose session time zone is {@code zoneId}. */ + private ConverterContext contextWithZone(String zoneId) { + Configuration conf = new Configuration(); + conf.setString("table.local-time-zone", zoneId); + return new ConverterContext(conf, null, getClass().getClassLoader(), context.getInputType()); + } + + private static String decodeStringLiteral(PhysicalExprNode node) throws IOException { + byte[] bytes = node.getLiteral().getIpcBytes().toByteArray(); + try (BufferAllocator alloc = new RootAllocator(Long.MAX_VALUE); + ArrowStreamReader reader = new ArrowStreamReader(new ByteArrayInputStream(bytes), alloc)) { + reader.loadNextBatch(); + VarCharVector vec = (VarCharVector) reader.getVectorSchemaRoot().getVector(0); + return vec.getObject(0).toString(); + } + } + private void assertComparison(org.apache.calcite.sql.SqlOperator op, String expectedOp) { RexNode call = makeCall(boolType(), op, makeIntRef(0), makeIntRef(0)); diff --git a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexLiteralConverterTest.java b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexLiteralConverterTest.java index b408272ba..c867903d0 100644 --- a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexLiteralConverterTest.java +++ b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexLiteralConverterTest.java @@ -18,6 +18,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import java.math.BigDecimal; @@ -168,4 +169,14 @@ void testUnsupportedTypeNotSupported() { assertFalse(converter.isSupported(tsLit, context)); } + + /** + * Contract: {@code stringLiteral} encodes plan-time constants the caller has already resolved, + * so a null value means the caller failed to resolve one. It is rejected at the boundary rather + * than encoded as a NULL literal, which would ship a silently wrong argument to the native side. + */ + @Test + void testStringLiteralRejectsNull() { + assertThrows(NullPointerException.class, () -> RexLiteralConverter.stringLiteral(null)); + } } diff --git a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/UnixTimestampOperatorIdentityTest.java b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/UnixTimestampOperatorIdentityTest.java new file mode 100644 index 000000000..01360316b --- /dev/null +++ b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/UnixTimestampOperatorIdentityTest.java @@ -0,0 +1,79 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.auron.flink.table.planner.converter; + +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertSame; + +import org.apache.auron.flink.table.AuronFlinkTableTestBase; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rex.RexCall; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.rex.RexShuttle; +import org.apache.calcite.sql.SqlOperator; +import org.apache.flink.table.api.Table; +import org.apache.flink.table.api.internal.TableImpl; +import org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable; +import org.apache.flink.table.planner.operations.PlannerQueryOperation; +import org.junit.jupiter.api.Test; + +/** + * Pins the operator-identity invariant the {@link RexCallConverter} dispatch relies on: Flink must + * resolve a SQL {@code UNIX_TIMESTAMP(...)} call to the singleton + * {@code FlinkSqlOperatorTable.UNIX_TIMESTAMP}. If a future Flink release routes it to a different + * operator (e.g. a bridging function), the reference-equality dispatch would silently miss and the + * function would permanently fall back with no error — this test turns that into a red build. + */ +class UnixTimestampOperatorIdentityTest extends AuronFlinkTableTestBase { + + @Test + void testSqlUnixTimestampResolvesToFlinkOperator() { + assertResolvesToUnixTimestampOperator("SELECT UNIX_TIMESTAMP(`ts`) FROM T1"); + assertResolvesToUnixTimestampOperator("SELECT UNIX_TIMESTAMP(`ts`, 'yyyy-MM-dd HH:mm:ss') FROM T1"); + } + + private void assertResolvesToUnixTimestampOperator(String sql) { + Table table = tableEnvironment.sqlQuery(sql); + RelNode tree = ((PlannerQueryOperation) ((TableImpl) table).getQueryOperation()).getCalciteTree(); + + SqlOperator operator = findUnixTimestampOperator(tree); + assertNotNull(operator, "SQL " + sql + " did not produce a UNIX_TIMESTAMP call"); + assertSame(FlinkSqlOperatorTable.UNIX_TIMESTAMP, operator); + } + + private static SqlOperator findUnixTimestampOperator(RelNode rel) { + SqlOperator[] holder = new SqlOperator[1]; + RexShuttle shuttle = new RexShuttle() { + @Override + public RexNode visitCall(RexCall call) { + if ("UNIX_TIMESTAMP".equals(call.getOperator().getName())) { + holder[0] = call.getOperator(); + } + return super.visitCall(call); + } + }; + walk(rel, shuttle); + return holder[0]; + } + + private static void walk(RelNode rel, RexShuttle shuttle) { + rel.accept(shuttle); + for (RelNode input : rel.getInputs()) { + walk(input, shuttle); + } + } +} diff --git a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/runtime/AuronFlinkCalcITCase.java b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/runtime/AuronFlinkCalcITCase.java index de52c25bc..7d70e0e07 100644 --- a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/runtime/AuronFlinkCalcITCase.java +++ b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/runtime/AuronFlinkCalcITCase.java @@ -18,6 +18,7 @@ import static org.assertj.core.api.Assertions.assertThat; +import java.time.ZoneId; import java.util.Arrays; import java.util.Collections; import java.util.Comparator; @@ -221,6 +222,43 @@ public void testFilterOrComparison() { assertThat(rows).isEqualTo(Arrays.asList(Row.of(1), Row.of(2))); } + /** UNIX_TIMESTAMP over the per-row {@code ts} string converts to the native ext function and + * yields the epoch seconds. The session timezone is set to Asia/Shanghai to make the result + * deterministic and to exercise timezone propagation into the native plan. */ + @Test + public void testUnixTimestamp() { + tableEnvironment.getConfig().setLocalTimeZone(ZoneId.of("Asia/Shanghai")); + List rows = CollectionUtil.iteratorToList(tableEnvironment + .executeSql("select UNIX_TIMESTAMP(`ts`) from T1") + .collect()); + rows.sort(Comparator.comparingLong(o -> (long) o.getField(0))); + assertThat(rows).isEqualTo(Arrays.asList(Row.of(1602259201L), Row.of(1602259202L), Row.of(1602259203L))); + } + + /** UNIX_TIMESTAMP yields the epoch seconds at zero offset when the session timezone is named + * without a region prefix. */ + @Test + public void testUnixTimestampUtcTimeZone() { + tableEnvironment.getConfig().setLocalTimeZone(ZoneId.of("UTC")); + List rows = CollectionUtil.iteratorToList(tableEnvironment + .executeSql("select UNIX_TIMESTAMP(`ts`) from T1") + .collect()); + rows.sort(Comparator.comparingLong(o -> (long) o.getField(0))); + assertThat(rows).isEqualTo(Arrays.asList(Row.of(1602288001L), Row.of(1602288002L), Row.of(1602288003L))); + } + + /** UNIX_TIMESTAMP yields the epoch seconds for the offset when the session timezone is a + * fixed-offset construction, a form Flink accepts that has no native equivalent. */ + @Test + public void testUnixTimestampFixedOffsetTimeZoneFallsBack() { + tableEnvironment.getConfig().setLocalTimeZone(ZoneId.of("GMT-08:00")); + List rows = CollectionUtil.iteratorToList(tableEnvironment + .executeSql("select UNIX_TIMESTAMP(`ts`) from T1") + .collect()); + rows.sort(Comparator.comparingLong(o -> (long) o.getField(0))); + assertThat(rows).isEqualTo(Arrays.asList(Row.of(1602316801L), Row.of(1602316802L), Row.of(1602316803L))); + } + /** A NOT LIKE filter keeps rows whose string does not match the pattern. */ @Test public void testFilterNotLike() { diff --git a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalcTest.java b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalcTest.java index a8d878824..9899d5457 100644 --- a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalcTest.java +++ b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalcTest.java @@ -22,11 +22,17 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import java.io.ByteArrayInputStream; +import java.io.IOException; import java.lang.reflect.Field; import java.lang.reflect.Method; import java.util.Arrays; import java.util.Collections; import java.util.List; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; +import org.apache.arrow.vector.VarCharVector; +import org.apache.arrow.vector.ipc.ArrowStreamReader; import org.apache.auron.flink.configuration.FlinkAuronConfiguration; import org.apache.auron.flink.runtime.operator.FlinkAuronCalcOperator; import org.apache.auron.flink.table.planner.UnsupportedFlinkNodeRecorder; @@ -34,6 +40,7 @@ import org.apache.auron.protobuf.ArrowType; import org.apache.auron.protobuf.FFIReaderExecNode; import org.apache.auron.protobuf.FilterExecNode; +import org.apache.auron.protobuf.PhysicalExprNode; import org.apache.auron.protobuf.PhysicalPlanNode; import org.apache.auron.protobuf.ProjectionExecNode; import org.apache.calcite.jdbc.JavaTypeFactoryImpl; @@ -50,6 +57,7 @@ import org.apache.flink.table.api.TableConfig; import org.apache.flink.table.data.RowData; import org.apache.flink.table.planner.delegation.PlannerBase; +import org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable; import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge; import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase; import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeConfig; @@ -61,6 +69,7 @@ import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.RawType; import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.table.types.logical.VarCharType; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -273,6 +282,55 @@ void testFallsBackWhenSchemaConversionThrows() throws Exception { assertEquals(1, UnsupportedFlinkNodeRecorder.peekEmitCount()); } + // ===================================================================== + // UNIX_TIMESTAMP integration + // ===================================================================== + + /** Contract: a zero-argument {@code UNIX_TIMESTAMP()} is unsupported, so a Calc projecting it + * falls back rather than converting to a native operator. */ + @Test + void testUnixTimestampZeroArgFallsBack() throws Exception { + Transformation stub = new FakeSourceTransformation(); + RexNode unixTs = + REX_BUILDER.makeCall(bigintType(), FlinkSqlOperatorTable.UNIX_TIMESTAMP, Collections.emptyList()); + CapturingTranslator node = new CapturingTranslator( + tableConfig, Arrays.asList(unixTs), null, inputProperty, RowType.of(new BigIntType()), "calc", stub); + wireFakeUpstream(node, TWO_INT_ROW); + + Transformation result = invokeTranslate(node); + + assertSame(stub, result); + assertEquals(1, node.fallbackCount); + } + + /** Contract: the effective {@link ExecNodeConfig} — not the empty persisted config — seeds the + * converter, so a session {@code table.local-time-zone} reaches the emitted native node's zone + * argument. Threading the persisted config here would silently default the zone. */ + @Test + void testSessionTimeZoneReachesNativePlan() throws Exception { + RowType inputRow = RowType.of(new LogicalType[] {new VarCharType()}, new String[] {"ts"}); + RexNode unixTs = + REX_BUILDER.makeCall(bigintType(), FlinkSqlOperatorTable.UNIX_TIMESTAMP, Arrays.asList(strRef(0))); + StreamExecCalc node = newCalc( + Arrays.asList(unixTs), null, RowType.of(new LogicalType[] {new BigIntType()}, new String[] {"u"})); + wireFakeUpstream(node, inputRow); + + Configuration conf = new Configuration(); + conf.setString("table.local-time-zone", "Asia/Shanghai"); + ExecNodeConfig zonedConfig = ExecNodeConfig.ofNodeConfig(conf, false); + + Transformation result = invokeTranslate(node, zonedConfig); + + PhysicalPlanNode plan = ((FlinkAuronCalcOperator) operatorOf(result)) + .getPhysicalPlanNodes() + .get(0); + PhysicalExprNode projected = plan.getProjection().getExpr(0); + assertTrue(projected.hasScalarFunction(), "UNIX_TIMESTAMP must convert to a scalar function node"); + assertEquals( + "Asia/Shanghai", + decodeStringLiteral(projected.getScalarFunction().getArgs(2))); + } + // ===================================================================== // Strict mode (FAIL_BACK_FLINK_ENGINE_ENABLED=false) // ===================================================================== @@ -396,6 +454,14 @@ private static RexNode intRef(int idx) { return REX_BUILDER.makeInputRef(intType(), idx); } + private static RelDataType varcharType() { + return TYPE_FACTORY.createSqlType(SqlTypeName.VARCHAR); + } + + private static RexNode strRef(int idx) { + return REX_BUILDER.makeInputRef(varcharType(), idx); + } + private static RexNode makeBinary( RelDataType returnType, org.apache.calcite.sql.SqlOperator op, RexNode left, RexNode right) { return REX_BUILDER.makeCall(returnType, op, Arrays.asList(left, right)); @@ -418,12 +484,16 @@ private static void wireFakeUpstream(StreamExecCalc node, RowType inputRowType) } private Transformation invokeTranslate(StreamExecCalc node) throws Exception { + return invokeTranslate(node, nodeConfig); + } + + private Transformation invokeTranslate(StreamExecCalc node, ExecNodeConfig config) throws Exception { Method m = ExecNodeBase.class.getDeclaredMethod( "translateToPlanInternal", PlannerBase.class, ExecNodeConfig.class); m.setAccessible(true); try { @SuppressWarnings("unchecked") - Transformation t = (Transformation) m.invoke(node, null, nodeConfig); + Transformation t = (Transformation) m.invoke(node, null, config); return t; } catch (java.lang.reflect.InvocationTargetException e) { if (e.getCause() instanceof RuntimeException) { @@ -433,6 +503,16 @@ private Transformation invokeTranslate(StreamExecCalc node) throws Exce } } + private static String decodeStringLiteral(PhysicalExprNode node) throws IOException { + byte[] bytes = node.getLiteral().getIpcBytes().toByteArray(); + try (BufferAllocator alloc = new RootAllocator(Long.MAX_VALUE); + ArrowStreamReader reader = new ArrowStreamReader(new ByteArrayInputStream(bytes), alloc)) { + reader.loadNextBatch(); + VarCharVector vec = (VarCharVector) reader.getVectorSchemaRoot().getVector(0); + return vec.getObject(0).toString(); + } + } + /** Extracts the {@link org.apache.flink.streaming.api.operators.StreamOperator} wrapped in * the {@link SimpleOperatorFactory} of a {@link OneInputTransformation}. */ private static Object operatorOf(Transformation result) {