AnissaLiu commented on code in PR #13491: URL: https://github.com/apache/gluten/pull/13491#discussion_r4450597674
########## gluten-flink/ut/src/test/java/org/apache/gluten/rexnode/IntervalArithmeticConverterTest.java: ########## @@ -0,0 +1,268 @@ +/* + * 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.gluten.rexnode; + +import org.apache.gluten.rexnode.functions.RexCallConverter; +import org.apache.gluten.rexnode.functions.RexCallConverterFactory; + +import io.github.zhztheplayer.velox4j.expression.CallTypedExpr; +import io.github.zhztheplayer.velox4j.type.BigIntType; +import io.github.zhztheplayer.velox4j.type.DateType; +import io.github.zhztheplayer.velox4j.type.IntervalDayTimeType; +import io.github.zhztheplayer.velox4j.type.IntervalYearMonthType; +import io.github.zhztheplayer.velox4j.type.TimestampType; +import io.github.zhztheplayer.velox4j.variant.BigIntValue; +import io.github.zhztheplayer.velox4j.variant.Variant; + +import org.apache.flink.table.planner.calcite.FlinkRexBuilder; +import org.apache.flink.table.planner.calcite.FlinkTypeFactory; +import org.apache.flink.table.planner.calcite.FlinkTypeSystem; + +import org.apache.calcite.avatica.util.TimeUnit; +import org.apache.calcite.rex.RexCall; +import org.apache.calcite.rex.RexLiteral; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.sql.SqlIntervalQualifier; +import org.apache.calcite.sql.parser.SqlParserPos; +import org.apache.calcite.sql.type.SqlTypeName; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +import java.math.BigDecimal; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +public class IntervalArithmeticConverterTest { + + private static FlinkTypeFactory typeFactory; + private static FlinkRexBuilder rexBuilder; + private static RexConversionContext context; + + @BeforeAll + public static void setUp() { + typeFactory = + new FlinkTypeFactory( + Thread.currentThread().getContextClassLoader(), FlinkTypeSystem.INSTANCE); + rexBuilder = new FlinkRexBuilder(typeFactory); + context = new RexConversionContext(List.of("c0", "c1")); + } + + private RexNode tsInputRef() { + return rexBuilder.makeInputRef(typeFactory.createSqlType(SqlTypeName.TIMESTAMP, 3), 0); + } + + private RexNode dateInputRef() { + return rexBuilder.makeInputRef(typeFactory.createSqlType(SqlTypeName.DATE), 0); + } + + private RexNode bigintInputRef() { + return rexBuilder.makeInputRef(typeFactory.createSqlType(SqlTypeName.BIGINT), 1); + } + + private RexNode hourInterval(long millis) { + SqlIntervalQualifier qualifier = + new SqlIntervalQualifier(TimeUnit.HOUR, null, SqlParserPos.ZERO); + return rexBuilder.makeIntervalLiteral(BigDecimal.valueOf(millis), qualifier); + } + + private RexNode monthInterval(int months) { + SqlIntervalQualifier qualifier = + new SqlIntervalQualifier(TimeUnit.MONTH, null, SqlParserPos.ZERO); + return rexBuilder.makeIntervalLiteral(BigDecimal.valueOf(months), qualifier); + } + + @Test + public void logicalTypeConverterMapsIntervals() { + // The global mapping keeps the legacy behavior: day-time intervals map + // to BIGINT (millis) so existing paths stay untouched; interval velox + // types are only produced by the interval converters. + org.apache.flink.table.types.logical.DayTimeIntervalType dayTime = + new org.apache.flink.table.types.logical.DayTimeIntervalType( + org.apache.flink.table.types.logical.DayTimeIntervalType.DayTimeResolution + .DAY_TO_SECOND, + 2, + 6); + org.apache.flink.table.types.logical.YearMonthIntervalType yearMonth = + new org.apache.flink.table.types.logical.YearMonthIntervalType( + org.apache.flink.table.types.logical.YearMonthIntervalType.YearMonthResolution.YEAR, 2); + assertThat(org.apache.gluten.util.LogicalTypeConverter.toVLType(dayTime)) + .isInstanceOf(BigIntType.class); + assertThat(org.apache.gluten.util.LogicalTypeConverter.toVLType(yearMonth)) + .isInstanceOf(IntervalYearMonthType.class); + } + + @Test + public void toVariantReadsIntervalLiterals() { + RexLiteral hour = (RexLiteral) hourInterval(3600000); + Variant variant = RexNodeConverter.toVariant(hour); + assertThat(variant).isInstanceOf(BigIntValue.class); + assertThat(((BigIntValue) variant).getValue()).isEqualTo(3600000L); + } + + @Test + public void timestampPlusIntervalDayTime() { + RexCall call = + (RexCall) + rexBuilder.makeCall( + org.apache.calcite.sql.fun.SqlStdOperatorTable.PLUS, + tsInputRef(), + hourInterval(3600000)); + assertThat(call.getOperator().getName()).isEqualTo("+"); + RexCallConverter converter = RexCallConverterFactory.getConverter(call, context); + CallTypedExpr expr = (CallTypedExpr) converter.toTypedExpr(call, context); + assertThat(expr.getFunctionName()).isEqualTo("plus"); + assertThat(expr.getInputs()).hasSize(2); + assertThat(expr.getInputs().get(1).getReturnType()).isInstanceOf(IntervalDayTimeType.class); + } + + @Test + public void timestampPlusIntervalYearMonth() { + RexCall call = + (RexCall) + rexBuilder.makeCall( + org.apache.calcite.sql.fun.SqlStdOperatorTable.PLUS, + tsInputRef(), + monthInterval(1)); + RexCallConverter converter = RexCallConverterFactory.getConverter(call, context); + CallTypedExpr expr = (CallTypedExpr) converter.toTypedExpr(call, context); + assertThat(expr.getFunctionName()).isEqualTo("plus"); + assertThat(expr.getInputs().get(1).getReturnType()).isInstanceOf(IntervalYearMonthType.class); + } + + @Test + public void intervalOnLeftPlusIsFlipped() { + RexCall call = + (RexCall) + rexBuilder.makeCall( + org.apache.calcite.sql.fun.SqlStdOperatorTable.PLUS, + hourInterval(3600000), + tsInputRef()); + RexCallConverter converter = RexCallConverterFactory.getConverter(call, context); + CallTypedExpr expr = (CallTypedExpr) converter.toTypedExpr(call, context); + assertThat(expr.getFunctionName()).isEqualTo("plus"); + // Interval flipped to the right: velox only registers (datetime, interval). + assertThat(expr.getInputs().get(0).getReturnType()).isInstanceOf(TimestampType.class); + assertThat(expr.getInputs().get(1).getReturnType()).isInstanceOf(IntervalDayTimeType.class); + } + + @Test + public void datePlusIntervalDayTimeToTimestamp() { Review Comment: Thanks for catching this. Added pipeline coverage in 23d779db2 (ScalarFunctionsTest#testTimestampAddSubDayOnDate): it runs TIMESTAMPADD(HOUR, 5, d), TIMESTAMPADD(MINUTE, -90, d) and TIMESTAMPADD(SECOND, 30, d) plus the handwritten DATE + INTERVAL form through the real pipeline and asserts the TIMESTAMP results against Flink semantics (2017-11-29 05:00:00 / 2017-11-28 22:30:00 / 2017-11-29 00:00:30, including the negative amount crossing midnight). DATE_FORMAT with an explicit second-precision pattern keeps the expectation rendering-stable — the sub-day results are whole seconds by construction, so the pattern carries the full observable content. The DATE input is derived via CAST from a string column because the values connector round-trips rows through their string form and its formatter has no DATE parsing yet (pre-existing gap in the connector formatter, unrelated to this path; filed separately). We also re-ran it end-to-end on the cluster with a real filesystem DATE source ( CSV) against native Flink: identical output, exact match. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
