This is an automated email from the ASF dual-hosted git repository.
vernedeng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 09f39dff6b [INLONG-10767][SDK] Transform SQL support temporal
functions(Including hour, minute and second) (#10772)
09f39dff6b is described below
commit 09f39dff6bb4cc0e8c6acfe3e68d8c4a7222fc29
Author: yfsn666 <61183968+yfsn...@users.noreply.github.com>
AuthorDate: Wed Aug 14 11:59:01 2024 +0800
[INLONG-10767][SDK] Transform SQL support temporal functions(Including
hour, minute and second) (#10772)
* [INLONG-10767][SDK] Transform SQL support temporal functions(Including
hour, minute and second)
---
.../process/function/DateExtractFunction.java | 12 ++--
.../process/function/TimestampExtractFunction.java | 73 ++++++++++++++++++++++
.../transform/process/operator/OperatorTools.java | 22 +++++++
.../transform/process/parser/TimestampParser.java | 43 +++++++++++++
.../TestTransformTemporalFunctionsProcessor.java | 33 ++++++++++
5 files changed, 177 insertions(+), 6 deletions(-)
diff --git
a/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/function/DateExtractFunction.java
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/function/DateExtractFunction.java
index 47061c4dcf..d763b657e2 100644
---
a/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/function/DateExtractFunction.java
+++
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/function/DateExtractFunction.java
@@ -35,12 +35,12 @@ import java.util.Locale;
/**
* DateExtractFunction
* description:
- * - year(date)--returns the year from SQL date date
- * - quarter(date)--returns the quarter of a year (an integer between 1 and 4)
from SQL date date
- * - month(date)--returns the month of a year (an integer between 1 and 12)
from SQL date date
- * - week(date)--returns the week of a year (an integer between 1 and 53) from
SQL date date
- * - dayofyear(date)--returns the day of a year (an integer between 1 and 366)
from SQL date date
- * - dayofmonth(date)--returns the day of a month (an integer between 1 and
31) from SQL date date
+ * - year(date)--returns the year from SQL date
+ * - quarter(date)--returns the quarter of a year (an integer between 1 and 4)
from SQL date
+ * - month(date)--returns the month of a year (an integer between 1 and 12)
from SQL date
+ * - week(date)--returns the week of a year (an integer between 1 and 53) from
SQL date
+ * - dayofyear(date)--returns the day of a year (an integer between 1 and 366)
from SQL date
+ * - dayofmonth(date)--returns the day of a month (an integer between 1 and
31) from SQL date
*/
public class DateExtractFunction implements ValueParser {
diff --git
a/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/function/TimestampExtractFunction.java
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/function/TimestampExtractFunction.java
new file mode 100644
index 0000000000..6af01e675e
--- /dev/null
+++
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/function/TimestampExtractFunction.java
@@ -0,0 +1,73 @@
+/*
+ * 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.inlong.sdk.transform.process.function;
+
+import org.apache.inlong.sdk.transform.decode.SourceData;
+import org.apache.inlong.sdk.transform.process.Context;
+import org.apache.inlong.sdk.transform.process.operator.OperatorTools;
+import org.apache.inlong.sdk.transform.process.parser.ValueParser;
+
+import net.sf.jsqlparser.expression.Expression;
+import net.sf.jsqlparser.expression.Function;
+
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.util.List;
+
+/**
+ * TimestampExtractFunction
+ * description:
+ * - hour(timestamp)--returns the hour of a day (an integer between 0 and 23)
from SQL timestamp
+ * - minute(timestamp)--returns the minute of an hour (an integer between 0
and 59) from SQL timestamp
+ * - second(timestamp)--returns the second of a minute (an integer between 0
and 59) from SQL timestamp
+ */
+public class TimestampExtractFunction implements ValueParser {
+
+ private TimestampExtractFunctionType type;
+ private ValueParser timestampParser;
+
+ public enum TimestampExtractFunctionType {
+ HOUR, MINUTE, SECOND
+ }
+
+ public TimestampExtractFunction(TimestampExtractFunctionType type,
Function expr) {
+ this.type = type;
+ List<Expression> expressions = expr.getParameters().getExpressions();
+ timestampParser = OperatorTools.buildParser(expressions.get(0));
+ }
+
+ @Override
+ public Object parse(SourceData sourceData, int rowIndex, Context context) {
+ Object timestampObj = timestampParser.parse(sourceData, rowIndex,
context);
+ Timestamp timestamp = OperatorTools.parseTimestamp(timestampObj);
+ LocalDateTime localDateTime = timestamp.toLocalDateTime();
+ switch (type) {
+ // hour(between 0 and 23)
+ case HOUR:
+ return localDateTime.getHour();
+ // minute(between 0 and 59)
+ case MINUTE:
+ return localDateTime.getMinute();
+ // second(between 0 and 59)
+ case SECOND:
+ return localDateTime.getSecond();
+ default:
+ return null;
+ }
+ }
+}
diff --git
a/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/operator/OperatorTools.java
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/operator/OperatorTools.java
index b1c950720e..8c48aedfe7 100644
---
a/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/operator/OperatorTools.java
+++
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/operator/OperatorTools.java
@@ -36,6 +36,7 @@ import
org.apache.inlong.sdk.transform.process.function.SinFunction;
import org.apache.inlong.sdk.transform.process.function.SinhFunction;
import org.apache.inlong.sdk.transform.process.function.SqrtFunction;
import org.apache.inlong.sdk.transform.process.function.SubstringFunction;
+import
org.apache.inlong.sdk.transform.process.function.TimestampExtractFunction;
import org.apache.inlong.sdk.transform.process.function.ToDateFunction;
import org.apache.inlong.sdk.transform.process.parser.AdditionParser;
import org.apache.inlong.sdk.transform.process.parser.ColumnParser;
@@ -46,6 +47,7 @@ import
org.apache.inlong.sdk.transform.process.parser.MultiplicationParser;
import org.apache.inlong.sdk.transform.process.parser.ParenthesisParser;
import org.apache.inlong.sdk.transform.process.parser.StringParser;
import org.apache.inlong.sdk.transform.process.parser.SubtractionParser;
+import org.apache.inlong.sdk.transform.process.parser.TimestampParser;
import org.apache.inlong.sdk.transform.process.parser.ValueParser;
import net.sf.jsqlparser.expression.DateValue;
@@ -55,6 +57,7 @@ import net.sf.jsqlparser.expression.LongValue;
import net.sf.jsqlparser.expression.NotExpression;
import net.sf.jsqlparser.expression.Parenthesis;
import net.sf.jsqlparser.expression.StringValue;
+import net.sf.jsqlparser.expression.TimestampValue;
import net.sf.jsqlparser.expression.operators.arithmetic.Addition;
import net.sf.jsqlparser.expression.operators.arithmetic.Division;
import net.sf.jsqlparser.expression.operators.arithmetic.Multiplication;
@@ -72,6 +75,7 @@ import org.apache.commons.lang.ObjectUtils;
import java.math.BigDecimal;
import java.sql.Date;
+import java.sql.Timestamp;
import java.util.HashMap;
import java.util.Map;
@@ -112,6 +116,14 @@ public class OperatorTools {
functionMap.put("week", func -> new
DateExtractFunction(DateExtractFunctionType.WEEK, func));
functionMap.put("dayofyear", func -> new
DateExtractFunction(DateExtractFunctionType.DAY_OF_YEAR, func));
functionMap.put("dayofmonth", func -> new
DateExtractFunction(DateExtractFunctionType.DAY_OF_MONTH, func));
+ functionMap.put("hour",
+ func -> new
TimestampExtractFunction(TimestampExtractFunction.TimestampExtractFunctionType.HOUR,
func));
+ functionMap.put("minute",
+ func -> new
TimestampExtractFunction(TimestampExtractFunction.TimestampExtractFunctionType.MINUTE,
+ func));
+ functionMap.put("second",
+ func -> new
TimestampExtractFunction(TimestampExtractFunction.TimestampExtractFunctionType.SECOND,
+ func));
}
public static ExpressionOperator buildOperator(Expression expr) {
@@ -158,6 +170,8 @@ public class OperatorTools {
return new DivisionParser((Division) expr);
} else if (expr instanceof DateValue) {
return new DateParser((DateValue) expr);
+ } else if (expr instanceof TimestampValue) {
+ return new TimestampParser((TimestampValue) expr);
} else if (expr instanceof Function) {
String exprString = expr.toString();
if (exprString.startsWith(ROOT_KEY) ||
exprString.startsWith(CHILD_KEY)) {
@@ -202,6 +216,14 @@ public class OperatorTools {
}
}
+ public static Timestamp parseTimestamp(Object value) {
+ if (value instanceof Timestamp) {
+ return (Timestamp) value;
+ } else {
+ return Timestamp.valueOf(String.valueOf(value));
+ }
+ }
+
/**
* compareValue
* @param left
diff --git
a/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/parser/TimestampParser.java
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/parser/TimestampParser.java
new file mode 100644
index 0000000000..1971ba722a
--- /dev/null
+++
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/parser/TimestampParser.java
@@ -0,0 +1,43 @@
+/*
+ * 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.inlong.sdk.transform.process.parser;
+
+import org.apache.inlong.sdk.transform.decode.SourceData;
+import org.apache.inlong.sdk.transform.process.Context;
+
+import net.sf.jsqlparser.expression.TimestampValue;
+
+import java.sql.Timestamp;
+
+/**
+ * DateParser
+ * description: parse the sql expression to a java.sql.Timestamp object
+ */
+public class TimestampParser implements ValueParser {
+
+ private final Timestamp timestampValue;
+
+ public TimestampParser(TimestampValue expr) {
+ this.timestampValue =
Timestamp.valueOf(expr.getValue().toLocalDateTime());
+ }
+
+ @Override
+ public Object parse(SourceData sourceData, int rowIndex, Context context) {
+ return timestampValue;
+ }
+}
diff --git
a/inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/TestTransformTemporalFunctionsProcessor.java
b/inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/TestTransformTemporalFunctionsProcessor.java
index 0e847160f3..9acf27ca70 100644
---
a/inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/TestTransformTemporalFunctionsProcessor.java
+++
b/inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/TestTransformTemporalFunctionsProcessor.java
@@ -183,4 +183,37 @@ public class TestTransformTemporalFunctionsProcessor {
Assert.assertEquals(1, output6.size());
Assert.assertEquals(output6.get(0), "result=29");
}
+
+ @Test
+ public void testTimestampExtractFunction() throws Exception {
+ String transformSql1 = "select hour(string1) from source";
+ TransformConfig config1 = new TransformConfig(transformSql1);
+ TransformProcessor<String, String> processor1 = TransformProcessor
+ .create(config1,
SourceDecoderFactory.createCsvDecoder(csvSource),
+ SinkEncoderFactory.createKvEncoder(kvSink));
+ // case1: hour(2024-08-12 12:23:34)
+ List<String> output1 = processor1.transform("2024-08-12 12:23:34", new
HashMap<>());
+ Assert.assertEquals(1, output1.size());
+ Assert.assertEquals(output1.get(0), "result=12");
+
+ String transformSql2 = "select minute(string1) from source";
+ TransformConfig config2 = new TransformConfig(transformSql2);
+ TransformProcessor<String, String> processor2 = TransformProcessor
+ .create(config2,
SourceDecoderFactory.createCsvDecoder(csvSource),
+ SinkEncoderFactory.createKvEncoder(kvSink));
+ // case2: minute(2024-08-12 12:23:34)
+ List<String> output2 = processor2.transform("2024-08-12 12:23:34", new
HashMap<>());
+ Assert.assertEquals(1, output2.size());
+ Assert.assertEquals(output2.get(0), "result=23");
+
+ String transformSql3 = "select second(string1) from source";
+ TransformConfig config3 = new TransformConfig(transformSql3);
+ TransformProcessor<String, String> processor3 = TransformProcessor
+ .create(config3,
SourceDecoderFactory.createCsvDecoder(csvSource),
+ SinkEncoderFactory.createKvEncoder(kvSink));
+ // case3: second(2024-08-12 12:23:34)
+ List<String> output3 = processor3.transform("2024-08-12 12:23:34", new
HashMap<>());
+ Assert.assertEquals(1, output3.size());
+ Assert.assertEquals(output3.get(0), "result=34");
+ }
}
