[Datastream] Introduce FlussSource #710

polyzos · 2025-04-02T06:45:25Z

This PR introduces the FlussSource for the Datastream API

polyzos · 2025-04-02T09:10:38Z

@wuchong RowConverters are used for testing now and can be improved and exposed as helper functions to users in another PR.
But let's discuss and keep track of the required improvements and overall thoughts.

wuchong

Thanks for the contribution @polyzos . I left some comments. Please also rebase your branch to the latest main branch, but NOT squash commits.

...link/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/FlussSourceBuilder.java

wuchong · 2025-04-27T13:18:50Z

...link/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/FlussSourceBuilder.java

+        return this;
+    }
+
+    public FlussSourceBuilder<IN> setProjectedFields(int[] projectedFields) {


The indexes are not friendly to use for end-users. Let's support projection by field names setProjectedFields(String... fieldNames).

@wuchong currently changing to fieldNames might break compatibility with the FlinkTableSource as it uses the same FlinkSource constructor, but it uses the indexes.. maybe leave it as is and introduce a separate ticket/PR to address this properly so its easier to track later?

@wuchong i created this for tracking.

I mean we only expose fieldNames projection in the FlussSourceBuilder, but still use field index projection in the underlying FlinkSource. We can do the mapping from fieldNames to field index in the build() method, as we have got the schema in it.

We can implement it in a follow-up PR.

...link/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/FlussSourceBuilder.java

...nk/fluss-flink-common/src/test/java/com/alibaba/fluss/flink/source/FlussSourceLogITCase.java

...ink/fluss-flink-common/src/test/java/com/alibaba/fluss/flink/source/FlussSourcePKITCase.java

...fluss-flink-common/src/test/java/com/alibaba/fluss/flink/source/testutils/MockDataUtils.java

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/row/RowConverters.java

polyzos · 2025-04-28T11:43:03Z

@wuchong thank you for your time and detailed feedback. PTAL as all comments should have been addressed.. let me know if I missed anything

wuchong

I only left some minor comments and pushed a commit to fix it.

Will merge it once CI is passed.

wuchong · 2025-04-29T08:02:30Z

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/FlinkSource.java

                offsetsInitializer,
                scanPartitionDiscoveryIntervalMs,
-                streaming);
+                true);


Why hard-cord this to true?

wuchong · 2025-04-29T08:03:35Z

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/FlussSource.java

+    }
+
+    @VisibleForTesting
+    public OffsetsInitializer getOffsetsInitializer() {


can be package visible

wuchong · 2025-04-29T08:03:54Z

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/FlussSource.java

+    }
+
+    @VisibleForTesting
+    public long getScanPartitionDiscoveryIntervalMs() {


never used, can remove

wuchong · 2025-04-29T08:06:08Z

...link/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/FlussSourceBuilder.java

+        return this;
+    }
+
+    public FlussSourceBuilder<OUT> setIsStreaming(boolean isStreaming) {


let's remove it as we don't support it yet.

wuchong · 2025-04-29T12:55:49Z

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/FlussSource.java

+public class FlussSource<OUT> extends FlinkSource<OUT> {
+    private static final long serialVersionUID = 1L;
+
+    public FlussSource(


Change to package visibility to avoid users direct use this constructor. As this constructor is evolving frequently.

wuchong · 2025-04-29T12:58:22Z

...link/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/FlussSourceBuilder.java

+        return this;
+    }
+
+    public FlussSourceBuilder<IN> setProjectedFields(int[] projectedFields) {


I mean we only expose fieldNames projection in the FlussSourceBuilder, but still use field index projection in the underlying FlinkSource. We can do the mapping from fieldNames to field index in the build() method, as we have got the schema in it.

We can implement it in a follow-up PR.

polyzos mentioned this pull request Mar 30, 2025

[Umbrella][Feature] Datastream API Support #569

Closed

12 tasks

polyzos marked this pull request as ready for review April 2, 2025 08:46

polyzos requested a review from wuchong April 2, 2025 09:10

wuchong reviewed Apr 27, 2025

View reviewed changes

polyzos force-pushed the datastream-fluss-source branch from 09ff0e0 to a947de9 Compare April 27, 2025 15:18

polyzos added 5 commits April 28, 2025 08:57

Introduce FlussSource and Builder

224f894

Introduce RowConverters

3d51e47

add FlussSourceBuilderTest test

0ef0fa4

add FlussSource IT tests

7fee648

improve param handling for builder

d62abee

polyzos force-pushed the datastream-fluss-source branch from a947de9 to d62abee Compare April 28, 2025 05:58

polyzos added 8 commits April 28, 2025 09:27

resolve comments v1

cf8d187

fix streaming mode in enumerator

1507b29

fix streaming mode in enumerator

3d8f56a

address comments v2 - merge it tests and add default values for configs

55e0aad

fix broken FlinkSource

d7b30f0

address comments v3

5610d83

fix broken builder tests

0ca8164

remove RowDataConverters class

2583f4c

polyzos requested a review from wuchong April 28, 2025 11:43

polyzos mentioned this pull request Apr 28, 2025

Make projection work with field names and not indexes #804

Closed

2 tasks

wuchong added 2 commits April 29, 2025 20:53

Jark's suggestion

0c8e6fe

move Order and OrderPartial to testutils package

95c7b9e

wuchong approved these changes Apr 29, 2025

View reviewed changes

fix

c6c50b8

wuchong merged commit 2412114 into apache:main Apr 29, 2025
3 checks passed

ZmmBigdata pushed a commit to ZmmBigdata/fluss that referenced this pull request Jun 20, 2025

[flink] Introduce FlussSource DataStream API (apache#710)

dde24e3

polyzos added a commit to polyzos/fluss that referenced this pull request Aug 30, 2025

[flink] Introduce FlussSource DataStream API (apache#710)

5133bdb

polyzos added a commit to Alibaba-HZY/fluss that referenced this pull request Aug 31, 2025

[flink] Introduce FlussSource DataStream API (apache#710)

3c8842b

[Datastream] Introduce FlussSource #710

[Datastream] Introduce FlussSource #710

Uh oh!

Conversation

polyzos commented Apr 2, 2025

Uh oh!

polyzos commented Apr 2, 2025

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

polyzos commented Apr 28, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

polyzos commented Apr 28, 2025 •

edited

Loading