You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "wwj6591812 (via GitHub)" <gi...@apache.org> on 2025/06/08 05:30:08 UTC
[PR] [Flink] Add batch read operator source. [paimon]
wwj6591812 opened a new pull request, #5715:
URL: https://github.com/apache/paimon/pull/5715
<!-- Please specify the module before the PR name: [core] ... or [flink] ... -->
### Purpose
In https://github.com/apache/paimon/pull/5650 we resolve the JM memory problem by support withShard read of Flink.
But that method has a drawback that each task will read a lot of the same Meta.
So, this pr resolve the problem by move the plan from JM to a single task, then this task assign the splits to the downstream readers.
<!-- Linking this pull request to the issue -->
Linked issue: close #xxx
<!-- What is the purpose of the change -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "yunfengzhou-hub (via GitHub)" <gi...@apache.org>.
yunfengzhou-hub commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2136796457
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/BatchReadSource.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * 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.paimon.flink.source.operator;
+
+import org.apache.paimon.flink.NestedProjectedRowData;
+import org.apache.paimon.flink.source.AbstractNonCoordinatedSource;
+import org.apache.paimon.flink.source.AbstractNonCoordinatedSourceReader;
+import org.apache.paimon.flink.source.SimpleSourceSplit;
+import org.apache.paimon.flink.utils.JavaTypeInfo;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableScan;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.ReaderOutput;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.data.RowData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+import static org.apache.paimon.flink.source.operator.MonitorSource.shuffleNonUnawareBucket;
+import static org.apache.paimon.flink.source.operator.MonitorSource.shuffleUnawareBucket;
+
+/**
+ * This is the single (non-parallel) batch task, it is responsible for:
+ *
+ * <ol>
+ * <li>Read the latest snapshot of the Paimon table.
+ * <li>Creating the {@link Split splits} corresponding to the DataFiles
+ * <li>Assigning them to downstream tasks for further processing.
+ * </ol>
+ *
+ * <p>The splits to be read are forwarded to the downstream {@link ReadOperator} which can have
+ * parallelism greater than one.
+ */
+public class BatchReadSource extends AbstractNonCoordinatedSource<Split> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(BatchReadSource.class);
+
+ private final ReadBuilder readBuilder;
+
+ private final String tableName;
+
+ public BatchReadSource(ReadBuilder readBuilder) {
+ this.readBuilder = readBuilder;
+ this.tableName = readBuilder.tableName();
+ }
+
+ @Override
+ public Boundedness getBoundedness() {
+ return Boundedness.BOUNDED;
+ }
+
+ @Override
+ public SourceReader<Split, SimpleSourceSplit> createReader(
+ SourceReaderContext sourceReaderContext) throws Exception {
+ return new Reader();
+ }
+
+ private class Reader extends AbstractNonCoordinatedSourceReader<Split> {
+
+ private final TableScan scan = readBuilder.newScan();
+
+ @Override
+ public InputStatus pollNext(ReaderOutput<Split> readerOutput) throws Exception {
+ List<Split> splits = scan.plan().splits();
+
+ if (splits.size() == 0) {
+ LOG.info("In BatchSource the table {} plan split is empty.", tableName);
+ } else {
+ DataSplit dataSplit = (DataSplit) splits.get(0);
+ LOG.info(
+ "Read the snapshot-{} of table {} in BatchSource.",
+ dataSplit.snapshotId(),
+ tableName);
+ }
+
+ splits.forEach(readerOutput::collect);
+
+ return InputStatus.END_OF_INPUT;
Review Comment:
How about reusing MonitorSource instead of introducing a new one? For example,
```java
if (isBounded) {
return InputStatus.END_OF_INPUT;
} else {
return InputStatus.MORE_AVAILABLE;
}
```
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java:
##########
@@ -294,6 +295,9 @@ public DataStream<RowData> build() {
}
if (sourceBounded) {
+ if (conf.get(CoreOptions.SCAN_BATCH_PLAN_MOVE_FROM_JM_TO_TM_ENABLED)) {
Review Comment:
If users configured this option and their job is in streaming mode, can this optimization still take effect? Would MonitorSource achieve a similar optimization in streaming mode?
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/BatchReadSource.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * 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.paimon.flink.source.operator;
+
+import org.apache.paimon.flink.NestedProjectedRowData;
+import org.apache.paimon.flink.source.AbstractNonCoordinatedSource;
+import org.apache.paimon.flink.source.AbstractNonCoordinatedSourceReader;
+import org.apache.paimon.flink.source.SimpleSourceSplit;
+import org.apache.paimon.flink.utils.JavaTypeInfo;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableScan;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.ReaderOutput;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.data.RowData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+import static org.apache.paimon.flink.source.operator.MonitorSource.shuffleNonUnawareBucket;
+import static org.apache.paimon.flink.source.operator.MonitorSource.shuffleUnawareBucket;
+
+/**
+ * This is the single (non-parallel) batch task, it is responsible for:
+ *
+ * <ol>
+ * <li>Read the latest snapshot of the Paimon table.
+ * <li>Creating the {@link Split splits} corresponding to the DataFiles
+ * <li>Assigning them to downstream tasks for further processing.
+ * </ol>
+ *
+ * <p>The splits to be read are forwarded to the downstream {@link ReadOperator} which can have
+ * parallelism greater than one.
+ */
+public class BatchReadSource extends AbstractNonCoordinatedSource<Split> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(BatchReadSource.class);
+
+ private final ReadBuilder readBuilder;
+
+ private final String tableName;
+
+ public BatchReadSource(ReadBuilder readBuilder) {
+ this.readBuilder = readBuilder;
+ this.tableName = readBuilder.tableName();
+ }
+
+ @Override
+ public Boundedness getBoundedness() {
+ return Boundedness.BOUNDED;
+ }
+
+ @Override
+ public SourceReader<Split, SimpleSourceSplit> createReader(
+ SourceReaderContext sourceReaderContext) throws Exception {
+ return new Reader();
+ }
+
+ private class Reader extends AbstractNonCoordinatedSourceReader<Split> {
+
+ private final TableScan scan = readBuilder.newScan();
+
+ @Override
+ public InputStatus pollNext(ReaderOutput<Split> readerOutput) throws Exception {
+ List<Split> splits = scan.plan().splits();
+
+ if (splits.size() == 0) {
+ LOG.info("In BatchSource the table {} plan split is empty.", tableName);
+ } else {
+ DataSplit dataSplit = (DataSplit) splits.get(0);
+ LOG.info(
+ "Read the snapshot-{} of table {} in BatchSource.",
+ dataSplit.snapshotId(),
+ tableName);
+ }
+
+ splits.forEach(readerOutput::collect);
+
+ return InputStatus.END_OF_INPUT;
+ }
+ }
+
+ public static DataStream<RowData> buildSource(
+ StreamExecutionEnvironment env,
+ String name,
+ TypeInformation<RowData> typeInfo,
+ ReadBuilder readBuilder,
+ boolean shuffleBucketWithPartition,
+ boolean unawareBucket,
+ NestedProjectedRowData nestedProjectedRowData,
+ Long limit) {
+ SingleOutputStreamOperator<Split> singleOutputStreamOperator =
+ env.fromSource(
+ new BatchReadSource(readBuilder),
+ WatermarkStrategy.noWatermarks(),
+ name + "-BatchRead",
+ new JavaTypeInfo<>(Split.class))
+ .forceNonParallel();
+
+ DataStream<Split> sourceDataStream =
+ unawareBucket
+ ? shuffleUnawareBucket(singleOutputStreamOperator)
+ : shuffleNonUnawareBucket(
+ singleOutputStreamOperator, shuffleBucketWithPartition);
+
+ return sourceDataStream.transform(
+ name + "-Reader",
+ typeInfo,
+ new ReadOperator(readBuilder::newRead, nestedProjectedRowData, limit));
Review Comment:
Unlike streaming mode, in batch mode Flink would enable auto parallelism inference by default. Given that the parallelism of downstream operators would be inferred according to `execution.batch.adaptive.auto-parallelism.avg-data-volume-per-task`, and the plan splits transmitted here are usually of very small size, I'm a little concerned that the parallelism of readOperator will be inferred into a small value, and users have to manually set the parallelism of this operator to the correct range, which obeys the convenient brought by auto parallelism inference. What do you think?
##########
paimon-api/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -1744,6 +1744,13 @@ public class CoreOptions implements Serializable {
"If true, it disables explicit type casting. For ex: it disables converting LONG type to INT type. "
+ "Users can enable this option to disable explicit type casting");
+ public static final ConfigOption<Boolean> SCAN_BATCH_PLAN_MOVE_FROM_JM_TO_TM_ENABLED =
+ key("scan.batch.plan.move-from-jm-to-tm.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "If true, scan plan will run on a single TM, you can enable it when Paimon source coordinator encounter with memory bottleneck in Flink batch job. This feature is not bounded with the consumer mechanism and it will change the DAG of the flink job, thus breaking checkpoint compatibility. So this option is not enabled by default and users have to consider this side effect before enabling it.");
Review Comment:
The current configuration name and description still seems complicated to me. In order to use this configuration from the current description, users have to know Flink and Paimon implementation details such as plan, jm, tm, which influences the readability of this configuration. Is there some better naming and description that could resolve this issue?
Besides, you can use Description.builder() to avoid a long line of description.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java:
##########
@@ -130,6 +137,14 @@ public void processElement(StreamRecord<Split> record) throws Exception {
reuseRecord.replace(nestedProjectedRowData);
}
output.collect(reuseRecord);
+
+ if (limit != null && numRecordsIn.getCount() == limit) {
+ LOG.info(
+ "When limit is not null, reader {} reach the limit record {}.",
+ this,
+ limit);
+ return;
Review Comment:
When will this condition be met? limit should have been applied to the ReadBuilder in FlinkSourceBuilder#createReadBuilder, so the limit should not come into effect here.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2136980337
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/BatchReadSource.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * 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.paimon.flink.source.operator;
+
+import org.apache.paimon.flink.NestedProjectedRowData;
+import org.apache.paimon.flink.source.AbstractNonCoordinatedSource;
+import org.apache.paimon.flink.source.AbstractNonCoordinatedSourceReader;
+import org.apache.paimon.flink.source.SimpleSourceSplit;
+import org.apache.paimon.flink.utils.JavaTypeInfo;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableScan;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.ReaderOutput;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.data.RowData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+import static org.apache.paimon.flink.source.operator.MonitorSource.shuffleNonUnawareBucket;
+import static org.apache.paimon.flink.source.operator.MonitorSource.shuffleUnawareBucket;
+
+/**
+ * This is the single (non-parallel) batch task, it is responsible for:
+ *
+ * <ol>
+ * <li>Read the latest snapshot of the Paimon table.
+ * <li>Creating the {@link Split splits} corresponding to the DataFiles
+ * <li>Assigning them to downstream tasks for further processing.
+ * </ol>
+ *
+ * <p>The splits to be read are forwarded to the downstream {@link ReadOperator} which can have
+ * parallelism greater than one.
+ */
+public class BatchReadSource extends AbstractNonCoordinatedSource<Split> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(BatchReadSource.class);
+
+ private final ReadBuilder readBuilder;
+
+ private final String tableName;
+
+ public BatchReadSource(ReadBuilder readBuilder) {
+ this.readBuilder = readBuilder;
+ this.tableName = readBuilder.tableName();
+ }
+
+ @Override
+ public Boundedness getBoundedness() {
+ return Boundedness.BOUNDED;
+ }
+
+ @Override
+ public SourceReader<Split, SimpleSourceSplit> createReader(
+ SourceReaderContext sourceReaderContext) throws Exception {
+ return new Reader();
+ }
+
+ private class Reader extends AbstractNonCoordinatedSourceReader<Split> {
+
+ private final TableScan scan = readBuilder.newScan();
+
+ @Override
+ public InputStatus pollNext(ReaderOutput<Split> readerOutput) throws Exception {
+ List<Split> splits = scan.plan().splits();
+
+ if (splits.size() == 0) {
+ LOG.info("In BatchSource the table {} plan split is empty.", tableName);
+ } else {
+ DataSplit dataSplit = (DataSplit) splits.get(0);
+ LOG.info(
+ "Read the snapshot-{} of table {} in BatchSource.",
+ dataSplit.snapshotId(),
+ tableName);
+ }
+
+ splits.forEach(readerOutput::collect);
+
+ return InputStatus.END_OF_INPUT;
+ }
+ }
+
+ public static DataStream<RowData> buildSource(
+ StreamExecutionEnvironment env,
+ String name,
+ TypeInformation<RowData> typeInfo,
+ ReadBuilder readBuilder,
+ boolean shuffleBucketWithPartition,
+ boolean unawareBucket,
+ NestedProjectedRowData nestedProjectedRowData,
+ Long limit) {
+ SingleOutputStreamOperator<Split> singleOutputStreamOperator =
+ env.fromSource(
+ new BatchReadSource(readBuilder),
+ WatermarkStrategy.noWatermarks(),
+ name + "-BatchRead",
+ new JavaTypeInfo<>(Split.class))
+ .forceNonParallel();
+
+ DataStream<Split> sourceDataStream =
+ unawareBucket
+ ? shuffleUnawareBucket(singleOutputStreamOperator)
+ : shuffleNonUnawareBucket(
+ singleOutputStreamOperator, shuffleBucketWithPartition);
+
+ return sourceDataStream.transform(
+ name + "-Reader",
+ typeInfo,
+ new ReadOperator(readBuilder::newRead, nestedProjectedRowData, limit));
Review Comment:
Thanks very much, you are right.
I have added some description in the option's description.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2137004291
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/BatchReadSource.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * 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.paimon.flink.source.operator;
+
+import org.apache.paimon.flink.NestedProjectedRowData;
+import org.apache.paimon.flink.source.AbstractNonCoordinatedSource;
+import org.apache.paimon.flink.source.AbstractNonCoordinatedSourceReader;
+import org.apache.paimon.flink.source.SimpleSourceSplit;
+import org.apache.paimon.flink.utils.JavaTypeInfo;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableScan;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.ReaderOutput;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.data.RowData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+import static org.apache.paimon.flink.source.operator.MonitorSource.shuffleNonUnawareBucket;
+import static org.apache.paimon.flink.source.operator.MonitorSource.shuffleUnawareBucket;
+
+/**
+ * This is the single (non-parallel) batch task, it is responsible for:
+ *
+ * <ol>
+ * <li>Read the latest snapshot of the Paimon table.
+ * <li>Creating the {@link Split splits} corresponding to the DataFiles
+ * <li>Assigning them to downstream tasks for further processing.
+ * </ol>
+ *
+ * <p>The splits to be read are forwarded to the downstream {@link ReadOperator} which can have
+ * parallelism greater than one.
+ */
+public class BatchReadSource extends AbstractNonCoordinatedSource<Split> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(BatchReadSource.class);
+
+ private final ReadBuilder readBuilder;
+
+ private final String tableName;
+
+ public BatchReadSource(ReadBuilder readBuilder) {
+ this.readBuilder = readBuilder;
+ this.tableName = readBuilder.tableName();
+ }
+
+ @Override
+ public Boundedness getBoundedness() {
+ return Boundedness.BOUNDED;
+ }
+
+ @Override
+ public SourceReader<Split, SimpleSourceSplit> createReader(
+ SourceReaderContext sourceReaderContext) throws Exception {
+ return new Reader();
+ }
+
+ private class Reader extends AbstractNonCoordinatedSourceReader<Split> {
+
+ private final TableScan scan = readBuilder.newScan();
+
+ @Override
+ public InputStatus pollNext(ReaderOutput<Split> readerOutput) throws Exception {
+ List<Split> splits = scan.plan().splits();
+
+ if (splits.size() == 0) {
+ LOG.info("In BatchSource the table {} plan split is empty.", tableName);
+ } else {
+ DataSplit dataSplit = (DataSplit) splits.get(0);
+ LOG.info(
+ "Read the snapshot-{} of table {} in BatchSource.",
+ dataSplit.snapshotId(),
+ tableName);
+ }
+
+ splits.forEach(readerOutput::collect);
+
+ return InputStatus.END_OF_INPUT;
Review Comment:
I don't want to reusing MonitorSource instead of BatchReadSource, because:
1、MonitorSource is a periodic check for new snapshots, which semantically means streaming processing, not batch processing.
2、There is lots of checkpoint states logic in MonitorSource, thus not used in batch mode. And the BatchReadSource's logic is simple.
3、It will add some if...else... in MonitorSource, for example in MonitorSource#getBoundedness、MonitorSource#Reader#pollNext、MonitorSource#addSplits.
So I don't think reusing MonitorSource is a better choose.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "yunfengzhou-hub (via GitHub)" <gi...@apache.org>.
yunfengzhou-hub commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2139550621
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java:
##########
@@ -294,6 +295,9 @@ public DataStream<RowData> build() {
}
if (sourceBounded) {
+ if (conf.get(CoreOptions.SCAN_GET_SPLITS_ON_SINGLE_TASK)) {
Review Comment:
Can we reuse CoreOptions#scanGetSplitsOnSingleTask here?
##########
paimon-api/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -1753,6 +1753,27 @@ public class CoreOptions implements Serializable {
+ "starting from the snapshot after this one. If found, commit will be aborted. "
+ "If the value of this option is -1, committer will not check for its first commit.");
+ public static final ConfigOption<Boolean> SCAN_GET_SPLITS_ON_SINGLE_TASK =
+ key("scan.get-splits-on-single-task")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ Description.builder()
+ .text(
+ "If true, generate splits when scan will run on a single task, this can resolve Paimon source coordinator encounter with memory bottleneck in Flink batch job.")
+ .linebreak()
+ .text("This feature have some side effects, for example : ")
+ .linebreak()
+ .text(
+ "1. It will change the DAG of the flink job, thus breaking checkpoint compatibility.")
+ .linebreak()
+ .text(
+ "2. It may infer a small parallelism of reader if you enable Flink AdaptiveBatchScheduler. You can set Flink CoreOptions.DEFAULT_PARALLELISM to avoid.")
+ .linebreak()
+ .text(
+ "So it's default value is false and you have to consider the side effects before enable it.")
Review Comment:
How about the following?
* key: scan.dedicated-split-generation
* description: If true, the split generation process would be performed during runtime on a Flink task, instead of on the JobManager during initialization phase.
As for the concerns users need to consider, we can put them in Paimon's document like the Query Performance document. In this way we can simplify the description here to avoid including too much non-critical information.
Besides, it comes to me that there is another concern for users: when this feature is enabled, the original possible regional failover would be changed into global failover now, as we have a BatchSource subtask that connects to all downstream ReadOperators. We can also add this bullet to the list in document.
##########
paimon-api/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -1753,6 +1753,27 @@ public class CoreOptions implements Serializable {
+ "starting from the snapshot after this one. If found, commit will be aborted. "
+ "If the value of this option is -1, committer will not check for its first commit.");
+ public static final ConfigOption<Boolean> SCAN_GET_SPLITS_ON_SINGLE_TASK =
Review Comment:
This option should be moved to FlinkConnectorOptions, instead of at CoreOptions.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2141674838
##########
docs/content/append-table/query-performance.md:
##########
@@ -75,3 +75,6 @@ we use the procedure, you should config appropriate configurations in target tab
`file-index.<filter-type>.columns` to the table.
How to invoke: see [flink procedures]({{< ref "flink/procedures#procedures" >}})
+
+## Optimize Memory By Dedicated Split Generation
+see [query performance]({{< ref "primary-key-table/query-performance#Optimize Memory By Dedicated Split Generation" >}})
Review Comment:
thanks
done
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "yunfengzhou-hub (via GitHub)" <gi...@apache.org>.
yunfengzhou-hub commented on PR #5715:
URL: https://github.com/apache/paimon/pull/5715#issuecomment-2965336590
Thanks for the update. LGTM
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2135286106
##########
docs/layouts/shortcodes/generated/core_configuration.html:
##########
@@ -228,7 +228,7 @@
<td><h5>consumer.mode</h5></td>
<td style="word-wrap: break-word;">exactly-once</td>
<td><p>Enum</p></td>
- <td>Specify the consumer consistency mode for table.<br /><br />Possible values:<ul><li>"exactly-once": Readers consume data at snapshot granularity, and strictly ensure that the snapshot-id recorded in the consumer is the snapshot-id + 1 that all readers have exactly consumed.</li><li>"at-least-once": Each reader consumes snapshots at a different rate, and the snapshot with the slowest consumption progress among all readers will be recorded in the consumer.</li></ul></td>
+ <td>Specify the consumer consistency mode for table.<br /><br />Possible values:<ul><li>"exactly-once": Readers consume data at snapshot granularity, and strictly ensure that the snapshot-id recorded in the consumer is the snapshot-id + 1 that all readers have exactly consumed.</li><li>"exactly-once-for-batch": Readers consume the latest snapshot data with batch mood, and the plan will run on TM.</li><li>"at-least-once": Each reader consumes snapshots at a different rate, and the snapshot with the slowest consumption progress among all readers will be recorded in the consumer.</li></ul></td>
Review Comment:
Very thanks, done.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2139749134
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java:
##########
@@ -294,6 +295,9 @@ public DataStream<RowData> build() {
}
if (sourceBounded) {
+ if (conf.get(CoreOptions.SCAN_GET_SPLITS_ON_SINGLE_TASK)) {
Review Comment:
moved to FlinkConnectorOptions
##########
paimon-api/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -1753,6 +1753,27 @@ public class CoreOptions implements Serializable {
+ "starting from the snapshot after this one. If found, commit will be aborted. "
+ "If the value of this option is -1, committer will not check for its first commit.");
+ public static final ConfigOption<Boolean> SCAN_GET_SPLITS_ON_SINGLE_TASK =
Review Comment:
thanks, done
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2141672321
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableScanUtils.java:
##########
@@ -45,14 +50,22 @@ public static void streamingReadingValidate(Table table) {
}
};
if (table.primaryKeys().size() > 0 && mergeEngineDesc.containsKey(mergeEngine)) {
- if (options.changelogProducer() == CoreOptions.ChangelogProducer.NONE) {
+ if (coreOptions.changelogProducer() == CoreOptions.ChangelogProducer.NONE) {
throw new RuntimeException(
mergeEngineDesc.get(mergeEngine)
+ " streaming reading is not supported. You can use "
+ "'lookup' or 'full-compaction' changelog producer to support streaming reading. "
+ "('input' changelog producer is also supported, but only returns input records.)");
}
}
+
+ if (options.containsKey(SCAN_DEDICATED_SPLIT_GENERATION.key())
+ && parseBoolean(options.get(SCAN_DEDICATED_SPLIT_GENERATION.key()))) {
Review Comment:
Thanks.
Change `Map<String, String> options = table.options();` to `Options options = Options.fromMap(table.options());`
Then cut the parseBoolean statement.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java:
##########
@@ -130,6 +137,11 @@ public void processElement(StreamRecord<Split> record) throws Exception {
reuseRecord.replace(nestedProjectedRowData);
}
output.collect(reuseRecord);
+
+ if (limit != null && numRecordsIn.getCount() == limit) {
Review Comment:
done, thx.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2136831526
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java:
##########
@@ -130,6 +137,14 @@ public void processElement(StreamRecord<Split> record) throws Exception {
reuseRecord.replace(nestedProjectedRowData);
}
output.collect(reuseRecord);
+
+ if (limit != null && numRecordsIn.getCount() == limit) {
+ LOG.info(
+ "When limit is not null, reader {} reach the limit record {}.",
+ this,
+ limit);
+ return;
Review Comment:
FlinkSourceBuilder#buildBatchReadOperator -> BatchReadSource#buildSource -> ReadOperator#ReadOperator
I have test whether add this condition. The result is :
一、with this condition（select * from `paimon-1.2-ali-165091-0608-001`.`tddl_sc_p4p_new_app_na175`.`p4p_cust_property` /*+ OPTIONS('scan.parallelism'='30', 'consumer.mode'='exactly-once-for-batch') */limit 10;）
二、without this condition（select * from `paimon-1.2-ali-165091-0608-002`.`tddl_sc_p4p_new_app_na175`.`p4p_cust_property` /*+ OPTIONS('scan.parallelism'='30', 'consumer.mode'='exactly-once-for-batch') */limit 10;
）
So, I think the limit should come into effect here.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "yunfengzhou-hub (via GitHub)" <gi...@apache.org>.
yunfengzhou-hub commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2141355829
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableScanUtils.java:
##########
@@ -45,14 +50,22 @@ public static void streamingReadingValidate(Table table) {
}
};
if (table.primaryKeys().size() > 0 && mergeEngineDesc.containsKey(mergeEngine)) {
- if (options.changelogProducer() == CoreOptions.ChangelogProducer.NONE) {
+ if (coreOptions.changelogProducer() == CoreOptions.ChangelogProducer.NONE) {
throw new RuntimeException(
mergeEngineDesc.get(mergeEngine)
+ " streaming reading is not supported. You can use "
+ "'lookup' or 'full-compaction' changelog producer to support streaming reading. "
+ "('input' changelog producer is also supported, but only returns input records.)");
}
}
+
+ if (options.containsKey(SCAN_DEDICATED_SPLIT_GENERATION.key())
+ && parseBoolean(options.get(SCAN_DEDICATED_SPLIT_GENERATION.key()))) {
Review Comment:
nit: change `CoreOptions coreOptions = CoreOptions.fromMap(table.options());` to `Options options = Options.fromMap(table.options());`, and then we won't need parseBoolean here.
##########
docs/content/primary-key-table/query-performance.md:
##########
@@ -73,3 +73,14 @@ we use the procedure, you should config appropriate configurations in target tab
`file-index.<filter-type>.columns` to the table.
How to invoke: see [flink procedures]({{< ref "flink/procedures#procedures" >}})
+
+## Optimize Memory By Dedicated Split Generation
+If you facing a Flink Paimon source coordinator memory bottleneck in batch job. You can set this option to true.
+* 'scan.dedicated-split-generation' = 'true'
+
+This feature have some side effects, for example :
+1. It will change the DAG of the flink job, thus breaking checkpoint compatibility.
+2. It may infer a small parallelism of reader if you enable Flink AdaptiveBatchScheduler. You can set Flink CoreOptions.DEFAULT_PARALLELISM to avoid.
+3. It will change the failover strategy from the original possible region failover to global failover. Because a Monitor subtask that connects to all downstream ReadOperators.
+So it's default value is false and you have to consider the side effects before enable it.
Review Comment:
How about the following description?
```markdown
## Dedicated Split Generation
When Paimon table snapshots contain large amount of source splits, Flink jobs reading from this table might endure long initialization time or even OOM in JobManagers. In this case, you can configure `'scan.dedicated-split-generation' = 'true'` to avoid such problem. This option would enable executing the source split generation process in a dedicated subtask that runs on TaskManager, instead of in the source coordinator on the JobManager.
Note that this feature could have some side effects on your Flink jobs. For example:
1. It will change the DAG of the flink job, thus breaking checkpoint compatibility if enabled on an existing job.
2. It may lead to the Flink AdaptiveBatchScheduler inferring a small parallelism for the source reader operator. you can configure `scan.infer-parallelism` to avoid this possible drawback.
3. The failover strategy of the Flink job would be forced into global failover instead of regional failover, given that the dedicated source split generation task would be connected to all downstream subtasks.
So please make sure these side effects are acceptable to you before enabling it.
```
##########
docs/content/append-table/query-performance.md:
##########
@@ -75,3 +75,6 @@ we use the procedure, you should config appropriate configurations in target tab
`file-index.<filter-type>.columns` to the table.
How to invoke: see [flink procedures]({{< ref "flink/procedures#procedures" >}})
+
+## Optimize Memory By Dedicated Split Generation
+see [query performance]({{< ref "primary-key-table/query-performance#Optimize Memory By Dedicated Split Generation" >}})
Review Comment:
It might be better to directly copy the content from the other document here instead of adding a link.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java:
##########
@@ -130,6 +137,11 @@ public void processElement(StreamRecord<Split> record) throws Exception {
reuseRecord.replace(nestedProjectedRowData);
}
output.collect(reuseRecord);
+
+ if (limit != null && numRecordsIn.getCount() == limit) {
Review Comment:
When the records in the first split have reached this limit, the read operator would still consume at least one record from the next received split, causing `numRecordsIn.getCount() == (limit + 1) != limit`. Thus it is better to move this logic to before iterator.next, and change such condition into `numRecordsIn.getCount() >= limit`.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2136847431
##########
paimon-api/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -1744,6 +1744,13 @@ public class CoreOptions implements Serializable {
"If true, it disables explicit type casting. For ex: it disables converting LONG type to INT type. "
+ "Users can enable this option to disable explicit type casting");
+ public static final ConfigOption<Boolean> SCAN_BATCH_PLAN_MOVE_FROM_JM_TO_TM_ENABLED =
+ key("scan.batch.plan.move-from-jm-to-tm.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "If true, scan plan will run on a single TM, you can enable it when Paimon source coordinator encounter with memory bottleneck in Flink batch job. This feature is not bounded with the consumer mechanism and it will change the DAG of the flink job, thus breaking checkpoint compatibility. So this option is not enabled by default and users have to consider this side effect before enabling it.");
Review Comment:
Thanks, done.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "yunfengzhou-hub (via GitHub)" <gi...@apache.org>.
yunfengzhou-hub commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2135098334
##########
docs/layouts/shortcodes/generated/core_configuration.html:
##########
@@ -228,7 +228,7 @@
<td><h5>consumer.mode</h5></td>
<td style="word-wrap: break-word;">exactly-once</td>
<td><p>Enum</p></td>
- <td>Specify the consumer consistency mode for table.<br /><br />Possible values:<ul><li>"exactly-once": Readers consume data at snapshot granularity, and strictly ensure that the snapshot-id recorded in the consumer is the snapshot-id + 1 that all readers have exactly consumed.</li><li>"at-least-once": Each reader consumes snapshots at a different rate, and the snapshot with the slowest consumption progress among all readers will be recorded in the consumer.</li></ul></td>
+ <td>Specify the consumer consistency mode for table.<br /><br />Possible values:<ul><li>"exactly-once": Readers consume data at snapshot granularity, and strictly ensure that the snapshot-id recorded in the consumer is the snapshot-id + 1 that all readers have exactly consumed.</li><li>"exactly-once-for-batch": Readers consume the latest snapshot data with batch mood, and the plan will run on TM.</li><li>"at-least-once": Each reader consumes snapshots at a different rate, and the snapshot with the slowest consumption progress among all readers will be recorded in the consumer.</li></ul></td>
Review Comment:
consumer.mode is supposed to be used together with consumer.id, and it seems that exactly-once-for-batch does not mean so. Thus how about introducing an individual option that have the following information for users?
1. The option could notify users that this feature is not bound with the consumer mechanism. They can enable this feature so long as they are facing a Flink Paimon source coordinator bottleneck in batch job.
2. The option changes the DAG of the Flink job, thus breaking checkpoint compatibility. So this option is not enabled by default and users have to consider this side effect before enabling 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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on PR #5715:
URL: https://github.com/apache/paimon/pull/5715#issuecomment-2954831551
The failed ut is not caused by my modification.
Please CC, Thx.
@JingsongLi
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2139749793
##########
paimon-api/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -1753,6 +1753,27 @@ public class CoreOptions implements Serializable {
+ "starting from the snapshot after this one. If found, commit will be aborted. "
+ "If the value of this option is -1, committer will not check for its first commit.");
+ public static final ConfigOption<Boolean> SCAN_GET_SPLITS_ON_SINGLE_TASK =
+ key("scan.get-splits-on-single-task")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ Description.builder()
+ .text(
+ "If true, generate splits when scan will run on a single task, this can resolve Paimon source coordinator encounter with memory bottleneck in Flink batch job.")
+ .linebreak()
+ .text("This feature have some side effects, for example : ")
+ .linebreak()
+ .text(
+ "1. It will change the DAG of the flink job, thus breaking checkpoint compatibility.")
+ .linebreak()
+ .text(
+ "2. It may infer a small parallelism of reader if you enable Flink AdaptiveBatchScheduler. You can set Flink CoreOptions.DEFAULT_PARALLELISM to avoid.")
+ .linebreak()
+ .text(
+ "So it's default value is false and you have to consider the side effects before enable it.")
Review Comment:
good idea! done
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2136831526
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java:
##########
@@ -130,6 +137,14 @@ public void processElement(StreamRecord<Split> record) throws Exception {
reuseRecord.replace(nestedProjectedRowData);
}
output.collect(reuseRecord);
+
+ if (limit != null && numRecordsIn.getCount() == limit) {
+ LOG.info(
+ "When limit is not null, reader {} reach the limit record {}.",
+ this,
+ limit);
+ return;
Review Comment:
FlinkSourceBuilder#buildBatchReadOperator -> BatchReadSource#buildSource -> ReadOperator#ReadOperator
I have test whether add this condition. The result is :
一、without this condition（select * from `paimon-1.2-ali-165091-0608-001`.`tddl_sc_p4p_new_app_na175`.`p4p_cust_property` /*+ OPTIONS('scan.parallelism'='30', 'consumer.mode'='exactly-once-for-batch') */limit 10;）
二、with this condition（select * from `paimon-1.2-ali-165091-0608-002`.`tddl_sc_p4p_new_app_na175`.`p4p_cust_property` /*+ OPTIONS('scan.parallelism'='30', 'consumer.mode'='exactly-once-for-batch') */limit 10;
）
So, I think the limit should come into effect here.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2137004291
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/BatchReadSource.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * 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.paimon.flink.source.operator;
+
+import org.apache.paimon.flink.NestedProjectedRowData;
+import org.apache.paimon.flink.source.AbstractNonCoordinatedSource;
+import org.apache.paimon.flink.source.AbstractNonCoordinatedSourceReader;
+import org.apache.paimon.flink.source.SimpleSourceSplit;
+import org.apache.paimon.flink.utils.JavaTypeInfo;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableScan;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.ReaderOutput;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.data.RowData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+import static org.apache.paimon.flink.source.operator.MonitorSource.shuffleNonUnawareBucket;
+import static org.apache.paimon.flink.source.operator.MonitorSource.shuffleUnawareBucket;
+
+/**
+ * This is the single (non-parallel) batch task, it is responsible for:
+ *
+ * <ol>
+ * <li>Read the latest snapshot of the Paimon table.
+ * <li>Creating the {@link Split splits} corresponding to the DataFiles
+ * <li>Assigning them to downstream tasks for further processing.
+ * </ol>
+ *
+ * <p>The splits to be read are forwarded to the downstream {@link ReadOperator} which can have
+ * parallelism greater than one.
+ */
+public class BatchReadSource extends AbstractNonCoordinatedSource<Split> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(BatchReadSource.class);
+
+ private final ReadBuilder readBuilder;
+
+ private final String tableName;
+
+ public BatchReadSource(ReadBuilder readBuilder) {
+ this.readBuilder = readBuilder;
+ this.tableName = readBuilder.tableName();
+ }
+
+ @Override
+ public Boundedness getBoundedness() {
+ return Boundedness.BOUNDED;
+ }
+
+ @Override
+ public SourceReader<Split, SimpleSourceSplit> createReader(
+ SourceReaderContext sourceReaderContext) throws Exception {
+ return new Reader();
+ }
+
+ private class Reader extends AbstractNonCoordinatedSourceReader<Split> {
+
+ private final TableScan scan = readBuilder.newScan();
+
+ @Override
+ public InputStatus pollNext(ReaderOutput<Split> readerOutput) throws Exception {
+ List<Split> splits = scan.plan().splits();
+
+ if (splits.size() == 0) {
+ LOG.info("In BatchSource the table {} plan split is empty.", tableName);
+ } else {
+ DataSplit dataSplit = (DataSplit) splits.get(0);
+ LOG.info(
+ "Read the snapshot-{} of table {} in BatchSource.",
+ dataSplit.snapshotId(),
+ tableName);
+ }
+
+ splits.forEach(readerOutput::collect);
+
+ return InputStatus.END_OF_INPUT;
Review Comment:
I don't want to reusing MonitorSource instead of BatchReadSource, because:
1、MonitorSource is a periodic check for new snapshots, which semantically means streaming processing, not batch processing.
2、There is lots of checkpoint states logic in MonitorSource, thus not used in batch mode.
3、It will add some if...else... in MonitorSource, for example in MonitorSource#getBoundedness、MonitorSource#Reader#pollNext.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2137004291
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/BatchReadSource.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * 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.paimon.flink.source.operator;
+
+import org.apache.paimon.flink.NestedProjectedRowData;
+import org.apache.paimon.flink.source.AbstractNonCoordinatedSource;
+import org.apache.paimon.flink.source.AbstractNonCoordinatedSourceReader;
+import org.apache.paimon.flink.source.SimpleSourceSplit;
+import org.apache.paimon.flink.utils.JavaTypeInfo;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableScan;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.ReaderOutput;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.data.RowData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+import static org.apache.paimon.flink.source.operator.MonitorSource.shuffleNonUnawareBucket;
+import static org.apache.paimon.flink.source.operator.MonitorSource.shuffleUnawareBucket;
+
+/**
+ * This is the single (non-parallel) batch task, it is responsible for:
+ *
+ * <ol>
+ * <li>Read the latest snapshot of the Paimon table.
+ * <li>Creating the {@link Split splits} corresponding to the DataFiles
+ * <li>Assigning them to downstream tasks for further processing.
+ * </ol>
+ *
+ * <p>The splits to be read are forwarded to the downstream {@link ReadOperator} which can have
+ * parallelism greater than one.
+ */
+public class BatchReadSource extends AbstractNonCoordinatedSource<Split> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(BatchReadSource.class);
+
+ private final ReadBuilder readBuilder;
+
+ private final String tableName;
+
+ public BatchReadSource(ReadBuilder readBuilder) {
+ this.readBuilder = readBuilder;
+ this.tableName = readBuilder.tableName();
+ }
+
+ @Override
+ public Boundedness getBoundedness() {
+ return Boundedness.BOUNDED;
+ }
+
+ @Override
+ public SourceReader<Split, SimpleSourceSplit> createReader(
+ SourceReaderContext sourceReaderContext) throws Exception {
+ return new Reader();
+ }
+
+ private class Reader extends AbstractNonCoordinatedSourceReader<Split> {
+
+ private final TableScan scan = readBuilder.newScan();
+
+ @Override
+ public InputStatus pollNext(ReaderOutput<Split> readerOutput) throws Exception {
+ List<Split> splits = scan.plan().splits();
+
+ if (splits.size() == 0) {
+ LOG.info("In BatchSource the table {} plan split is empty.", tableName);
+ } else {
+ DataSplit dataSplit = (DataSplit) splits.get(0);
+ LOG.info(
+ "Read the snapshot-{} of table {} in BatchSource.",
+ dataSplit.snapshotId(),
+ tableName);
+ }
+
+ splits.forEach(readerOutput::collect);
+
+ return InputStatus.END_OF_INPUT;
Review Comment:
I don't want to reusing MonitorSource instead of BatchReadSource, because:
1、MonitorSource is a periodic check for new snapshots, which semantically means streaming processing, not batch processing.
2、There is lots of checkpoint states logic in MonitorSource, thus not used in batch mode. And the BatchReadSource's logic is simple.
3、It will add some if...else... in MonitorSource, for example in MonitorSource#getBoundedness、MonitorSource#Reader#pollNext.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2141672415
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java:
##########
@@ -130,6 +137,11 @@ public void processElement(StreamRecord<Split> record) throws Exception {
reuseRecord.replace(nestedProjectedRowData);
}
output.collect(reuseRecord);
+
+ if (limit != null && numRecordsIn.getCount() == limit) {
Review Comment:
done, thx.
And add this condition check at the beginning of processElement. This can pass the follow splits if already reach the limit.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2136855051
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java:
##########
@@ -294,6 +295,9 @@ public DataStream<RowData> build() {
}
if (sourceBounded) {
+ if (conf.get(CoreOptions.SCAN_BATCH_PLAN_MOVE_FROM_JM_TO_TM_ENABLED)) {
Review Comment:
Thanks.
1、Add some description in the option's description.
2、Add parameter check in TableScanUtils#streamingReadingValidate.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2139414488
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java:
##########
@@ -130,6 +137,14 @@ public void processElement(StreamRecord<Split> record) throws Exception {
reuseRecord.replace(nestedProjectedRowData);
}
output.collect(reuseRecord);
+
+ if (limit != null && numRecordsIn.getCount() == limit) {
+ LOG.info(
+ "When limit is not null, reader {} reach the limit record {}.",
+ this,
+ limit);
+ return;
Review Comment:
@yunfengzhou-hub
Hi, thanks for your reply.
1、The reason why I add this condition here is :
The reader should end as soon as possible after reading the limit data, it should not read unnecessary data.
2、The limit of FlinkSourceBuilder#createReadBuilder is used in DataTableBatchScan#applyPushDownLimit which try it best to cut splits.
3、In testBatchReadSourceWithSnapshot, remove this logic from ReadOperator, the `SELECT ... LIMIT 2` syntax still functioned correctly, the reason is:
Flink plan will generate a global limit operator, it will limit 2.
![Uploading image.png…]()
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java:
##########
@@ -130,6 +137,14 @@ public void processElement(StreamRecord<Split> record) throws Exception {
reuseRecord.replace(nestedProjectedRowData);
}
output.collect(reuseRecord);
+
+ if (limit != null && numRecordsIn.getCount() == limit) {
+ LOG.info(
+ "When limit is not null, reader {} reach the limit record {}.",
+ this,
+ limit);
+ return;
Review Comment:
@yunfengzhou-hub
Hi, thanks for your reply.
1、The reason why I add this condition here is :
The reader should end as soon as possible after reading the limit data, it should not read unnecessary data.
2、The limit of FlinkSourceBuilder#createReadBuilder is used in DataTableBatchScan#applyPushDownLimit which try it best to cut splits.
3、In testBatchReadSourceWithSnapshot, remove this logic from ReadOperator, the `SELECT ... LIMIT 2` syntax still functioned correctly, the reason is:
Flink plan will generate a global limit operator, it will limit 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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on PR #5715:
URL: https://github.com/apache/paimon/pull/5715#issuecomment-2958932419
@yunfengzhou-hub
hi, I have addressd, thanks.
please CC
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2139414488
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java:
##########
@@ -130,6 +137,14 @@ public void processElement(StreamRecord<Split> record) throws Exception {
reuseRecord.replace(nestedProjectedRowData);
}
output.collect(reuseRecord);
+
+ if (limit != null && numRecordsIn.getCount() == limit) {
+ LOG.info(
+ "When limit is not null, reader {} reach the limit record {}.",
+ this,
+ limit);
+ return;
Review Comment:
@yunfengzhou-hub
Hi, thanks for your reply.
The reason why I add this condition here is :
The reader should end as soon as possible after reading the limit data, it should not read unnecessary data.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on PR #5715:
URL: https://github.com/apache/paimon/pull/5715#issuecomment-2965066392
@yunfengzhou-hub
Hi, I have addressd, thanks.
please CC
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2141674302
##########
docs/content/primary-key-table/query-performance.md:
##########
@@ -73,3 +73,14 @@ we use the procedure, you should config appropriate configurations in target tab
`file-index.<filter-type>.columns` to the table.
How to invoke: see [flink procedures]({{< ref "flink/procedures#procedures" >}})
+
+## Optimize Memory By Dedicated Split Generation
+If you facing a Flink Paimon source coordinator memory bottleneck in batch job. You can set this option to true.
+* 'scan.dedicated-split-generation' = 'true'
+
+This feature have some side effects, for example :
+1. It will change the DAG of the flink job, thus breaking checkpoint compatibility.
+2. It may infer a small parallelism of reader if you enable Flink AdaptiveBatchScheduler. You can set Flink CoreOptions.DEFAULT_PARALLELISM to avoid.
+3. It will change the failover strategy from the original possible region failover to global failover. Because a Monitor subtask that connects to all downstream ReadOperators.
+So it's default value is false and you have to consider the side effects before enable it.
Review Comment:
Very thanks. done
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2136855051
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java:
##########
@@ -294,6 +295,9 @@ public DataStream<RowData> build() {
}
if (sourceBounded) {
+ if (conf.get(CoreOptions.SCAN_BATCH_PLAN_MOVE_FROM_JM_TO_TM_ENABLED)) {
Review Comment:
Thanks very much, you are right.
1、Add some description in the option's description.
2、Add parameter check in TableScanUtils#streamingReadingValidate.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "yunfengzhou-hub (via GitHub)" <gi...@apache.org>.
yunfengzhou-hub commented on code in PR #5715:
URL: https://github.com/apache/paimon/pull/5715#discussion_r2139355173
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java:
##########
@@ -130,6 +137,14 @@ public void processElement(StreamRecord<Split> record) throws Exception {
reuseRecord.replace(nestedProjectedRowData);
}
output.collect(reuseRecord);
+
+ if (limit != null && numRecordsIn.getCount() == limit) {
+ LOG.info(
+ "When limit is not null, reader {} reach the limit record {}.",
+ this,
+ limit);
+ return;
Review Comment:
In this example, the limit works as expected because there is only one source split, and only one subtask among the ReadOperators is working. In cases where the amount of data grows bigger, the actual records sent in this way would be `limit * numOfSourceSplits`. This is unexpected.
Besides, I tried running testBatchReadSourceWithSnapshot after removing this logic from ReadOperator, and the `SELECT ... LIMIT 2` syntax still functioned correctly. This demonstrates that there is another place for limit to work, which is at FlinkSourceBuilder#createReadBuilder.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink] Add batch read operator source. [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #5715:
URL: https://github.com/apache/paimon/pull/5715
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org