You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by "shuai-xu (via GitHub)" <gi...@apache.org> on 2025/05/07 08:04:16 UTC
[PR] [GLUTEN-9518] support watermark assigner [incubator-gluten]
shuai-xu opened a new pull request, #9541:
URL: https://github.com/apache/incubator-gluten/pull/9541
## What changes were proposed in this pull request?
This pr support WatermarkAssginer.
(Fixes: \#9518)
## How was this patch tested?
This patch was tested by manual tests.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9518][FLINK] support watermark assigner [incubator-gluten]
Posted by "PHILO-HE (via GitHub)" <gi...@apache.org>.
PHILO-HE commented on code in PR #9541:
URL: https://github.com/apache/incubator-gluten/pull/9541#discussion_r2084007523
##########
gluten-flink/runtime/src/main/java/org/apache/gluten/util/LogicalTypeConverter.java:
##########
@@ -58,6 +59,9 @@ public static Type toVLType(LogicalType logicalType) {
return new io.github.zhztheplayer.velox4j.type.DecimalType(
decimalType.getPrecision(),
decimalType.getScale());
+ } else if (logicalType instanceof DayTimeIntervalType) {
+ // TODO: it seems interval now can be used as bigint for nexmark.
+ return new io.github.zhztheplayer.velox4j.type.BigIntType();
Review Comment:
Can we use `IntervalDayTimeType()`? IntervalDayTimeType is implemented based on BigIntType in Velox, but they are different logical types.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9518][FLINK] Support watermark assigner [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer merged PR #9541:
URL: https://github.com/apache/incubator-gluten/pull/9541
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9518] support watermark assigner [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #9541:
URL: https://github.com/apache/incubator-gluten/pull/9541#issuecomment-2857510450
https://github.com/apache/incubator-gluten/issues/9518
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9518][FLINK] support watermark assigner [incubator-gluten]
Posted by "shuai-xu (via GitHub)" <gi...@apache.org>.
shuai-xu commented on code in PR #9541:
URL: https://github.com/apache/incubator-gluten/pull/9541#discussion_r2081377241
##########
gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/functions/SubtractFunctionConverter.java:
##########
@@ -0,0 +1,54 @@
+/*
+ * 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.functions;
+
+import io.github.zhztheplayer.velox4j.expression.CallTypedExpr;
+import io.github.zhztheplayer.velox4j.expression.TypedExpr;
+import io.github.zhztheplayer.velox4j.type.BigIntType;
+import io.github.zhztheplayer.velox4j.type.Type;
+
+import org.apache.flink.util.Preconditions;
+import java.util.List;
+
+/** Subtract function converter. */
+public class SubtractFunctionConverter implements FunctionConverter {
+ private final String function;
+
+ public SubtractFunctionConverter(String function) {
+ this.function = function;
+ }
+
+ @Override
+ public CallTypedExpr toVeloxFunction(Type nodeType, List<TypedExpr> params) {
+ Preconditions.checkNotNull(params.size() == 2, "Subtract must contain exactly two parameters");
+
+ // TODO: need refine for more type cast
+ if (params.get(0).getReturnType() != params.get(1).getReturnType()) {
Review Comment:
Oh, We need to add compare method to Type.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9518][FLINK] support watermark assigner [incubator-gluten]
Posted by "shuai-xu (via GitHub)" <gi...@apache.org>.
shuai-xu commented on code in PR #9541:
URL: https://github.com/apache/incubator-gluten/pull/9541#discussion_r2081377241
##########
gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/functions/SubtractFunctionConverter.java:
##########
@@ -0,0 +1,54 @@
+/*
+ * 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.functions;
+
+import io.github.zhztheplayer.velox4j.expression.CallTypedExpr;
+import io.github.zhztheplayer.velox4j.expression.TypedExpr;
+import io.github.zhztheplayer.velox4j.type.BigIntType;
+import io.github.zhztheplayer.velox4j.type.Type;
+
+import org.apache.flink.util.Preconditions;
+import java.util.List;
+
+/** Subtract function converter. */
+public class SubtractFunctionConverter implements FunctionConverter {
+ private final String function;
+
+ public SubtractFunctionConverter(String function) {
+ this.function = function;
+ }
+
+ @Override
+ public CallTypedExpr toVeloxFunction(Type nodeType, List<TypedExpr> params) {
+ Preconditions.checkNotNull(params.size() == 2, "Subtract must contain exactly two parameters");
+
+ // TODO: need refine for more type cast
+ if (params.get(0).getReturnType() != params.get(1).getReturnType()) {
Review Comment:
Oh, We need to add compare function to Type.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9518][FLINK] support watermark assigner [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #9541:
URL: https://github.com/apache/incubator-gluten/pull/9541#discussion_r2081414727
##########
gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/functions/SubtractFunctionConverter.java:
##########
@@ -0,0 +1,54 @@
+/*
+ * 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.functions;
+
+import io.github.zhztheplayer.velox4j.expression.CallTypedExpr;
+import io.github.zhztheplayer.velox4j.expression.TypedExpr;
+import io.github.zhztheplayer.velox4j.type.BigIntType;
+import io.github.zhztheplayer.velox4j.type.Type;
+
+import org.apache.flink.util.Preconditions;
+import java.util.List;
+
+/** Subtract function converter. */
+public class SubtractFunctionConverter implements FunctionConverter {
+ private final String function;
+
+ public SubtractFunctionConverter(String function) {
+ this.function = function;
+ }
+
+ @Override
+ public CallTypedExpr toVeloxFunction(Type nodeType, List<TypedExpr> params) {
+ Preconditions.checkNotNull(params.size() == 2, "Subtract must contain exactly two parameters");
+
+ // TODO: need refine for more type cast
+ if (params.get(0).getReturnType() != params.get(1).getReturnType()) {
Review Comment:
Yes we can open a issue for that. So do we still need to perform this check in the PR?
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9518][FLINK] support watermark assigner [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #9541:
URL: https://github.com/apache/incubator-gluten/pull/9541#discussion_r2081284548
##########
gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/functions/SubtractFunctionConverter.java:
##########
@@ -0,0 +1,54 @@
+/*
+ * 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.functions;
+
+import io.github.zhztheplayer.velox4j.expression.CallTypedExpr;
+import io.github.zhztheplayer.velox4j.expression.TypedExpr;
+import io.github.zhztheplayer.velox4j.type.BigIntType;
+import io.github.zhztheplayer.velox4j.type.Type;
+
+import org.apache.flink.util.Preconditions;
+import java.util.List;
+
+/** Subtract function converter. */
+public class SubtractFunctionConverter implements FunctionConverter {
+ private final String function;
+
+ public SubtractFunctionConverter(String function) {
+ this.function = function;
+ }
+
+ @Override
+ public CallTypedExpr toVeloxFunction(Type nodeType, List<TypedExpr> params) {
+ Preconditions.checkNotNull(params.size() == 2, "Subtract must contain exactly two parameters");
+
+ // TODO: need refine for more type cast
+ if (params.get(0).getReturnType() != params.get(1).getReturnType()) {
Review Comment:
Is this code reliable as `!=` only compares references?
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9518][FLINK] support watermark assigner [incubator-gluten]
Posted by "PHILO-HE (via GitHub)" <gi...@apache.org>.
PHILO-HE commented on code in PR #9541:
URL: https://github.com/apache/incubator-gluten/pull/9541#discussion_r2084007523
##########
gluten-flink/runtime/src/main/java/org/apache/gluten/util/LogicalTypeConverter.java:
##########
@@ -58,6 +59,9 @@ public static Type toVLType(LogicalType logicalType) {
return new io.github.zhztheplayer.velox4j.type.DecimalType(
decimalType.getPrecision(),
decimalType.getScale());
+ } else if (logicalType instanceof DayTimeIntervalType) {
+ // TODO: it seems interval now can be used as bigint for nexmark.
+ return new io.github.zhztheplayer.velox4j.type.BigIntType();
Review Comment:
Can we use `IntervalDayTimeType()`? IntervalDayTimeType is implemented based on BigIntType in Velox, but they are logically different.
##########
gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/FunctionMappings.java:
##########
@@ -16,25 +16,32 @@
*/
package org.apache.gluten.rexnode;
+import org.apache.gluten.rexnode.functions.DefaultFunctionConverter;
+import org.apache.gluten.rexnode.functions.FunctionConverter;
+import org.apache.gluten.rexnode.functions.SubtractFunctionConverter;
+
import java.util.HashMap;
import java.util.Map;
/** Mapping of flink function and velox function. */
public class FunctionMappings {
// A map stores the relationship between flink function name and velox function.
- private static Map<String, String> functionMappings = new HashMap() {
+ private static Map<String, FunctionConverter> functionMappings = new HashMap() {
{
// TODO: support more functions.
- put(">", "greaterthan");
- put("<", "lessthan");
- put("=", "equalto");
- put("CAST", "cast");
- put("CASE", "if");
- put("*", "multiply");
+ put(">", new DefaultFunctionConverter("greaterthan"));
+ put("<", new DefaultFunctionConverter("lessthan"));
+ put("=", new DefaultFunctionConverter("equalto"));
+ put("CAST", new DefaultFunctionConverter("cast"));
+ put("CASE", new DefaultFunctionConverter("if"));
+ put("*", new DefaultFunctionConverter("multiply"));
+ put("-", new SubtractFunctionConverter("subtract"));
+ put("MOD", new DefaultFunctionConverter("pmod"));
Review Comment:
Should be mapped to "remainder". "pmod" has different meaning, which returns positive remainder.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9518][FLINK] support watermark assigner [incubator-gluten]
Posted by "shuai-xu (via GitHub)" <gi...@apache.org>.
shuai-xu commented on code in PR #9541:
URL: https://github.com/apache/incubator-gluten/pull/9541#discussion_r2083702544
##########
gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/functions/SubtractFunctionConverter.java:
##########
@@ -0,0 +1,54 @@
+/*
+ * 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.functions;
+
+import io.github.zhztheplayer.velox4j.expression.CallTypedExpr;
+import io.github.zhztheplayer.velox4j.expression.TypedExpr;
+import io.github.zhztheplayer.velox4j.type.BigIntType;
+import io.github.zhztheplayer.velox4j.type.Type;
+
+import org.apache.flink.util.Preconditions;
+import java.util.List;
+
+/** Subtract function converter. */
+public class SubtractFunctionConverter implements FunctionConverter {
+ private final String function;
+
+ public SubtractFunctionConverter(String function) {
+ this.function = function;
+ }
+
+ @Override
+ public CallTypedExpr toVeloxFunction(Type nodeType, List<TypedExpr> params) {
+ Preconditions.checkNotNull(params.size() == 2, "Subtract must contain exactly two parameters");
+
+ // TODO: need refine for more type cast
+ if (params.get(0).getReturnType() != params.get(1).getReturnType()) {
Review Comment:
Use class compare first
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9518][FLINK] support watermark assigner [incubator-gluten]
Posted by "shuai-xu (via GitHub)" <gi...@apache.org>.
shuai-xu commented on code in PR #9541:
URL: https://github.com/apache/incubator-gluten/pull/9541#discussion_r2084061345
##########
gluten-flink/runtime/src/main/java/org/apache/gluten/util/LogicalTypeConverter.java:
##########
@@ -58,6 +59,9 @@ public static Type toVLType(LogicalType logicalType) {
return new io.github.zhztheplayer.velox4j.type.DecimalType(
decimalType.getPrecision(),
decimalType.getScale());
+ } else if (logicalType instanceof DayTimeIntervalType) {
+ // TODO: it seems interval now can be used as bigint for nexmark.
+ return new io.github.zhztheplayer.velox4j.type.BigIntType();
Review Comment:
IntervalDayTimeType can not be cast to BigIntType now in velox. Change it to BigIntType temporary, I will assgin this issuse to others to fix it.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org