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/10/17 02:38:24 UTC
[PR] [GLUTEN-9553][FLINK] Support kafka connector for data source [incubator-gluten]
KevinyhZou opened a new pull request, #9554:
URL: https://github.com/apache/incubator-gluten/pull/9554
## What changes were proposed in this pull request?
(Please fill in changes proposed in this fix)
(Fixes: \#9553 )
related to pr: https://github.com/bigo-sg/velox4j/pull/2, https://github.com/bigo-sg/velox/pull/2
## 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-9553][FLINK] Support kafka connector for data source [incubator-gluten]
Posted by "zhanglistar (via GitHub)" <gi...@apache.org>.
zhanglistar commented on code in PR #9554:
URL: https://github.com/apache/incubator-gluten/pull/9554#discussion_r2597311882
##########
gluten-flink/ut/src/test/resources/nexmark/ddl_kafka.sql:
##########
@@ -0,0 +1,46 @@
+CREATE TABLE kafka (
+ event_type int,
+ person ROW<
+ id BIGINT,
+ name VARCHAR,
+ emailAddress VARCHAR,
+ creditCard VARCHAR,
+ city VARCHAR,
+ state VARCHAR,
+ `dateTime` TIMESTAMP(3),
+ extra VARCHAR>,
+ auction ROW<
+ id BIGINT,
+ itemName VARCHAR,
+ description VARCHAR,
+ initialBid BIGINT,
+ reserve BIGINT,
+ `dateTime` TIMESTAMP(3),
+ expires TIMESTAMP(3),
+ seller BIGINT,
+ category BIGINT,
+ extra VARCHAR>,
+ bid ROW<
+ auction BIGINT,
+ bidder BIGINT,
+ price BIGINT,
+ channel VARCHAR,
+ url VARCHAR,
+ `dateTime` TIMESTAMP(3),
+ extra VARCHAR>,
+ `dateTime` AS
+ CASE
+ WHEN event_type = 0 THEN person.`dateTime`
+ WHEN event_type = 1 THEN auction.`dateTime`
+ ELSE bid.`dateTime`
+ END,
+ WATERMARK FOR `dateTime` AS `dateTime` - INTERVAL '4' SECOND
+) WITH (
+ 'connector' = 'kafka',
+ 'topic' = 'nexmark',
+ 'properties.bootstrap.servers' = '${BOOTSTRAP_SERVERS}',
+ 'properties.group.id' = 'nexmark',
+ 'scan.startup.mode' = 'earliest-offset',
+ 'sink.partitioner' = 'round-robin',
+ 'format' = 'json'
+);
Review Comment:
empty line
--
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-9553][FLINK] Support kafka connector for data source [incubator-gluten]
Posted by "zhanglistar (via GitHub)" <gi...@apache.org>.
zhanglistar commented on code in PR #9554:
URL: https://github.com/apache/incubator-gluten/pull/9554#discussion_r2580262885
##########
gluten-flink/planner/src/main/resources/META-INF/services/org.apache.gluten.velox.VeloxSourceSinkFactory:
##########
@@ -0,0 +1,3 @@
+org.apache.gluten.velox.FromElementsSourceFactory
+org.apache.gluten.velox.KafkaSourceSinkFactory
+org.apache.gluten.velox.PrintSinkFactory
Review Comment:
empty line
--
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-9553][FLINK] Support kafka connector for data source [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #9554:
URL: https://github.com/apache/incubator-gluten/pull/9554#issuecomment-3148922789
This PR is stale because it has been open 45 days with no activity. Remove stale label or comment or this will be closed in 10 days.
--
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-9553][FLINK] Support kafka connector for data source [incubator-gluten]
Posted by "KevinyhZou (via GitHub)" <gi...@apache.org>.
KevinyhZou closed pull request #9554: [GLUTEN-9553][FLINK] Support kafka connector for data source
URL: https://github.com/apache/incubator-gluten/pull/9554
--
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-9553][FLINK] Support kafka connector for data source [incubator-gluten]
Posted by "KevinyhZou (via GitHub)" <gi...@apache.org>.
KevinyhZou closed pull request #9554: [GLUTEN-9553][FLINK] Support kafka connector for data source
URL: https://github.com/apache/incubator-gluten/pull/9554
--
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-9553][FLINK] Support kafka connector for data source [incubator-gluten]
Posted by "KevinyhZou (via GitHub)" <gi...@apache.org>.
KevinyhZou commented on PR #9554:
URL: https://github.com/apache/incubator-gluten/pull/9554#issuecomment-3631936928
@PHILO-HE Could you help to review 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-9553][FLINK] Support kafka connector for data source [incubator-gluten]
Posted by "KevinyhZou (via GitHub)" <gi...@apache.org>.
KevinyhZou commented on PR #9554:
URL: https://github.com/apache/incubator-gluten/pull/9554#issuecomment-2962212502
> This pr goes in a wrong way, please refer to
>
> https://github.com/apache/incubator-gluten/blob/209dbd04b8b917e8c4ef75285e81e7d455e7bc18/gluten-flink/runtime/src/main/java/org/apache/flink/streaming/runtime/translators/SourceTransformationTranslator.java#L85
Now the kafka source is modified according to your advice, please review the code. @shuai-xu
--
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-9553][FLINK] Support kafka connector for data source [incubator-gluten]
Posted by "lgbo-ustc (via GitHub)" <gi...@apache.org>.
lgbo-ustc commented on PR #9554:
URL: https://github.com/apache/incubator-gluten/pull/9554#issuecomment-3539700374
@KevinyhZou Would you mind implementing a factory strategy in `VeloxSourceBuilder` and placing each source implementation in separate files？This would helping avoid having everything in a single file and make the code clearer
--
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-9553][GLUTEN-9555][FLINK] Support kafka connector for data source and `json`/`raw` data format [incubator-gluten]
Posted by "lgbo-ustc (via GitHub)" <gi...@apache.org>.
lgbo-ustc commented on code in PR #9554:
URL: https://github.com/apache/incubator-gluten/pull/9554#discussion_r2080810487
##########
gluten-flink/connectors/src/main/java/org/apache/gluten/connectors/kafka/GlutenKafkaSource.java:
##########
@@ -0,0 +1,158 @@
+/*
+ * 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.connectors.kafka;
+
+ import java.util.Properties;
+
+ import org.apache.flink.api.common.typeinfo.TypeInformation;
+ import org.apache.flink.api.connector.source.Boundedness;
+ import org.apache.flink.api.connector.source.Source;
+ import org.apache.flink.api.connector.source.SourceReader;
+ import org.apache.flink.api.connector.source.SourceReaderContext;
+ import org.apache.flink.api.connector.source.SplitEnumerator;
+ import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+ import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+ import org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumState;
+ import org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumStateSerializer;
+ import org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumerator;
+ import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
+ import org.apache.flink.connector.kafka.source.enumerator.subscriber.KafkaSubscriber;
+ import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
+ import org.apache.flink.connector.kafka.source.split.KafkaPartitionSplit;
+ import org.apache.flink.connector.kafka.source.split.KafkaPartitionSplitSerializer;
+ import org.apache.flink.core.io.SimpleVersionedSerializer;
+ import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
+ import org.apache.flink.table.types.DataType;
+ import org.slf4j.Logger;
+ import org.slf4j.LoggerFactory;
+
+ public class GlutenKafkaSource<OUT> implements Source<OUT, KafkaPartitionSplit, KafkaSourceEnumState>, ResultTypeQueryable<OUT> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(GlutenKafkaSource.class);
+
+ private final Properties properties;
+
+ private final String format;
+
+ private final DataType outputType;
+
+ private final KafkaRecordDeserializationSchema<OUT> deserializationSchema;
+
+ private final KafkaSubscriber subscriber;
+
+ private final OffsetsInitializer startingOffsetsInitializer;
+
+ private final OffsetsInitializer stoppingOffsetsInitializer;
+
+ private String planNodeId;
+
+ public GlutenKafkaSource(
+ String planNodeId,
+ String format,
+ Properties props,
+ DataType outpuType,
+ KafkaRecordDeserializationSchema<OUT> deserializationSchema,
+ KafkaSubscriber subscriber,
+ OffsetsInitializer startOffsetsInitializer,
+ OffsetsInitializer stopOffsetsInitializer) {
+ this.planNodeId = planNodeId;
+ this.format = format;
+ this.properties = props;
+ this.outputType = outpuType;
+ this.deserializationSchema = deserializationSchema;
+ this.subscriber = subscriber;
+ this.startingOffsetsInitializer = startOffsetsInitializer;
+ this.stoppingOffsetsInitializer = stopOffsetsInitializer;
+ }
+
+ @Override
+ public Boundedness getBoundedness() {
+ return Boundedness.CONTINUOUS_UNBOUNDED;
+ }
+
+ @Override
+ public SourceReader<OUT, KafkaPartitionSplit> createReader(SourceReaderContext readerContext) throws Exception {
+ return new GlutenKafkaSourceReader<OUT>(planNodeId, format, outputType, properties);
+ }
+
+ @Override
+ public TypeInformation<OUT> getProducedType() {
+ return deserializationSchema.getProducedType();
+ }
+
+ public void setPlanNodeId(String planNodeId) {
+ this.planNodeId = planNodeId;
+ }
+
+ public void setEnableAutoCommitOffset(boolean enabled) {
+ this.properties.setProperty("enable.auto.commit", String.valueOf(enabled));
+ }
+
+ public void setStartupMode(StartupMode mode) {
+ switch(mode) {
+ case GROUP_OFFSETS :
+ this.properties.setProperty("scan.startup.mode", "group-offsets");
+ break;
+ case LATEST:
+ this.properties.setProperty("scan.startup.mode", "latest-offsets");
+ break;
+ case EARLIEST:
+ this.properties.setProperty("scan.startup.mode", "earliest-offsets");
+ break;
+ default:
+ this.properties.setProperty("scan.startup.mode", "group-offsets");
+ break;
+ }
+ }
+
+ @Override
+ public SplitEnumerator<KafkaPartitionSplit, KafkaSourceEnumState> createEnumerator(
+ SplitEnumeratorContext<KafkaPartitionSplit> enumContext) throws Exception {
+ return new KafkaSourceEnumerator(
+ subscriber,
+ startingOffsetsInitializer,
+ stoppingOffsetsInitializer,
+ properties,
+ enumContext,
+ getBoundedness());
+ }
+
+ @Override
+ public SplitEnumerator<KafkaPartitionSplit, KafkaSourceEnumState> restoreEnumerator(
+ SplitEnumeratorContext<KafkaPartitionSplit> enumContext, KafkaSourceEnumState checkpoint) throws Exception {
+ return new KafkaSourceEnumerator(
+ subscriber,
+ startingOffsetsInitializer,
+ stoppingOffsetsInitializer,
+ properties,
+ enumContext,
+ getBoundedness(),
+ checkpoint);
+ }
+
+ @Override
+ public SimpleVersionedSerializer<KafkaPartitionSplit> getSplitSerializer() {
+ return new KafkaPartitionSplitSerializer();
+ }
+
+ @Override
+ public SimpleVersionedSerializer<KafkaSourceEnumState> getEnumeratorCheckpointSerializer() {
+ return new KafkaSourceEnumStateSerializer();
+ }
+
+ }
Review Comment:
add a new line at the end
##########
gluten-flink/connectors/src/main/java/org/apache/gluten/connectors/kafka/GlutenKafkaDynamicSource.java:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.connectors.kafka;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.regex.Pattern;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.connector.kafka.source.KafkaSource;
+import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
+import org.apache.flink.connector.kafka.source.enumerator.subscriber.KafkaSubscriber;
+import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.connectors.kafka.config.BoundedMode;
+import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+import org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSource;
+import org.apache.flink.table.connector.ProviderContext;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DataStreamScanProvider;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.formats.raw.RawFormatDeserializationSchema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class GlutenKafkaDynamicSource extends KafkaDynamicSource{
+
+ private static final Logger LOG = LoggerFactory.getLogger(GlutenKafkaDynamicSource.class);
+
+ public GlutenKafkaDynamicSource(DataType physicalDataType,
+ DecodingFormat<DeserializationSchema<RowData>> keyDecodingFormat,
+ DecodingFormat<DeserializationSchema<RowData>> valueDecodingFormat,
+ int[] keyProjection,
+ int[] valueProjection,
+ String keyPrefix,
+ List<String> topics,
+ Pattern topicPattern,
+ Properties properties,
+ StartupMode startupMode,
+ Map<KafkaTopicPartition, Long> specificStartupOffsets,
+ long startupTimestampMillis,
+ BoundedMode boundedMode,
+ Map<KafkaTopicPartition, Long> specificBoundedOffsets,
+ long boundedTimestampMillis,
+ boolean upsertMode,
+ String tableIdentifier) {
+ super(physicalDataType, keyDecodingFormat, valueDecodingFormat, keyProjection, valueProjection, keyPrefix, topics,
+ topicPattern, properties, startupMode, specificStartupOffsets, startupTimestampMillis, boundedMode,
+ specificBoundedOffsets, boundedTimestampMillis, upsertMode, tableIdentifier);
+ }
+
+ public static GlutenKafkaDynamicSource copyFrom(KafkaDynamicSource source) throws Exception {
+ final GlutenKafkaDynamicSource copy =
+ new GlutenKafkaDynamicSource(
+ getFieldFromKafkaSource(source, "physicalDataType", DataType.class),
+ getFieldFromKafkaSource(source, "keyDecodingFormat", DecodingFormat.class),
+ getFieldFromKafkaSource(source, "valueDecodingFormat", DecodingFormat.class),
+ getFieldFromKafkaSource(source, "keyProjection", int[].class),
+ getFieldFromKafkaSource(source, "valueProjection", int[].class),
+ getFieldFromKafkaSource(source, "keyPrefix", String.class),
+ getFieldFromKafkaSource(source, "topics", List.class),
+ getFieldFromKafkaSource(source, "topicPattern", Pattern.class),
+ getFieldFromKafkaSource(source, "properties", Properties.class),
+ getFieldFromKafkaSource(source, "startupMode", StartupMode.class),
+ getFieldFromKafkaSource(source, "specificStartupOffsets", Map.class),
+ getFieldFromKafkaSource(source, "startupTimestampMillis", long.class),
+ getFieldFromKafkaSource(source, "boundedMode", BoundedMode.class),
+ getFieldFromKafkaSource(source, "specificBoundedOffsets", Map.class),
+ getFieldFromKafkaSource(source, "boundedTimestampMillis", long.class),
+ getFieldFromKafkaSource(source, "upsertMode", boolean.class),
+ getFieldFromKafkaSource(source, "tableIdentifier", String.class));
+ copy.producedDataType = getFieldFromKafkaSource(source, "producedDataType", DataType.class);
+ copy.metadataKeys = getFieldFromKafkaSource(source, "metadataKeys", List.class);
+ copy.watermarkStrategy = getFieldFromKafkaSource(source, "watermarkStrategy", WatermarkStrategy.class);
+ return copy;
+ }
+
+ private static <T> T getFieldValue(Object source, Class<?> sourceClazz, String fieldName, Class<T> fieldClazz) {
+ try {
+ Field f = sourceClazz.getDeclaredField(fieldName);
+ f.setAccessible(true);
+ Class<?> fClazz = f.getType();
+ if (fClazz.equals(fieldClazz)) {
+ return (T) f.get(source);
+ } else {
+ String errMsg = String.format("Faield to get field %s from KafkaDynamimcSource, the field class {} not match with given class: {}",
+ fieldName, fClazz.getName(), fieldClazz.getName());
+ throw new RuntimeException(errMsg);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static <T> T getFieldFromKafkaSource(KafkaDynamicSource source, String fieldName, Class<T> fieldClazz) {
+ return getFieldValue(source, KafkaDynamicSource.class, fieldName, fieldClazz);
+ }
+
+ private static <T> T getFieldFromKafkaSource(KafkaSource<?> source, String fieldName, Class<T> fieldClazz) {
+ return getFieldValue(source, KafkaSource.class, fieldName, fieldClazz);
+ }
+
+ private static <T> T invokeMethodFromKafkaSource(KafkaDynamicSource source, String methodName, Class<?>[] argClazzs, Object[] args) {
+ try {
+ Method m = KafkaDynamicSource.class.getDeclaredMethod(methodName, argClazzs);
+ if (m != null) {
+ m.setAccessible(true);
+ return (T) m.invoke(source, args);
+ } else {
+ throw new RuntimeException("Failed to get method name of: " + methodName);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
Review Comment:
Is the catch here unnecessary?
--
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-9553][FLINK] Support kafka connector for data source [incubator-gluten]
Posted by "KevinyhZou (via GitHub)" <gi...@apache.org>.
KevinyhZou commented on PR #9554:
URL: https://github.com/apache/incubator-gluten/pull/9554#issuecomment-3539703047
> @KevinyhZou Would you mind implementing a factory strategy in `VeloxSourceBuilder` and placing each source implementation in separate files？This would helping avoid having everything in a single file and make the code clearer
yes, I will do that.
--
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-9553][FLINK] Support kafka connector for data source [incubator-gluten]
Posted by "lgbo-ustc (via GitHub)" <gi...@apache.org>.
lgbo-ustc commented on code in PR #9554:
URL: https://github.com/apache/incubator-gluten/pull/9554#discussion_r2600715388
##########
gluten-flink/planner/src/main/java/org/apache/gluten/velox/FromElementsSourceFactory.java:
##########
@@ -40,22 +42,27 @@
import java.util.List;
import java.util.Map;
-public class VeloxSourceBuilder {
+public class FromElementsSourceFactory implements VeloxSourceSinkFactory {
- public static Transformation<RowData> build(
- Transformation<RowData> transformation, ScanTableSource scanTableSource) {
+ @SuppressWarnings("rawtypes")
+ @Override
+ public boolean match(Transformation<RowData> transformation) {
if (transformation instanceof LegacySourceTransformation) {
- if (scanTableSource.getClass().getSimpleName().equals("TestValuesScanLookupTableSource")) {
- return buildFromElementsSource(transformation, scanTableSource);
+ StreamSource source = ((LegacySourceTransformation) transformation).getOperator();
+ String sourceFunctionName = source.getUserFunction().getClass().getSimpleName();
+ if (sourceFunctionName.equals("FromElementsFunction")) {
Review Comment:
Is `isinstanceof` better?
--
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-9553]Support kafka connector for data source [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #9554:
URL: https://github.com/apache/incubator-gluten/pull/9554#issuecomment-2861711760
https://github.com/apache/incubator-gluten/issues/9553
--
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-9553][GLUTEN-9555][FLINK] Support kafka connector for data source and `json`/`raw` data format [incubator-gluten]
Posted by "KevinyhZou (via GitHub)" <gi...@apache.org>.
KevinyhZou commented on PR #9554:
URL: https://github.com/apache/incubator-gluten/pull/9554#issuecomment-2864929235
Please give the pr reference which implements the `NexmarkSource`， @shuai-xu
--
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-9553][GLUTEN-9555][FLINK] Support kafka connector for data source and `json`/`raw` data format [incubator-gluten]
Posted by "shuai-xu (via GitHub)" <gi...@apache.org>.
shuai-xu commented on PR #9554:
URL: https://github.com/apache/incubator-gluten/pull/9554#issuecomment-2864923425
We don't implement source like this, please refer to NexmarkSource
--
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