You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by "KevinyhZou (via GitHub)" <gi...@apache.org> on 2025/06/16 02:16:01 UTC
[PR] [GLUTEN-9950][Flink]Make nexmark source num events configuratable [incubator-gluten]
KevinyhZou opened a new pull request, #9978:
URL: https://github.com/apache/incubator-gluten/pull/9978
## What changes were proposed in this pull request?
(Please fill in changes proposed in this fix)
(Fixes: \#9950)
## How was this patch tested?
(Please explain how this patch was tested. E.g. unit tests, integration tests, manual tests)
(If this patch involves UI changes, please attach a screenshot; otherwise, remove this)
--
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-9950][Flink]Make nexmark source num events configuratable [incubator-gluten]
Posted by "KevinyhZou (via GitHub)" <gi...@apache.org>.
KevinyhZou commented on code in PR #9978:
URL: https://github.com/apache/incubator-gluten/pull/9978#discussion_r2149242227
##########
gluten-flink/runtime/src/main/java/org/apache/flink/streaming/runtime/translators/SourceTransformationTranslator.java:
##########
@@ -81,12 +82,29 @@ private Collection<Integer> translateInternal(
final ExecutionConfig executionConfig = streamGraph.getExecutionConfig();
// --- Begin Gluten-specific code changes ---
- if (transformation.getSource().getClass().getSimpleName().equals("NexmarkSource")) {
+ Class<?> sourceClazz = transformation.getSource().getClass();
+ if (sourceClazz.getSimpleName().equals("NexmarkSource")) {
RowType outputType =
(RowType)
LogicalTypeConverter.toVLType(
((InternalTypeInfo) transformation.getOutputType()).toLogicalType());
String id = PlanNodeIdGenerator.newId();
+ Object nexmarkSource = transformation.getSource();
+ List<?> nexmarkSourceSplits =
+ (List<?>)
+ ReflectUtils.invokeObjectMethod(
+ sourceClazz,
+ nexmarkSource,
+ "getSplits",
+ new Class<?>[] {Integer.class},
+ new Object[] {transformation.getParallelism()});
+ Object nexmarkSourceSplit = nexmarkSourceSplits.get(0);
+ Object generatorConfig =
+ ReflectUtils.getObjectField(
+ nexmarkSourceSplit.getClass(), nexmarkSourceSplit, "generatorConfig");
+ Integer maxEvents =
Review Comment:
fixed.
--
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-9950][Flink]Make nexmark source num events configuratable [incubator-gluten]
Posted by "KevinyhZou (via GitHub)" <gi...@apache.org>.
KevinyhZou commented on code in PR #9978:
URL: https://github.com/apache/incubator-gluten/pull/9978#discussion_r2149410884
##########
gluten-flink/runtime/src/main/java/org/apache/flink/streaming/runtime/translators/SourceTransformationTranslator.java:
##########
@@ -81,12 +82,29 @@ private Collection<Integer> translateInternal(
final ExecutionConfig executionConfig = streamGraph.getExecutionConfig();
// --- Begin Gluten-specific code changes ---
- if (transformation.getSource().getClass().getSimpleName().equals("NexmarkSource")) {
+ Class<?> sourceClazz = transformation.getSource().getClass();
+ if (sourceClazz.getSimpleName().equals("NexmarkSource")) {
RowType outputType =
(RowType)
LogicalTypeConverter.toVLType(
((InternalTypeInfo) transformation.getOutputType()).toLogicalType());
String id = PlanNodeIdGenerator.newId();
+ Object nexmarkSource = transformation.getSource();
+ List<?> nexmarkSourceSplits =
+ (List<?>)
+ ReflectUtils.invokeObjectMethod(
+ sourceClazz,
+ nexmarkSource,
+ "getSplits",
+ new Class<?>[] {Integer.class},
+ new Object[] {transformation.getParallelism()});
+ Object nexmarkSourceSplit = nexmarkSourceSplits.get(0);
+ Object generatorConfig =
+ ReflectUtils.getObjectField(
+ nexmarkSourceSplit.getClass(), nexmarkSourceSplit, "generatorConfig");
+ Integer maxEvents =
Review Comment:
fixed
--
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-9950][Flink]Make nexmark source num events configuratable [incubator-gluten]
Posted by "yuanhang-dev (via GitHub)" <gi...@apache.org>.
yuanhang-dev commented on code in PR #9978:
URL: https://github.com/apache/incubator-gluten/pull/9978#discussion_r2148974803
##########
gluten-flink/runtime/src/main/java/org/apache/flink/streaming/runtime/translators/SourceTransformationTranslator.java:
##########
@@ -81,12 +82,29 @@ private Collection<Integer> translateInternal(
final ExecutionConfig executionConfig = streamGraph.getExecutionConfig();
// --- Begin Gluten-specific code changes ---
- if (transformation.getSource().getClass().getSimpleName().equals("NexmarkSource")) {
+ Class<?> sourceClazz = transformation.getSource().getClass();
+ if (sourceClazz.getSimpleName().equals("NexmarkSource")) {
RowType outputType =
(RowType)
LogicalTypeConverter.toVLType(
((InternalTypeInfo) transformation.getOutputType()).toLogicalType());
String id = PlanNodeIdGenerator.newId();
+ Object nexmarkSource = transformation.getSource();
+ List<?> nexmarkSourceSplits =
+ (List<?>)
+ ReflectUtils.invokeObjectMethod(
+ sourceClazz,
+ nexmarkSource,
+ "getSplits",
+ new Class<?>[] {Integer.class},
+ new Object[] {transformation.getParallelism()});
+ Object nexmarkSourceSplit = nexmarkSourceSplits.get(0);
+ Object generatorConfig =
+ ReflectUtils.getObjectField(
+ nexmarkSourceSplit.getClass(), nexmarkSourceSplit, "generatorConfig");
+ Integer maxEvents =
Review Comment:
When running nexmark q0, an exception will be thrown here. This feature can run nexmark q0 to ensure correctness.
```
Caused by: java.lang.ClassCastException: class java.lang.Long cannot be cast to class java.lang.Integer (java.lang.Long and java.lang.Integer are in module java.base of loader 'bootstrap')
at org.apache.flink.streaming.runtime.translators.SourceTransformationTranslator.translateInternal(SourceTransformationTranslator.java:98) ~[gluten-flink-runtime-1.5.0-SNAPSHOT.jar:1.19.2]
at org.apache.flink.streaming.runtime.translators.SourceTransformationTranslator.translateForStreamingInternal(SourceTransformationTranslator.java:69) ~[gluten-flink-runtime-1.5.0-SNAPSHOT.jar:1.19.2]
at org.apache.flink.streaming.runtime.translators.SourceTransformationTranslator.translateForStreamingInternal(SourceTransformationTranslator.java:53) ~[gluten-flink-runtime-1.5.0-SNAPSHOT.jar:1.19.2]
at org.apache.flink.streaming.api.graph.SimpleTransformationTranslator.translateForStreaming(SimpleTransformationTranslator.java:62) ~[flink-dist-1.19.2.jar:1.19.2]
--
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-9950][Flink]Make nexmark source num events configurable [incubator-gluten]
Posted by "shuai-xu (via GitHub)" <gi...@apache.org>.
shuai-xu commented on code in PR #9978:
URL: https://github.com/apache/incubator-gluten/pull/9978#discussion_r2148939930
##########
gluten-flink/runtime/src/main/java/org/apache/flink/streaming/runtime/translators/SourceTransformationTranslator.java:
##########
@@ -81,12 +82,20 @@ private Collection<Integer> translateInternal(
final ExecutionConfig executionConfig = streamGraph.getExecutionConfig();
// --- Begin Gluten-specific code changes ---
- if (transformation.getSource().getClass().getSimpleName().equals("NexmarkSource")) {
+ Class<?> sourceClazz = transformation.getSource().getClass();
+ if (sourceClazz.getSimpleName().equals("NexmarkSource")) {
RowType outputType =
(RowType)
LogicalTypeConverter.toVLType(
((InternalTypeInfo) transformation.getOutputType()).toLogicalType());
String id = PlanNodeIdGenerator.newId();
+ Object nexmarkSource = transformation.getSource();
+ Object generatorConfig = ReflectUtils.getObjectField(sourceClazz, nexmarkSource, "config");
Review Comment:
should call method `getSplits` instead of using config.
--
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-9950][Flink]Make nexmark source num events configuratable [incubator-gluten]
Posted by "yuanhang-dev (via GitHub)" <gi...@apache.org>.
yuanhang-dev commented on code in PR #9978:
URL: https://github.com/apache/incubator-gluten/pull/9978#discussion_r2149280557
##########
gluten-flink/runtime/src/main/java/org/apache/flink/streaming/runtime/translators/SourceTransformationTranslator.java:
##########
@@ -81,12 +82,29 @@ private Collection<Integer> translateInternal(
final ExecutionConfig executionConfig = streamGraph.getExecutionConfig();
// --- Begin Gluten-specific code changes ---
- if (transformation.getSource().getClass().getSimpleName().equals("NexmarkSource")) {
+ Class<?> sourceClazz = transformation.getSource().getClass();
+ if (sourceClazz.getSimpleName().equals("NexmarkSource")) {
RowType outputType =
(RowType)
LogicalTypeConverter.toVLType(
((InternalTypeInfo) transformation.getOutputType()).toLogicalType());
String id = PlanNodeIdGenerator.newId();
+ Object nexmarkSource = transformation.getSource();
+ List<?> nexmarkSourceSplits =
+ (List<?>)
+ ReflectUtils.invokeObjectMethod(
+ sourceClazz,
+ nexmarkSource,
+ "getSplits",
+ new Class<?>[] {Integer.class},
+ new Object[] {transformation.getParallelism()});
+ Object nexmarkSourceSplit = nexmarkSourceSplits.get(0);
+ Object generatorConfig =
+ ReflectUtils.getObjectField(
+ nexmarkSourceSplit.getClass(), nexmarkSourceSplit, "generatorConfig");
+ Integer maxEvents =
Review Comment:
nexmark q0 is still unable to run
```
Caused by: java.lang.NoSuchMethodException: com.github.nexmark.flink.source.NexmarkSource.getSplits(java.lang.Integer)
at java.lang.Class.getDeclaredMethod(Class.java:2475) ~[?:?]
at org.apache.gluten.util.ReflectUtils.invokeObjectMethod(ReflectUtils.java:37) ~[gluten-flink-runtime-1.5.0-SNAPSHOT.jar:1.5.0-SNAPSHOT]
at org.apache.flink.streaming.runtime.translators.SourceTransformationTranslator.translateInternal(SourceTransformationTranslator.java:95) ~[gluten-flink-runtime-1.5.0-SNAPSHOT.jar:1.19.2]
at org.apache.flink.streaming.runtime.translators.SourceTransformationTranslator.translateForStreamingInternal(SourceTransformationTranslator.java:69) ~[gluten-flink-runtime-1.5.0-SNAPSHOT.jar:1.19.2]
at org.apache.flink.streaming.runtime.translators.SourceTransformationTranslator.translateForStreamingInternal(SourceTransformationTranslator.java:53) ~[gluten-flink-runtime-1.5.0-SNAPSHOT.jar:1.19.2]
--
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-9950][FLINK] Make nexmark source num events configurable [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #9978:
URL: https://github.com/apache/incubator-gluten/pull/9978#discussion_r2151686859
##########
gluten-flink/runtime/src/main/java/org/apache/flink/streaming/runtime/translators/SourceTransformationTranslator.java:
##########
@@ -81,12 +82,29 @@ private Collection<Integer> translateInternal(
final ExecutionConfig executionConfig = streamGraph.getExecutionConfig();
// --- Begin Gluten-specific code changes ---
- if (transformation.getSource().getClass().getSimpleName().equals("NexmarkSource")) {
+ Class<?> sourceClazz = transformation.getSource().getClass();
+ if (sourceClazz.getSimpleName().equals("NexmarkSource")) {
Review Comment:
Just curious why reflection is needed here? Is it because the released velox4j jar doesn't include these source files?
--
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-9950][FLINK] Make nexmark source num events configurable [incubator-gluten]
Posted by "KevinyhZou (via GitHub)" <gi...@apache.org>.
KevinyhZou commented on code in PR #9978:
URL: https://github.com/apache/incubator-gluten/pull/9978#discussion_r2151744067
##########
gluten-flink/runtime/src/main/java/org/apache/flink/streaming/runtime/translators/SourceTransformationTranslator.java:
##########
@@ -81,12 +82,29 @@ private Collection<Integer> translateInternal(
final ExecutionConfig executionConfig = streamGraph.getExecutionConfig();
// --- Begin Gluten-specific code changes ---
- if (transformation.getSource().getClass().getSimpleName().equals("NexmarkSource")) {
+ Class<?> sourceClazz = transformation.getSource().getClass();
+ if (sourceClazz.getSimpleName().equals("NexmarkSource")) {
Review Comment:
The source jars is not introduced as dependencies by maven, because `nexmark-connector` source jar can notbe found at maven central repo; so here we use java reflection. And the `runtime` module of gluten-flink is not good to introduce some `connector` jars, I think.
--
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-9950][FLINK] Make nexmark source num events configurable [incubator-gluten]
Posted by "shuai-xu (via GitHub)" <gi...@apache.org>.
shuai-xu merged PR #9978:
URL: https://github.com/apache/incubator-gluten/pull/9978
--
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-9950][FLINK] Make nexmark source num events configurable [incubator-gluten]
Posted by "KevinyhZou (via GitHub)" <gi...@apache.org>.
KevinyhZou commented on code in PR #9978:
URL: https://github.com/apache/incubator-gluten/pull/9978#discussion_r2151744067
##########
gluten-flink/runtime/src/main/java/org/apache/flink/streaming/runtime/translators/SourceTransformationTranslator.java:
##########
@@ -81,12 +82,29 @@ private Collection<Integer> translateInternal(
final ExecutionConfig executionConfig = streamGraph.getExecutionConfig();
// --- Begin Gluten-specific code changes ---
- if (transformation.getSource().getClass().getSimpleName().equals("NexmarkSource")) {
+ Class<?> sourceClazz = transformation.getSource().getClass();
+ if (sourceClazz.getSimpleName().equals("NexmarkSource")) {
Review Comment:
The source jars is not introduced as dependencies by maven, because `nexmark-connector` source jar can notbe found at maven central repo; so here we use java reflection.
--
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-9950][Flink]Make nexmark source num events configuratable [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #9978:
URL: https://github.com/apache/incubator-gluten/pull/9978#issuecomment-2974912530
https://github.com/apache/incubator-gluten/issues/9950
--
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