From 2040a2bf0bbe18c15287a687b00ae9ec45831d54 Mon Sep 17 00:00:00 2001 From: weiqingy Date: Mon, 20 Jul 2026 20:03:20 -0700 Subject: [PATCH 1/3] [AURON #1863] Support native Flink UNIX_TIMESTAMP: converter integration Wire Flink's UNIX_TIMESTAMP (1-arg and 2-arg forms) to the native Flink_UnixTimestamp function. The converter matches the operator by reference identity (as TRY_CAST does), translates the supported subset of Java date-format patterns to the native format, resolves the session time zone at plan time, and builds the native scalar-function node. Formats outside the supported subset, non-literal format arguments, the 0-argument form, and format patterns whose lenient parse cannot be represented after translation all fall back to Flink's engine. The 0-argument form is rejected explicitly at both the gate and the builder, so a call that parses nothing and reads the wall clock per record can never reach a native function that expects a value operand to size its output against. Also propagate the effective node config (carrying table.local-time-zone) into the standalone-Calc converter path; the persisted config it previously used does not carry the session time zone, which would silently default the native evaluation to UTC. --- .../FlinkDateTimeFormatConverter.java | 227 ++++++++++++++++++ .../converter/FlinkNodeConverterUtils.java | 25 ++ .../planner/converter/RexCallConverter.java | 104 +++++++- .../converter/RexLiteralConverter.java | 31 +++ .../nodes/exec/stream/StreamExecCalc.java | 15 +- .../FlinkDateTimeFormatConverterTest.java | 90 +++++++ .../converter/RexCallConverterTest.java | 93 +++++++ .../UnixTimestampOperatorIdentityTest.java | 79 ++++++ .../table/runtime/AuronFlinkCalcITCase.java | 14 ++ .../nodes/exec/stream/StreamExecCalcTest.java | 82 ++++++- 10 files changed, 751 insertions(+), 9 deletions(-) create mode 100644 auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/FlinkDateTimeFormatConverter.java create mode 100644 auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/FlinkDateTimeFormatConverterTest.java create mode 100644 auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/UnixTimestampOperatorIdentityTest.java 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..c9216f4dd --- /dev/null +++ b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/FlinkDateTimeFormatConverter.java @@ -0,0 +1,227 @@ +/* + * 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.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 + * @return the translated native format string, or {@link Optional#empty()} if any part of the + * pattern is outside the natively supported surface + */ + public static Optional translate(String javaPattern) { + 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..512243876 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,13 @@ *

{@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. */ public class RexCallConverter implements FlinkRexNodeConverter { @@ -72,6 +85,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 +143,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); + } SqlKind kind = call.getKind(); if (!SUPPORTED_KINDS.contains(kind)) { return false; @@ -215,11 +234,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 +412,80 @@ 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 always can (it uses Flink's default format). The two-argument form can only when the + * format operand is a compile-time literal whose pattern the native parser handles identically + * (see {@link FlinkDateTimeFormatConverter}). + * + *

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) { + List operands = call.getOperands(); + if (operands.isEmpty()) { + 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; + } + + /** + * 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, or a format literal outside the natively supported surface (both 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()); + + 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..5552b5d93 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 @@ -112,6 +112,37 @@ 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 + * @return a {@link PhysicalExprNode} carrying the value as a native literal + */ + public static PhysicalExprNode stringLiteral(String value) { + 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..6c79a6e48 --- /dev/null +++ b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/FlinkDateTimeFormatConverterTest.java @@ -0,0 +1,90 @@ +/* + * 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 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")); + } + + 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..f9812389b 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; @@ -505,8 +515,91 @@ 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 { + Configuration conf = new Configuration(); + conf.setString("table.local-time-zone", "Asia/Shanghai"); + ConverterContext tzContext = + new ConverterContext(conf, null, getClass().getClassLoader(), context.getInputType()); + RexNode call = makeCall(bigintType(), FlinkSqlOperatorTable.UNIX_TIMESTAMP, strRef(5)); + + PhysicalExprNode result = converter.convert(call, tzContext); + + assertEquals( + "Asia/Shanghai", decodeStringLiteral(result.getScalarFunction().getArgs(2))); + } + + /** 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 ---- + 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/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..783ecaf64 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,19 @@ 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))); + } + /** 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) { From cccb1acb9aa39f66a55e9c9c018f18ae20b6a97d Mon Sep 17 00:00:00 2001 From: weiqingy Date: Tue, 4 Aug 2026 22:29:05 -0700 Subject: [PATCH 2/3] [AURON #1863] Reject null at the UNIX_TIMESTAMP converter's public entry points stringLiteral and translate both encode plan-time constants the caller has already resolved. A null argument means the caller never resolved one, which is a plumbing bug rather than an unsupported expression. translate rejects null instead of returning Optional.empty(): empty means the user wrote a pattern outside the native surface and the Calc should fall back, so reusing it for null would route a caller bug into the same silent fallback and hide it. --- .../converter/FlinkDateTimeFormatConverter.java | 9 ++++++++- .../table/planner/converter/RexLiteralConverter.java | 7 ++++++- .../converter/FlinkDateTimeFormatConverterTest.java | 11 +++++++++++ .../planner/converter/RexLiteralConverterTest.java | 11 +++++++++++ 4 files changed, 36 insertions(+), 2 deletions(-) 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 index c9216f4dd..bb098ce18 100644 --- 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 @@ -18,6 +18,7 @@ import java.util.ArrayList; import java.util.List; +import java.util.Objects; import java.util.Optional; /** @@ -63,11 +64,17 @@ private FlinkDateTimeFormatConverter() { * Translates the given Java {@code SimpleDateFormat} pattern to the native {@code strftime}-style * format string. * - * @param javaPattern the Java date-time format pattern + * @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(); 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 5552b5d93..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; @@ -119,10 +120,14 @@ private static boolean isSupportedType(SqlTypeName typeName) { * 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 + * @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); 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 index 6c79a6e48..f7f89b6eb 100644 --- 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 @@ -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 java.util.Optional; import org.junit.jupiter.api.Test; @@ -84,6 +85,16 @@ void testAdjacencyRule() { 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/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)); + } } From 9216d963abb39ab65400fa9abe9a92ce80a0deca Mon Sep 17 00:00:00 2001 From: weiqingy Date: Thu, 6 Aug 2026 23:49:43 -0700 Subject: [PATCH 3/3] [AURON #1863] Fall back when the session time zone is not natively resolvable The converter resolved the session time zone at plan time and passed ZoneId.getId() to the native Flink_UnixTimestamp function, which resolves zone ids by exact-match lookup in the IANA time zone database. Flink's table.local-time-zone also accepts fixed-offset constructions such as GMT-08:00, which name an offset rather than a region and have no entry in that database. Those reached the native call and failed it: Flink_UnixTimestamp: invalid timezone GMT-08:00 The failure lands after the plan has been converted, where the Calc operator has no run-time fallback, so the task died rather than degrading. Besides an explicit configuration, this was reachable through setLocalTimeZone(ZoneId.of("+08:00")), which normalizes to GMT+08:00, and through the default value, which resolves to ZoneId.systemDefault(). Reject such zones at plan time instead, so the Calc falls back to Flink and returns correct results, matching how an untranslatable format literal is already handled. The legacy SystemV/* aliases need an explicit exclusion because getAvailableZoneIds() carries them while the native lookup does not. Pin the session zone in the converter test setup as well: those tests inherited the machine's default zone, which the gate now reads. --- .../planner/converter/RexCallConverter.java | 51 ++++++++++++++++--- .../converter/RexCallConverterTest.java | 40 +++++++++++++-- .../table/runtime/AuronFlinkCalcITCase.java | 24 +++++++++ 3 files changed, 102 insertions(+), 13 deletions(-) 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 512243876..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 @@ -77,7 +77,8 @@ * {@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. + * 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 { @@ -149,7 +150,7 @@ public boolean isSupported(RexNode node, ConverterContext context) { return isCastTypeSupported(call.getOperands().get(0).getType(), call.getType()); } if (call.getOperator() == FlinkSqlOperatorTable.UNIX_TIMESTAMP) { - return isUnixTimestampSupported(call); + return isUnixTimestampSupported(call, context); } SqlKind kind = call.getKind(); if (!SUPPORTED_KINDS.contains(kind)) { @@ -414,9 +415,10 @@ private PhysicalExprNode buildTryCast(RexCall call, ConverterContext context) { /** * Returns {@code true} if this {@code UNIX_TIMESTAMP} call can run natively. The single-argument - * form always can (it uses Flink's default format). The two-argument form can only when the - * format operand is a compile-time literal whose pattern the native parser handles identically - * (see {@link FlinkDateTimeFormatConverter}). + * 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 @@ -424,11 +426,15 @@ private PhysicalExprNode buildTryCast(RexCall call, ConverterContext context) { * 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) { + 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; } @@ -444,6 +450,30 @@ private static boolean isUnixTimestampSupported(RexCall call) { 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 @@ -454,8 +484,9 @@ private static boolean isUnixTimestampSupported(RexCall call) { * 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, or a format literal outside the natively supported surface (both unreachable - * via the factory, which gates on {@link #isSupported} first) + * 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(); @@ -473,6 +504,10 @@ private PhysicalExprNode buildUnixTimestamp(RexCall call, ConverterContext conte .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()) 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 f9812389b..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 @@ -82,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 @@ -551,10 +556,7 @@ void testUnixTimestampLiteralFormatIsSupportedAndTranslated() throws IOException * enables. */ @Test void testUnixTimestampTimezonePropagatesToNode() throws IOException { - Configuration conf = new Configuration(); - conf.setString("table.local-time-zone", "Asia/Shanghai"); - ConverterContext tzContext = - new ConverterContext(conf, null, getClass().getClassLoader(), context.getInputType()); + ConverterContext tzContext = contextWithZone("Asia/Shanghai"); RexNode call = makeCall(bigintType(), FlinkSqlOperatorTable.UNIX_TIMESTAMP, strRef(5)); PhysicalExprNode result = converter.convert(call, tzContext); @@ -563,6 +565,27 @@ void testUnixTimestampTimezonePropagatesToNode() throws IOException { "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 @@ -590,6 +613,13 @@ void testUnixTimestampUnsupportedFormatTokenFallsBack() { // ---- 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); 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 783ecaf64..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 @@ -235,6 +235,30 @@ public void testUnixTimestamp() { 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() {